【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和endstat_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 =>6666codec => 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: logenabled: 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