目錄
一、架構概述
二、安裝部署
三、Filebeat配置
四、Logstash配置
一、架構概述
? ? 本文使用將使用filebeat收集mysql日志信息，發送到redis中緩存，由logstash從redis中取出，發送es中存儲，再從kibana中展示。
二、安裝部署
? ELK各中間件的安裝部署參考章節：審計日志>ELK日志收集，此處不再贅述。
三、Filebeat配置
filebeat.inputs:
- type: logenabled: truepaths:- /usr/local/mysql-8.2.0/mysql_slow.logscan_frequency: 10smultiline.pattern: '^\s*# Time:'multiline.negate: truemultiline.match: after output.redis:enabled: truehosts: ["192.168.122.227:6379","192.168.122.237:6379","192.168.122.238:6379"]key: "uap-mysql-slow-log"datatype: listpassword: "Redis@123456" db: 0codec: [ json ]loadbalance: truelogging.level: info
logging.to_files: true
logging.files:path: /opt/module/filebeat-8.11.0name: filebeat.log四、Logstash配置
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.# 從redis里面拿日志數據
input {redis {batch_count => 1 #返回的事件數量，此屬性僅在list模式下起作用。data_type => "list" #logstash redis插件工作方式key => "ipu-cbs-mysql-slow-log" #監聽的鍵值host => "192.168.122.227" #redis地址port => 6379 #redis端口號password => "Redis@123456" #如果有安全認證，此項為密碼db => 0 #redis數據庫的編號threads => 1 #啟用線程數量tags => ["uap-mysql-slow-log-159"]}redis {batch_count => 1 #返回的事件數量，此屬性僅在list模式下起作用。data_type => "list" #logstash redis插件工作方式key => "ipu-cbs-mysql-slow-log" #監聽的鍵值host => "192.168.122.237" #redis地址port => 6379 #redis端口號password => "Redis@123456" #如果有安全認證，此項為密碼db => 0 #redis數據庫的編號threads => 1 #啟用線程數量tags => ["uap-mysql-slow-log-159"]}redis {batch_count => 1 #返回的事件數量，此屬性僅在list模式下起作用。data_type => "list" #logstash redis插件工作方式key => "ipu-cbs-mysql-slow-log" #監聽的鍵值host => "192.168.122.238" #redis地址port => 6379 #redis端口號password => "Redis@123456" #如果有安全認證，此項為密碼db => 0 #redis數據庫的編號threads => 1 #啟用線程數量tags => ["uap-mysql-slow-log-159"]}}filter {if "uap-mysql-slow-log-159" in[tags] {mutate {gsub => ["message", "# ", "","message", "\n", " ","message", "\s*@\s*", "@"]}grok {match => { "message" => "Time: %{TIMESTAMP_ISO8601:log_time} User@Host: %{DATA:user_host} Id:\s+%{NUMBER:id:int} Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time:\s+%{NUMBER:lock_time:float}\s+Rows_sent:\s+%{NUMBER:rows_sent:int}\s+Rows_examined:\s+%{NUMBER:rows_examined:int} use\s+%{DATA:database};\s+SET\s+timestamp=%{NUMBER:timestamp}; %{GREEDYDATA:sql}" }}if [sql] {grok {match => { "sql" => "\/\* %{GREEDYDATA:comment} \*\/ %{GREEDYDATA:slow_sql}" }}}if ![slow_sql] {mutate {add_field => { "slow_sql" => "%{sql}"}}}# 將logdate的值賦值給@timestampdate {match => [ "log_time", "ISO8601" ]target => "@timestamp"timezone =>"+08:00"}mutate {remove_field => ["timestamp","input","ecs","log","@version","agent","comment","event","log_time","sql"]}}
}output {if "uap-mysql-slow-log-159" in [tags] {if "tm_aseanbank_tst" in [database]{elasticsearch {hosts => ["https://192.168.122.118:9200","https://192.168.122.119:9200","https://192.168.122.120:9200"]index => "ipu-cbs-mysql-slow-log-test"user => "elastic"password => "elastic"ssl_certificate_verification => truetruststore => "/opt/module/logstash-8.11.0/config/certs/http.p12"truststore_password => "123456"}}else if "tm_aseanbank_dev" in [database] {elasticsearch {hosts => ["https://192.168.122.118:9200","https://192.168.122.119:9200","https://192.168.122.120:9200"]index => "ipu-cbs-mysql-slow-log-dev"user => "elastic"password => "elastic"ssl_certificate_verification => truetruststore => "/opt/module/logstash-8.11.0/config/certs/http.p12"truststore_password => "123456"}}}
}注意事項：??上面用的ipu-cbs-mysql-slow-log-dev 和? ipu-cbs-mysql-slow-log-test 兩個索引，如果es中沒有配置索引缺失自動生成，那么需要手動在es中生成這兩個索引，索引的字段沒有要求，缺的字段它在存儲數據時會自行添加。
