Logstash分析MySQL慢查询日志

最近在使用ELKStack对系统日志进行分析，在网上也有看到有使用logstash的案例，但是发现不能正常解析出来，于是重新花时间去进行正则计算，主要代码如下:

input {
  file {
    type => "mysql-slow"
    path => "/var/lib/mysql/slow.log"
        start_position => beginning
                sincedb_write_interval => 0
    codec => multiline {
      pattern => "^# [email protected]:"
      negate => true
      what => "previous"
    }
  }
}
filter {

if [message] =~ "^Tcp" {
        drop {}
}

if [message] =~ "^Time" {
        drop {}
}

if [message] =~ "^\/usr" {
        drop {}
}
  grok {
    match => { "message" => "SELECT SLEEP" }
    add_tag => [ "sleep_drop" ]
    tag_on_failure => []
  }

  if "sleep_drop" in [tags] {
    drop {}
  }
    grok {
    match => { "message" => "(?m)^# [email protected]: %{USER:User}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:Client_IP})?\]\s.*# Query_time: %{NUMBER:Query_Time:float}\s+Lock_time: %{NUMBER:Lock_Time:float}\s+Rows_sent: %{NUMBER:Rows_Sent:int}\s+Rows_examined: %{NUMBER:Rows_Examined:int}\s*(?:use %{DATA:Database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<Query>(?<Action>\w+)\s+.*)\n# Time:.*$" }
    }

  date {
    match => [ "timestamp", "UNIX" ]
    remove_field => [ "timestamp" ]
  }

}
output {
            redis {
                host => "192.168.1.2:6379"
                data_type => "list"
                key => "logstash:mysql_slow_log"
            }

}

说明:

在使用codec/multiline搭配使用的时候，需要注意，grok和普通正则一样默认是不支持匹配回车换行的。就像你需要=～//m一样也需要单独指定，具体写法是在表达式开始位置加(?m)标记

开始的时候一直没加(?m)，在http://grokdebug.herokuapp.com/调试的时候正常通过，但是到了logstash执行的时候就是不能正常解析。