logstash解析mysql慢日志

Posted 2021-01-11

在工作中需要在elk中展示mysql的慢语句，以便于DBA每天查看并对比进行优化；
mysql5.5，mysql5.6，mysql5.7的慢日志格式都不相同，根据自已的需要进行收集；
mysql5.5日志样例：

# Time: 180911 10:50:31
# [email protected]: osdb[osdb] @  [172.25.14.78]
# Query_time: 12.597483  Lock_time: 0.000137 Rows_sent: 451  Rows_examined: 2637425
SET timestamp=1536634231;
SELECT id,name,contenet from cs_tables;

mysql5.6日志样例：

# Time: 180911 11:36:20
# [email protected]: root[root] @ localhost []  Id:  1688
# Query_time: 3.006539  Lock_time: 0.000000 Rows_sent: 1  Rows_examined: 0
SET timestamp=1536550580;
SELECT id,name,contenet from cs_tables;

mysql5.7日志样例：

# Time: 2018-09-10T06:26:40.895801Z
# [email protected]: root[root] @  [172.16.213.120]  Id:   208
# Query_time: 3.032884  Lock_time: 0.000139 Rows_sent: 46389  Rows_examined: 46389
use cmsdb;
SET timestamp=1536560800;
select * from cstable;

通过分析上面三个mysql版本的慢查询日志，得出如下结论：

（1）每个Mysql版本的慢查询日志中Time字段格式都不一样 
（2）在mysql5.6、5.7版本中有一个id字段，而在mysql5.5版本中是没有Id字段的。 
（3）每个慢查询语句是分多行完成的，并且每行中有多少不等的空格、回车等字符。
（4）use db语句可能出现在慢查询中，也可以不出现。
（5）每个慢查询语句的最后一部分是具体执行的sql，这个sql可能跨多行，也可能是多条sql语句。

filebeat先读取mysql的慢日志，写入redis中：

filebeat.inputs:
- type: log
  paths:
  - /data/mysqldata/mysql-slow.log
  tags: ["oms-slow-logs"]
  exclude_lines: [‘^# Time‘]
  fields:
    type: "oms-slow-logs"
  fields_under_root: true
  multiline:
    pattern: ‘^# Time|^# User‘
    negate: true
    match: after
processors:
  - drop_fields:
      fields: ["source","input","beat","prospector","offset"]
name: 10.10.7.32
output.redis:
  hosts: ["10.78.1.180"]
  key: "oms-slow-logs"
  type: list

logstash向redis读取数据，解析过滤之后写入elastic中：

input {
    redis {
    host => "10.78.1.180"
        port => 6379
        data_type => list
        key => "oms-slow-logs"
    }
}
filter {
    grok {
       # 有ID有use
        match => [ "message", "^#[email protected]:s+%{USER:user}[[^]]+][email protected]s+(?:(?<clienthost>S*) )?[(?:%{IP:clientip})?]s+Id:s+%{NUMBER:id}
# Query_time: %{NUMBER:query_time}s+Lock_time: %{NUMBER:lock_time}s+Rows_sent: %{NUMBER:rows_sent}s+Rows_examined: %{NUMBER:rows_examined}
uses(?<dbname>w+);
SETs+timestamp=%{NUMBER:timestamp_mysql};
(?<query>[sS]*)" ]

        # 有ID无use
        match => [ "message", "^#[email protected]:s+%{USER:user}[[^]]+][email protected]s+(?:(?<clienthost>S*) )?[(?:%{IP:clientip})?]s+Id:s+%{NUMBER:id}
# Query_time: %{NUMBER:query_time}s+Lock_time: %{NUMBER:lock_time}s+Rows_sent: %{NUMBER:rows_sent}s+Rows_examined: %{NUMBER:rows_examined}
SETs+timestamp=%{NUMBER:timestamp_mysql};
(?<query>[sS]*)" ]

        # 无ID有use
        match => [ "message", "^#[email protected]:s+%{USER:user}[[^]]+][email protected]s+(?:(?<clienthost>S*) )?[(?:%{IP:clientip})?]
# Query_time: %{NUMBER:query_time}s+Lock_time: %{NUMBER:lock_time}s+Rows_sent: %{NUMBER:rows_sent}s+Rows_examined: %{NUMBER:rows_examined}
uses(?<dbname>w+);
SETs+timestamp=%{NUMBER:timestamp_mysql};
(?<query>[sS]*)" ]

        # 无ID无use
        match => [ "message", "^#[email protected]:s+%{USER:user}[[^]]+][email protected]s+(?:(?<clienthost>S*) )?[(?:%{IP:clientip})?]
# Query_time: %{NUMBER:query_time}s+Lock_time: %{NUMBER:lock_time}s+Rows_sent: %{NUMBER:rows_sent}s+Rows_examined: %{NUMBER:rows_examined}
SETs+timestamp=%{NUMBER:timestamp_mysql};
(?<query>[sS]*)" ]
    }
    date {
            match => ["timestamp_mysql","UNIX"]
            target => "@timestamp"
    }
    mutate {
            remove_field => ["@version","message","timestamp_mysql"]
    }
}
output {
    if [type] == "oms-slow-logs" {
        if [tags][0] == "oms-slow-logs" {
        elasticsearch {
                hosts  => ["10.10.5.78:9200","10.10.5.79:9200","10.10.5.80:9200"]
                index  => "%{type}-%{+YYYY.MM.dd}"
            }
        }
    }
}

kibana展示：