当 SQL 语句执行时间超过所设定的阈值时，便会记录到指定的日志文件中，所记录内容称之为慢查询日志

数据库在运行期间，可能会存在 SQL 语句查询过慢，那我们如何快速定位、分析哪些 SQL语句需要优化处理，又是哪些 SQL 语句给业务系统造成影响呢？当我们进行统一的收集分析，SQL语句执行的时间，以及执行的SQL语句，一目了然。

1.安装 MySQL；开启 MySQL 慢查询日志；
2 配置 filebeat 收集本地慢查询日志路径；
2.1 使用 exclude_lines 排除无用行；
2.2 使用 multiline 对内容进行合并；
3 配置 logstash
3.1.使用 grok 插件将 mysql 慢日志格式化为 json格式；
3.2.使用 date 插件将 timestamp 时间转换为本地时间，然后覆盖 @timestamp；
3.3.检查 json 格式是否成功，成功后可以将没用的字段删除；
3.4.最后将输出到屏幕的内容，输出至Elasticsearch集群；

vim /etc/my.cnf
[mysqld]
...
slow_query_log=ON
slow_query_log_file=/var/log/mariadb/slow.log # 需要mysql用户有权限写
long_query_time=3
...

重启mysql ,并模拟产生慢日志
MariaDB [(none)]> select sleep(1) user,host from mysql.user;
MariaDB [(none)]> select sleep(0.5) user,host from mysql.user;
MariaDB [(none)]> select sleep(0.6) user,host from mysql.user;

慢日志格式
[root@sonarqube mysqldb]# vim slow.log
# Time: 211031 17:22:29
# User@Host: root[root] @ localhost []  Id:     5
# Query_time: 7.013641  Lock_time: 0.000081 Rows_sent: 7  Rows_examined: 7
SET timestamp=1635672149;
select sleep (1) user,host from mysql.user;
# Time: 211031 17:22:51
# User@Host: root[root] @ localhost []  Id:     5
# Query_time: 3.518726  Lock_time: 0.000080 Rows_sent: 7  Rows_examined: 7
SET timestamp=1635672171;
select sleep (0.5) user,host from mysql.user;
# Time: 211031 17:23:00
# User@Host: root[root] @ localhost []  Id:     5
# Query_time: 4.913531  Lock_time: 0.000074 Rows_sent: 7  Rows_examined: 7
SET timestamp=1635672180;
select sleep (0.7) user,host from mysql.user;
~

cat /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths: /var/log/mariadb/slow.log
  exclude_lines: ['^\# Time'] # 排除匹配的行
  multiline.pattern: '^\# User'
  multiline.negate: true
  multiline.match: after
  multiline.max_lines: 1000

output.logstash:
  hosts: ["172.16.1.151:5044"]

cat mysql_beat_logstash_es.conf 
input {

	beats {
		port => 5044
	}
}

filter {
	mutate {
		gsub => [ "message", "\n", " "] # 去除\n

	}
   	grok {
        	match => { "message" => "(?m)^# User@Host: %{USER:User}\[%{USER-2:User}\] @ (?:(?<Clienthost>\S*) )?\[(?:%{IP:Client_IP})?\] # Thread_id: %{NUMBER:Thread_id:integer}\s+ Schema: (?:(?<DBname>\S*) )\s+QC_hit: (?:(?<QC_hit>\S*) )# Query_time: %{NUMBER:Query_Time}\s+ Lock_time: %{NUMBER:Lock_Time}\s+ Rows_sent: %{NUMBER:Rows_Sent:integer}\s+Rows_examined: %{NUMBER:Rows_Examined:integer} SET timestamp=%{NUMBER:timestamp}; \s*(?<Query>(?<Action>\w+)\s+.*)" }
    }
	date {

		match => ["timestamp","UNIX","YYYY-MM-dd HH:mm:ss"] # unix时间转换
		target => "@timestamp"
		timezone => "Asia/Shanghai"
	}
	mutate {
        #移除message等字段
        remove_field => ["message","input","timestamp"]

        #对Query_time Lock_time 格式转换为浮点数
        convert => ["Lock_Time","float"]
        convert => ["Query_Time","float"]
        
        #添加索引名称，官方建议这种写法，也可以直接写死
      add_field => { "[@metadata][target_index]" => "logstash-mysql-%{+YYYY.MM.dd}" } 
    }

}
output {

	stdout {
		codec => rubydebug
	}
	elasticsearch {
        hosts => ["172.16.1.162:9200","172.16.1.163:9200"]
        index => "%{[@metadata][target_index]}" # 直接调用索引
        template_overwrite => true
    }
}