【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-plugins
https://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.conflogstash利用 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