logstash收集MySQL慢查询日志

#此处以收集mysql慢查询日志为准，根据文件名不同添加不同的字段值
input {
    file {
        path => "/data/order-slave-slow.log"
        type => "mysql-slow-log"
        start_position => "beginning"
        codec => multiline {
            pattern => "^# User@Host:"
            negate => true
            what => previous
        }
    }
    file {
        path => "/data/other-slave-slow.log"
        type => "mysql-slow-log"
        start_position => "beginning"
        codec => multiline {
            pattern => "^# User@Host:"
            negate => true
            what => previous
        }
    }
    file {
        path => "/data/order-master-slow.log"
        type => "mysql-slow-log"
        start_position => "beginning"
        codec => multiline {
            pattern => "^# User@Host:"
            negate => true
            what => previous
        }
    }
    file {
        path => "/data/other-master-slow.log"
        type => "mysql-slow-log"
        start_position => "beginning"
        codec => multiline {
            pattern => "^# User@Host:"
            negate => true
            what => previous
        }
    }
}
filter {
    if [path] =~ "order-slave-slow" { //根据文件内容不同，增加不同的字段
        grok {
            match => { "message" => "(?m)^#s+User@Host:s+%{USER:user}[[^]]+]s+@s+(?:(?<clientip>S*) )?[(?:%{IPV4:clientip})?]s+Id:s+%{NUMBER:row_id:int}
#s+Query_time:s+%{NUMBER:Query_time:float}s+Lock_time:s+%{NUMBER:lock_time:float}s+Rows_sent:s+%{NUMBER:Row_sent:int}s+Rows_examined:s+%{NUMBER:Rows_examined:int}
s*(?:use %{DATA:database};s*
)?SETs+timestamp=%{NUMBER:timestamp};
s*(?<sql>(?<action>w+).*;)s*(?:
#s+Time)?.*$" }
            remove_field => [ "message" ]
        }
        mutate {
            replace => [ "host" , "%{host}" ]
            add_field => [ "nsCode", "%{nsCode}" ]
            add_field => [ "envCode", "%{envCode}" ]
            add_field => [ "mysqlType", "%{mysqlType}" ]
            gsub => [ "sql", "
# Time: d+s+d+:d+:d+", "" ]
        }
    }
    if [path] =~ "other-slave-slow" {
        grok {
            match => { "message" => "(?m)^#s+User@Host:s+%{USER:user}[[^]]+]s+@s+(?:(?<clientip>S*) )?[(?:%{IPV4:clientip})?]s+Id:s+%{NUMBER:row_id:int}
#s+Query_time:s+%{NUMBER:Query_time:float}s+Lock_time:s+%{NUMBER:lock_time:float}s+Rows_sent:s+%{NUMBER:Row_sent:int}s+Rows_examined:s+%{NUMBER:Rows_examined:int}
s*(?:use %{DATA:database};s*
)?SETs+timestamp=%{NUMBER:timestamp};
s*(?<sql>(?<action>w+).*;)s*(?:
#s+Time)?.*$" }
            remove_field => [ "message" ]
        }
        mutate {
            replace => [ "host" , "%{host}" ]
            add_field => [ "nsCode", "%{nsCode}" ]
            add_field => [ "envCode", "%{envCode}" ]
            add_field => [ "mysqlType", "%{mysqlType}" ]
            gsub => [ "sql", "
# Time: d+s+d+:d+:d+", "" ]
        }
    }
    if [path] =~ "order-master-slow" {
        grok {
            match => { "message" => "(?m)^#s+User@Host:s+%{USER:user}[[^]]+]s+@s+(?:(?<clientip>S*) )?[(?:%{IPV4:clientip})?]s+Id:s+%{NUMBER:row_id:int}
#s+Query_time:s+%{NUMBER:Query_time:float}s+Lock_time:s+%{NUMBER:lock_time:float}s+Rows_sent:s+%{NUMBER:Row_sent:int}s+Rows_examined:s+%{NUMBER:Rows_examined:int}
s*(?:use %{DATA:database};s*
)?SETs+timestamp=%{NUMBER:timestamp};
s*(?<sql>(?<action>w+).*;)s*(?:
#s+Time)?.*$" }
            remove_field => [ "message" ]
        }
        mutate {
            replace => [ "host" , "%{host}" ]
            add_field => [ "nsCode", "%{nsCode}" ]
            add_field => [ "envCode", "%{envCode}" ]
            add_field => [ "mysqlType", "%{mysqlType}" ]
            gsub => [ "sql", "
# Time: d+s+d+:d+:d+", "" ]
        }
    }
    if [path] =~ "other-master-slow" {
        grok {
            match => { "message" => "(?m)^#s+User@Host:s+%{USER:user}[[^]]+]s+@s+(?:(?<clientip>S*) )?[(?:%{IPV4:clientip})?]s+Id:s+%{NUMBER:row_id:int}
#s+Query_time:s+%{NUMBER:Query_time:float}s+Lock_time:s+%{NUMBER:lock_time:float}s+Rows_sent:s+%{NUMBER:Row_sent:int}s+Rows_examined:s+%{NUMBER:Rows_examined:int}
s*(?:use %{DATA:database};s*
)?SETs+timestamp=%{NUMBER:timestamp};
s*(?<sql>(?<action>w+).*;)s*(?:
#s+Time)?.*$" }
            remove_field => [ "message" ]
        }
        mutate {
　　　　　　　#替换原有host字段的值
            replace => [ "host" , "%{host}" ]
　　　　　　  #新增三个字段
            add_field => [ "nsCode", "%{nsCode}" ]
            add_field => [ "envCode", "%{envCode}" ]
            add_field => [ "mysqlType", "%{mysqlType}" ]
　　　　　　　//sql字段的值进行切分，"
# Time: d+s+d+:d+:d+"匹配到的内容替换为空。
　　　　　　　gsub => [ "sql", "
# Time: d+s+d+:d+:d+", "" ] 
　　　　} 
　　} 
}

//此处输出至redis服务器中
output {
    if [type] == "mysql-slow-log" {
        redis {
            host => "%{ES_SEVER}" //此处指向redis服务器地址
            data_type => "list"
            key => "mysql-slow-log"
        }
    }
}