ELK logstash 处理MySQL慢查询日志(26th)

A+

所属分类：ELK

在生产环境下，logstash 经常会遇到处理多种格式的日志，不同的日志格式，解析方法不同。下面来说说logstash处理多行日志的例子，对MySQL慢查询日志进行分析，这个经常遇到过，网络上疑问也很多。

MySQL慢查询日志格式如下：

# User@Host: ttlsa[ttlsa] @  [10.4.10.12]  Id: 69641319
# Query_time: 0.000148  Lock_time: 0.000023 Rows_sent: 0  Rows_examined: 202
SET timestamp=1456717595;
select  `Id`, `Url` from `File` where `Id` in ('201319', '201300');
# Time: 160229 11:46:37

# User@Host: ttlsa[ttlsa] @ [10.4.10.12] Id: 69641319

# Query_time: 0.000148 Lock_time: 0.000023 Rows_sent: 0 Rows_examined: 202

SET timestamp=1456717595;

select `Id`, `Url` from `File` where `Id` in ('201319', '201300');

# Time: 160229 11:46:37

filebeat配置

我这里是使用filebeat 1.1.1版本的，之前版本没有multiline配置项，具体方法看后面那种。

filebeat:
  prospectors:
    -
      paths:
        - /www.ttlsa.com/logs/mysql/slow.log
      document_type: mysqlslowlog
      input_type: log
      multiline:
        negate: true
        match: after
  registry_file: /var/lib/filebeat/registry
output:
  logstash:
    hosts: ["10.6.66.14:5046"]
shipper:
logging:
  files:

filebeat:

prospectors:

paths:

- /www.ttlsa.com/logs/mysql/slow.log

document_type: mysqlslowlog

input_type: log

multiline:

negate: true

match: after

registry_file: /var/lib/filebeat/registry

output:

logstash:

hosts: ["10.6.66.14:5046"]

shipper:

logging:

files:

logstash配置

1. input段配置

# vi /etc/logstash/conf.d/01-beats-input.conf 
input {
  beats {
    port => 5046
    host => "10.6.66.14"
 }
}

# vi /etc/logstash/conf.d/01-beats-input.conf

input {

beats {

port => 5046

host => "10.6.66.14"

}

2. filter 段配置

# vi /etc/logstash/conf.d/16-mysqlslowlog.log 
filter {
  if [type] == "mysqlslowlog" {
  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", "UNIX", "YYYY-MM-dd HH:mm:ss"]
      remove_field => [ "timestamp" ]
    }
  }
}

# vi /etc/logstash/conf.d/16-mysqlslowlog.log

filter {

if [type] == "mysqlslowlog" {

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", "UNIX", "YYYY-MM-dd HH:mm:ss"]

remove_field => [ "timestamp" ]

}

关键之重是grok正则的配置。

3. output段配置

# vi /etc/logstash/conf.d/30-beats-output.conf 
output {
    if "_grokparsefailure" in [tags] {
      file { path => "/var/log/logstash/grokparsefailure-%{[type]}-%{+YYYY.MM.dd}.log" }
    }

  if [@metadata][type] in [ "mysqlslowlog" ] {
    elasticsearch {
      hosts => ["10.6.66.14:9200"]
      sniffing => true
      manage_template => false
      template_overwrite => true
      index => "%{[@metadata][beat]}-%{[type]}-%{+YYYY.MM.dd}"
      document_type => "%{[@metadata][type]}"
    }
  }
}

# vi /etc/logstash/conf.d/30-beats-output.conf

output {

if "_grokparsefailure" in [tags] {

file { path => "/var/log/logstash/grokparsefailure-%{[type]}-%{+YYYY.MM.dd}.log" }

}

if [@metadata][type] in [ "mysqlslowlog" ] {

elasticsearch {

hosts => ["10.6.66.14:9200"]

sniffing => true

manage_template => false

template_overwrite => true

index => "%{[@metadata][beat]}-%{[type]}-%{+YYYY.MM.dd}"

document_type => "%{[@metadata][type]}"

}

标准输出结果截图

Elasticsearch

elasticsearch结果截图

Elasticsearch

如果是使用filebeat1.1.1之前的版本，配置如下：

1. filebeat配置

filebeat:
  prospectors:
    -
      paths:
        - /www.ttlsa.com/logs/mysql/slow.log
      document_type: mysqlslowlog
      input_type: log
  registry_file: /var/lib/filebeat/registry
output:
  logstash:
    hosts: ["10.6.66.14:5046"]
shipper:
logging:
  files:

filebeat:

prospectors:

paths:

- /www.ttlsa.com/logs/mysql/slow.log

document_type: mysqlslowlog

input_type: log

registry_file: /var/lib/filebeat/registry

output:

logstash:

hosts: ["10.6.66.14:5046"]

shipper:

logging:

files:

2. logstash input段配置

input {
  beats {
    port => 5046
    host => "10.6.66.14"
    codec => multiline {
      pattern => "^# User@Host:"
      negate => true
      what => previous
    }
 }
}

input {

beats {

port => 5046

host => "10.6.66.14"

codec => multiline {

pattern => "^# User@Host:"

negate => true

what => previous

}

其它配置不变。

如有疑问跟帖，或者加群沟通。

微信公众号

扫一扫关注运维生存时间公众号，获取最新技术文章~

评论已关闭！

目前评论：5 其中：访客 5 博主 0

bin 1
18/10/2016 上午 10:11 沙发

您好，请问这个如果我收集日志的就用logstash来收集慢日志，过滤就用上文中的可以吗
taotao 1
20/05/2016 下午 1:33 板凳

创建MySQL慢日志对应的索引详细信息能发下看看吗
taotao 1
20/05/2016 下午 1:18 地板

es 中创建慢日志对应的索引信息能发下看看吗
luopeijue 0
11/03/2016 上午 10:44 4楼

如果有多个地方发慢日志过来，如果保证合并的是有序且同设备同一个信息的正确性呢，input这个插件是单线程安全的吗？
- 默北 6
  11/03/2016 上午 11:18 1层
  
  @luopeijue 时间是有序的。不管啥时候读取到慢日志，日志里面的时间是不会改变的。
  安全的，即时挂了也会知道从何处读取。