【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