elk收集mysql慢查询日志

mysql:5.7.28

filebeat:7.6.2

elk:7.6.2

软件安装和mysql慢查询开启配置就略过了，直接上配置

不同版本mysql的慢查询日志内容有所差异，我这个版本的日志大概类似这样

# Time: 2023-08-01T17:46:17.511148+08:00
# User@Host: root[root] @ localhost []  Id:    14
# Query_time: 1.968041  Lock_time: 0.000301 Rows_sent: 360959  Rows_examined: 360959
SET timestamp=1690883177;
select * from info_202305 where id>1000000 and id <1400000;

filebeat.yml配置

filebeat.inputs:
- input_type: log
  enabled: true
  paths:
    - /data/mysql/data/mysql_slow.log   #mysql慢查询日志文件
  exclude_lines: ['^\# Time|^/data/mysql/bin/mysqld|^Tcp port|^Time'] #排除行
  multiline.negate: true        #多行合并
  multiline.match: after
  multiline.pattern: '^\# User|^\# Time'
  tags: ["mysql-slow-log"]    #打一个tags
 
output.logstash:    #输出至logstash
  hosts: ["172.2.2.10:55044"]
 
processors:
  - add_host_metadata: ~
  - add_cloud_metadata: ~
  - drop_fields:      #删除无用的字段   
      fields: ["beat", "input", "source", "offset", "prospector"]

logstash配置

input：

[root@elk conf.d]# cat input-mysql-slow.conf 
#mysql-slow-log
input {
  beats {
    port => 55044
  }
}
[root@elk conf.d]#

output：

filter {
    json {
        source => "message"
    }
}
output {
      elasticsearch {
              hosts => ["172.2.2.10:9200"]
              index => "mysql-slowlog-%{+YYYY-MM-dd}"
              user => "elastic"
              password => "123@abc"
      }
}

这样输出的日志信息是在message字段中，如果想过滤可以将这个字段删除，这样展示更清晰，下面是加了过滤后的output配置

filter {
       json {
              source => "message"                                                                              
       }
       if "mysql-slow-log" in [tags] {
              json {
                     source => "message"
              }
              grok {
                     match => [ "message" , "(?m)^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IPV4:clientip})?\]\s+Id:\s+%{NUMBER:row_id:int}\n#\s+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}\n\s*(?:use %{DATA:database};\s*\n)?SET\s+timestamp=%{NUMBER:timestamp};\n\s*(?<sql>(?<action>\w+)\b.*;)\s*(?:\n#\s+Time)?.*$" ]
              }
​
              date {
                     match => ["timestamp_mysql","UNIX"]   #这里我们需要对日志输出进行时间格式转换，默认将timestamp_mysql的unix时间格式之后直接赋值给timestamp
                     target => "@timestamp"
              }
              mutate {
                     remove_field => "@version"   #版本字段，删除不需要的字段
                     remove_field => "message"    #在上述，我们已经对mysql的慢日志输出分段了，所以message输出已经是多余的了，故此删除
              }
       }
}
​
output {
       elasticsearch {
              hosts => ["172.2.2.10:9200"]
              index => "mysql-slowlog-%{+YYYY-MM-dd}"
              user => "elastic"
              password => "123@abc"
       }
}