ELK架构监控MySQL慢日志

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/local/mysql-8.2.0/mysql_slow.log
  scan_frequency: 10s
  multiline.pattern: '^\s*# Time:'
  multiline.negate: true
  multiline.match: after 
 
 
output.redis:
  enabled: true
  hosts: ["192.168.122.227:6379","192.168.122.237:6379","192.168.122.238:6379"]
  key: "uap-mysql-slow-log"
  datatype: list
  password: "Redis@123456" 
  db: 0
  codec: [ json ]
  loadbalance: true
 
logging.level: info
logging.to_files: true
logging.files:
  path: /opt/module/filebeat-8.11.0
  name: filebeat.log

# 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的值赋值给@timestamp
     date {
         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 => true
       truststore => "/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 => true
       truststore => "/opt/module/logstash-8.11.0/config/certs/http.p12"
       truststore_password => "123456"
     }
   }
 }
}