Logstash 处理多种格式日志

franket

发布于 2022-01-20 03:01:10

1.5K00

代码可运行

文章被收录于专栏：技术杂记技术杂记

运行总次数：0

代码可运行

前言

生产环境下使用 logstash 经常会遇到多种格式的日志，比如 mongodb 的慢日志，nginx或apache的访问日志，系统syslog日志，mysql的慢日志等

不同日志的解析方法不一样，产生的输出也不一样，如果使用同一个 input|filter|output 流必将导致混乱，最常见的问题就是日志无法获得正确解析，message中的内容还是一整条，并没有从中捕获出我们关心的域值，依旧是schemaless的状态，同时tags中会产生 _grokparsefailure 的标记，然后多种日志都存到了同一个index中，混乱不以，给后期的日志分析也带来不便

logstash提供了一种判断机制来对不同内容进行判断，然后分别处理

这里简单分享一下 logstash 中同时处理 mysql慢日志 和 nginx访问日志 的配置过程，相关的详细内容可以参考 Event Dependent Configuration 和 Logstash Configuration Examples

Tip: 当前的最新版本为 Logstash 2.1.1 、filebeat-1.0.1

概要

软件版本

percona server 5.6.27-75.0 #是这个版本产生的慢日志，其它版本使用下面的正则，不一定能正常解析
elasticsearch 2.1.1
logstash 2.1.1
filebeat-1.0.1

logstash配置

[root@logstash-server conf.d]# cat filebeat-logstash-es.conf 
input {
	
	beats{
		port => 5077
		tags => ["nginx_access_log"]
	}
	beats{
                port => 5088
                codec => multiline {
                   pattern => "^# User@Host:"
                   negate => true
                   what => previous
                }
		tags => ["mysql_slow_log"]
        }	
}

filter {

 if "nginx_access_log" in [tags] {
  	grok {
    	match => { "message" => "%{COMBINEDAPACHELOG}" }
  	}
 }
 if "mysql_slow_log" in [tags] {
	grok {
	match => [ "message", "(?m)^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+%{USER:clienthost}\s+\[(?:%{IP:clientip})?\]\s+Id:\s+%{NUMBER:id:int}\n#\s+Schema:\s+%{USER:schema}\s+Last_errno:\s+%{NUMBER:lasterrorno:int}\s+Killed:\s+%{NUMBER:killedno: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}\s+Rows_affected:\s+%{NUMBER:rows_affected:int}\n#\s+Bytes_sent:\s+%{NUMBER:bytes_sent:int}\n\s*(?:use\s+%{USER:usedatabase};\s*\n)?SET\s+timestamp=%{NUMBER:timestamp};\n\s*(?<query>(?<action>\w+)\b.*)\s*(?:\n#\s+Time)?.*$"]
	}
        date {
            match => [ "timestamp", "UNIX" ]
            #remove_field => [ "timestamp" ]
        }

 }

}

output {


    if "nginx_access_log" in [tags] {
	elasticsearch {
		hosts=>["localhost:9200"]
		index=>"%{[@metadata][beat]}-%{+YYYY.MM.dd}"
		document_type => "%{[@metadata][type]}"
	}
    }

    if "mysql_slow_log" in [tags] {
	elasticsearch {
		hosts=>["localhost:9200"]
		index=>"mysql-slow-log-%{+YYYY.MM.dd}"
	}
    }
}
[root@logstash-server conf.d]#