【Logstash02】企业级日志分析系统ELK之Logstash 输入 Input 插件

Logstash 使用

Logstash 命令

官方文档

https://www.elastic.co/guide/en/logstash/current/first-event.html
#各种插件
https://www.elastic.co/guide/en/logstash/current/input-plugins.html
https://www.elastic.co/guide/en/logstash/current/filter-plugins.html
https://www.elastic.co/guide/en/logstash/current/output-plugins.html
https://www.elastic.co/guide/en/logstash/7.6/input-plugins.html
https://www.elastic.co/guide/en/logstash/7.6/filter-plugins.html
https://www.elastic.co/guide/en/logstash/7.6/output-plugins.html

范例：查看帮助

[root@logstash ~]#/usr/share/logstash/bin/logstash --help
#常用选项
-e 指定配置内容
-f 指定配置文件，支持绝对路径，如果用相对路径，是相对于/usr/share/logstash/的路径
-t 语法检查
-r 修改配置文件后自动加载生效,注意:有时候修改配置还需要重新启动生效

#服务方式启动,由于默认没有配置文件,所以7.X无法启动，8.X可以启动
[root@logstash ~]#systemctl start logstash

各种插件帮助

Logstash Reference [8.17] | Elastic

范例: 列出所有插件

[root@logstash ~]#/usr/share/logstash/bin/logstash-plugin list

Github logstash插件链接

https://github.com/logstash-pluginshttps://github.com/logstash-plugins

Logstash 输入 Input 插件

官方链接

Input plugins | Logstash Reference [7.6] | Elastic

标准输入

codec 用于输入数据的编解码器，默认值为plain表示单行字符串，若设置为json，表示按照json方式解析

范例: 交互式实现标准输入

#标准输入和输出,codec => rubydebug指输出格式，是默认值，可以省略，也支持设为json，以json格式输出
/usr/share/logstash/bin/logstash  -e 'input { stdin{} } output { stdout{ codec => rubydebug }}'
#后续还可继续输入其它信息，按ctrl+c退出

#指定输入信息为Json格式
[root@logstash ~]#/usr/share/logstash/bin/logstash -e 'input { stdin{ codec => json } } output { stdout{ codec => rubydebug }}'
{"name":"wang","age": "18","gender":"male"}  #输入Json格式信息

#自动解析
{
          "name" => "wang",
         "event" => {
        "original" => "{\"name\":\"wang\",\"age\": \"18\",\"gender\":\"male\"} \n"
    },
    "@timestamp" => 2025-01-03T05:00:30.673936999Z,
           "age" => "18",
          "host" => {
        "hostname" => "logstash"
    },
        "gender" => "male",
      "@version" => "1"
}
#输入非Json格式信息，告警提示无法自动解析，存放message字段
hello,world

[WARN ] 2025-01-03 05:01:04.357 [[main]<stdin] jsonlines - JSON parse error, original data now in message field {:message=>"Unrecognized token 'hello': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')\n at [Source: (String)\"hello,world\"; line: 1, column: 6]", :exception=>LogStash::Json::ParserError, :data=>"hello,world"}
{
         "event" => {
        "original" => "hello,world\n"
    },
       "message" => "hello,world",
    "@timestamp" => 2025-01-03T05:01:04.359617946Z,
          "host" => {
        "hostname" => "logstash"
    },
          "tags" => [
        [0] "_jsonparsefailure"
    ],
      "@version" => "1"
}

范例: 以配置文件实现标准输入

#配置文件
[root@logstash ~]#cat /etc/logstash/conf.d/stdin_to_stdout.conf
input {
    stdin {
        type => "stdin_type"  #自定义事件类型，可用于后续判断    
        tags => "stdin_tag"   #自定义事件tag，可用于后续判断     
        codec => "json"       #指定Json 格式    
    }
}
 output {
    stdout {
        codec => "rubydebug" #输出格式,此为默认值,可省略
    }
}
#语法检查
[root@logstash ~]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/stdin_to_stdout.conf -t
........
Configuration OK
[INFO ] 2025-01-03 05:07:47.505 [LogStash::Runner] runner - Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash

#执行logstash,选项-r表示动态加载配
[root@logstash ~]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/stdin_to_stdout.conf -r

从文件输入

Logstash 会记录每个文件的读取位置,下次自动从此位置继续向后读取

每个文件的读取位置记录在 /var/lib/logstash/plugins/inputs/file/.sincedb_xxxx 或者 /usr/share/logstash/data/plugins/inputs/file/ 对应的文件中

此文件包括文件的 inode号, 大小等信息

修改 Logstash 配置文件

[root@logstash ~]#cat /etc/logstash/conf.d/file_to_stdout.conf
input {
    file {
        path => "/tmp/wang.*"
        type => "wanglog"     #添加自定义的type字段,可以用于条件判断,和filebeat中tag功能相似
        exclude => "*.txt"    #排除不采集数据的文件，使用通配符glob匹配语法 
        start_position => "beginning" #第一次从头开始读取文件,可以取值为:beginning和end
        stat_interval => "3"        #定时检查文件是否更新，默认1s   
        codec => json                #如果文件是Json格式,需要指定此项才能解析,如果不是Json格式而添加此行也不会影响结果
    }
    file {
        path => "/var/log/syslog"
        type => "syslog"
        start_position => "beginning"
        stat_interval => "3"
    }
}
output {
    stdout {
        codec => rubydebug
    }
}

