logstash安裝和使用

官網：https://www.elastic.co/cn/logstash/

1.上傳Linux安裝包
在這里插入圖片描述
2.解壓安裝包且重命名

[root@VM-4-10-centos logstash]# tar -zxvf logstash-8.11.1-linux-x86_64.tar.gz -C ../software/

[root@VM-4-10-centos logstash]# mv logstash-8.11.1/ logstash

3.啟動測試

運行最基本的 Logstash 管道，控制臺輸入控制臺打印輸出方便測試

[root@VM-4-10-centos logstash]# bin/logstash -e 'input { stdin { } } output { stdout {} }'

在這里插入圖片描述
4.常用數據采集案例

輸入腳本配置

官方輸入插件：https://www.elastic.co/guide/en/logstash/current/input-plugins.html

采集Logstash的心跳輸出到控制臺

heartbeat.conf

input {heartbeat {#采集數據的頻率10s一次interval => 10  type => "heartbeat"}
}output {stdout {codec => rubydebug}
}

[root@VM-4-10-centos logstash]# bin/logstash -f confdata/heartbeat.conf

在這里插入圖片描述

進入到 Logstash 安裝目錄，并修改 config/logstash.yml 文件。我們把config.reload.automatic 設置為 true。
這樣設置的好處是，每當我修改完我的配置文件后，我不需要每次都退出我的Logstash，然后再重新運行。Logstash 會自動偵測到最新的配置文件的變化。

監控端口數據變化輸出到控制臺

創建一個叫做 weblog.conf 的配置文件，并輸入一下的內容：

input {tcp {port => 8848}
}output {stdout { }
}

[root@VM-4-10-centos logstash]# bin/logstash -f confdata/heartbeat.conf

[root@VM-4-10-centos ~]# echo 'hello logstash' | nc localhost 8848

在這里插入圖片描述

定期運行 shell 命令，并將shell命令返回的結果整個輸出

input {exec {command => "echo 'hi!'"#每30秒執行一次interval => 30}
}
output {stdout { }
}

在這里插入圖片描述

Input插件監控日志

input {file {#檢測文件路徑path => "/opt/software/logstash/logdata/test.log"#檢測頻率stat_interval => 30#從開頭進行檢測start_position => "beginning" }
}
output{stdout{}
}

在這里插入圖片描述

生成隨機日志事件

這樣做的一般目的是測試插件的性能

input {generator  {count => 3lines => ["java","python","helloworld"]ecs_compatibility => disabled}
}
output {stdout { }
}

在這里插入圖片描述

調用 HTTP API接口采集數據

input {http_poller {urls => {url => "http://api.openweathermap.org/data/2.5/weather?q=London,uk&APPID=7dbe7341764f682c2242e744c4f167b0&units=metric"}request_timeout => 60schedule => { every => "5s"}codec => "json"metadata_target => "http_poller_metadata"}
}
output {stdout {}
}

在這里插入圖片描述

JDBC 輸入插件

首先，我們將適當的 JDBC 驅動程序庫放在我們當前的

input {jdbc {#驅動包的位置jdbc_driver_library => "/opt/software/logstash/lib/mysql-connector-java-8.0.27.jar"jdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_connection_string => "jdbc:mysql://10.0.4.10:3306/metastore"jdbc_user => "root"jdbc_password => "123456"schedule => "* * * * *"statement => "select * from DBS;"}
}
output {stdout {}
}

在這里插入圖片描述

過濾插件配置

官網過濾插件：https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

刪除過濾器插件（刪除進入此篩選器的所有內容）

在這里插入圖片描述

input {jdbc {#驅動包的位置jdbc_driver_library => "/opt/software/logstash/lib/mysql-connector-java-8.0.27.jar"jdbc_driver_class => "com.mysql.cj.jdbc.Driver"jdbc_connection_string => "jdbc:mysql://10.0.4.10:3306/metastore"jdbc_user => "root"jdbc_password => "123456"schedule => "* * * * *"statement => "select * from DBS;"}
}
filter {if [name] == "zhangtest" {drop { }}}
output {stdout {}
}

添加過濾插件后name=zhangtest的那條數據就沒有采集
在這里插入圖片描述

Grok filter plugin

Grok 是將非結構化日志數據解析為結構化和可查詢內容的好方法

input {generator {message => "2019-09-09T13:00:00Z Whose woods these are I think I know."count => 1}
}filter {grok {match => ["message", "%{TIMESTAMP_ISO8601:timestamp_string}%{SPACE}%{GREEDYDATA:line}"]}
}output {stdout { codec => rubydebug }
}

在這里插入圖片描述

Dissect filter

input {generator {message => "<1>Oct 16 20:21:22 www1 1,2016/10/16 20:21:20,3,THREAT,SCAN,6,2016/10/16 20:21:20,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47,48,49,50,51,52,53,54"count => 1}
}filter {if [message] =~ "THREAT," {dissect {mapping => {message => "<%{priority}>%{syslog_timestamp} %{+syslog_timestamp} %{+syslog_timestamp} %{logsource} %{pan_fut_use_01},%{pan_rec_time},%{pan_serial_number},%{pan_type},%{pan_subtype},%{pan_fut_use_02},%{pan_gen_time},%{pan_src_ip},%{pan_dst_ip},%{pan_nat_src_ip},%{pan_nat_dst_ip},%{pan_rule_name},%{pan_src_user},%{pan_dst_user},%{pan_app},%{pan_vsys},%{pan_src_zone},%{pan_dst_zone},%{pan_ingress_intf},%{pan_egress_intf},%{pan_log_fwd_profile},%{pan_fut_use_03},%{pan_session_id},%{pan_repeat_cnt},%{pan_src_port},%{pan_dst_port},%{pan_nat_src_port},%{pan_nat_dst_port},%{pan_flags},%{pan_prot},%{pan_action},%{pan_misc},%{pan_threat_id},%{pan_cat},%{pan_severity},%{pan_direction},%{pan_seq_number},%{pan_action_flags},%{pan_src_location},%{pan_dst_location},%{pan_content_type},%{pan_pcap_id},%{pan_filedigest},%{pan_cloud},%{pan_user_agent},%{pan_file_type},%{pan_xff},%{pan_referer},%{pan_sender},%{pan_subject},%{pan_recipient},%{pan_report_id},%{pan_anymore}"}}}
}output {stdout { codec => rubydebug }
}

在這里插入圖片描述

KV filter

解析鍵/值對中數據的簡便方法

input {generator {message => "pin=12345~0&d=123&e=foo@bar.com&oq=bobo&ss=12345"count => 1}
}filter {kv {source => "message"target => "parsed"field_split => "&?"}
}output {stdout { codec => rubydebug }
}

在這里插入圖片描述

JSON filter

input {generator {message => '{"id":2,"timestamp":"2019-08-11T17:55:56Z","paymentType":"Visa","name":"Darby Dacks","gender":"Female","ip_address":"77.72.239.47","purpose":"Shoes","country":"Poland","age":55}'count => 1}
}filter {json {source => "message"}
}output {stdout {codec => rubydebug}
}

在這里插入圖片描述

輸出插件配置

官網輸出插件：https://www.elastic.co/guide/en/logstash/current/output-plugins.html

Elasticsearch 輸出插件

input {generator {message => '{"id":2,"timestamp":"2019-08-11T17:55:56Z","paymentType":"Visa","name":"Darby Dacks","gender":"Female","ip_address":"77.72.239.47","purpose":"Shoes","country":"Poland","age":55}'count => 1}
}filter {json {source => "message"}
}output {stdout {codec => rubydebug}elasticsearch {hosts => ["10.0.4.10:9200"]index => "flinkdata"workers => 1}
}