验证日志数据

#语法检查
[root@logstash ~]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file_to_stdout.conf -t
[root@logstash ~]#echo line1 >> /tmp/wang.log
#执行
[root@logstash ~]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file_to_stdout.conf

logstash利用 sincedb 文件记录了logstash收集的记录文件的信息，比如位置，以方便下次接着从此位置继续收集日志

[root@logstash logstash]#cat  /usr/share/logstash/data/plugins/inputs/file/.*
2232798 0 2052 15 1735885320.283595 /var/log/test.log  #记录了收集文件的inode和大小等信息

[root@logstash logstash]#ll -li /var/log/test.log
2232798 -rw-r--r-- 1 root root 15 Jan  3 14:12 /var/log/test.log

从 Http 请求采取数据

[root@logstash ~]# cat /etc/logstash/conf.d/http_to_stdout.conf
input {
    http {
        port =>6666
        codec => json
    }
}
output {
    stdout {
        codec => rubydebug
    }
}
[root@logstash ~]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/http_to_stdout.conf -r

#执行下面访问可以看到上面信息
[root@ubuntu2004 ~]#curl http://logstash.wang.org:6666
ok
[root@ubuntu2004 ~]#curl  -XPOST -d'test log message' http://logstash.wang.org:6666

#提交Json格式数据，可以自动解析
[root@ubuntu2004 ~]#curl  -XPOST -d'{ "name":"wang","age": "18","gender":"male"}' http://logstash.wang.org:6666

从 Filebeat 读取数据

filebeat配置

filebeat.inputs:
- type: log
  enabled: true             #开启日志           
  paths:
  - /var/log/nginx/access_json.log    #指定收集的日志文件  
  json.keys_under_root: true #默认false,只识别为普通文本,会将全部日志数据存储至message字段，改为true则会以Json格式存储
  json.overwrite_keys: true  #设为true,使用json格式日志中自定义的key替代默认的message字段,此项可选
  tags: ["nginx-access"]
output.logstash:
  hosts: ["10.0.0.104:5044"]  #指定Logstash服务器的地址和端口

Logstash配置

[root@logstash ~]#cat /etc/logstash/conf.d/filebeat_to_stdout.conf
input {
    beats {
        port => 5044
    }
}
output {
    stdout {
        codec => rubydebug
    }
}

访问filebeat生成日志

[root@logstash conf.d]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/filebeat_to_stdout.conf -r

{
       "upstreamtime" => "-",
              "agent" => {
        "ephemeral_id" => "b5311807-a0a9-428f-a076-a3c8c5b9db02",
                  "id" => "a3acb99e-b483-4367-a2df-535d8a39a0fa",
                "name" => "kibana",
             "version" => "8.8.2",
                "type" => "filebeat"
    },
                "ecs" => {
        "version" => "8.0.0"
    },
            "tcp_xff" => "-",
            "referer" => "-",
             "domain" => "10.0.0.186",
               "tags" => [
        [0] "nginx-access",
        [1] "beats_input_raw_event"
    ],
          "http_host" => "10.0.0.186",
       "upstreamhost" => "-",
                "xff" => "-",
               "host" => {
        "name" => "kibana"
    },
                "log" => {
        "offset" => 2576,
          "file" => {
            "path" => "/var/log/nginx/access_json.log"
        }
    },
           "clientip" => "10.0.0.181",
    "http_user_agent" => "curl/7.81.0",
       "responsetime" => 0,
             "status" => "404",
              "input" => {
        "type" => "log"
    },
               "size" => 162,
           "@version" => "1",
         "@timestamp" => 2025-01-03T07:13:49.000Z,
                "uri" => "/adada"
}

从 Redis 中读取数据

支持由多个 Logstash 从 Redis 读取日志,提高性能

Logstash 从 Redis 收集完数据后,将删除对应的列表Key

官方链接:

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-redis.html
https://www.elastic.co/guide/en/logstash/7.6/plugins-inputs-redis.html

范例:

[root@logstash ~]#cat /etc/logstash/conf.d/redis_to_stdout.conf
input {
     redis {
         host => 'Redis_IP'
         port => "6379"
         password => "123456"
         db => "0"
         data_type => 'list'
         key => "nginx-accesslog"
     }
}
output {
    stdout {
        codec => rubydebug
    }
}
[root@logstash ~]#/usr/share/logstash/bin/logstash -f  /etc/logstash/conf.d/redis_to_stdout.conf -r

从 Kafka 中读取数据

官方链接:

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html
https://www.elastic.co/guide/en/logstash/7.6/plugins-inputs-kafka.html

范例:

[root@logstash ~]#cat /etc/logstash/conf.d/kakfa_to_stdout.conf
input {
   kafka {
       bootstrap_servers => "10.0.0.201:9092,10.0.0.202:9092,10.0.0.203:9092"
       #group_id => "logstash"
       topics => ["nginx-accesslog","nginx-errorlog"]
       #topics => "nginx-log"
       codec => "json"
       consumer_threads => 8
   }
}
output {
   stdout {
       codec => rubydebug
   }
}
[root@logstash ~]#/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kakfa_to_stdout.conf -r