logstash解析日志文件的配置

话不多说,直接上干货！

如有问题下方评论区评论,本人直接回复

以下是读取两个java服务的日志文件进行解析

input {
	file {
    		#日志文件地址
    		path => "/usr/local/sffkszh_app/scm/logs/*.log.*"
    		type => "scm"
			#排除不想监听的文件
        	#exclude => "1.log"
        	
        	#添加自定义的字段
        	#add_field => {"test"=>"test"}
        	#增加标签
        	#tags => "tag1"
		
        	#设置新事件的标志
        	#delimiter => "\n"
		
        	#设置多长时间扫描目录，发现新文件
        	#discover_interval => 15
        	#设置多长时间检测文件是否修改
        	#stat_interval => 1
		
        	#监听文件 beginning起始位置，默认是end
        	start_position => "beginning"
		
        	#监听文件读取信息记录的位置,重启logstash不会重复读取数据
        	sincedb_path => "../scmRecord.txt"
        	#设置多长时间会写入读取的位置信息
        	#sincedb_write
			#_interval => 15
			
			#docinfo => true
  }
  
  	file {
    		path => "/usr/local/sffkszh_app/asu/log/*.log"
    		type => "asu"
			#排除不想监听的文件
        	#exclude => "1.log"
        	
        	#添加自定义的字段
        	#add_field => {"test"=>"test"}
        	#增加标签
        	#tags => "tag1"
		
        	#设置新事件的标志
        	#delimiter => "\n"
		
        	#设置多长时间扫描目录，发现新文件
        	#discover_interval => 15
        	#设置多长时间检测文件是否修改
        	#stat_interval => 1
		
        	#监听文件 beginning起始位置，默认是end
        	start_position => "beginning"
		
        	#监听文件读取信息记录的位置,重启logstash不会重复读取数据
        	sincedb_path => "../asuRecord.txt"
        	#设置多长时间会写入读取的位置信息
        	#sincedb_write
			#_interval => 15
			
			#docinfo => true
  }
}
 
filter {
	multiline {
		pattern => "%{DATESTAMP}"
		negate => true
		what => "previous"
	}
	
	if [type] == 'scm' {
		if "index:" in [message] {
			grok {
            	match => [
                # 解析日志文件定义的打印数据
                	"message","%{URIHOST:logDate} %{TIME:logTime} %{SYSLOG5424SD} %{LOGLEVEL:logLevel}  %{JAVACLASS}- index:%{NOTSPACE:esIndex},type:%{NOTSPACE:esType},data:%{DATA:esData},description:%{NOTSPACE:esDescription}"
      	    	]
        	}
		}else {
			drop {}
		}
	}
	
	if [type] == 'asu' {
		if "index:" in [message] {
			grok {
            	match => [
					"message","%{URIHOST:logDate} %{HAPROXYTIME:logTime} %{SYSLOG5424SD} %{LOGLEVEL:logLevel}  %{JAVACLASS} %{SYSLOG5424SD} - index:%{DATA:esIndex},type:%{DATA:esType},data:%{DATA:esData},description:%{NOTSPACE:esDescription}"
     	    	]
        	}
		}else {
			drop {}
		}
	}
	
	ruby {
		code => "event['time'] = event['@timestamp']"
	}
  # 时间名称修改
	mutate {
		add_field => ["time", "%{@timestamp}"]
	}
	
  # 删除不需要的字段
	mutate {
		remove_field => ["@timestamp"]
		remove_field => ["message"]
		remove_field => ["@version"]
		remove_field => ["tags"]
    }
	
		
   # ruby {
   # code => "
#		event.set('userMessage', event.get('userMessage'))
#		event.set('userId', event.get('userId'))
#	"
 # }

	if [logLevel] == "WARN" {
		drop {}
	}

	if [logLevel] == "ERROR" {
		drop {}
	}
	
	# message中不包含index的将删除
	#if ([message] !~ "^index") {
	#	drop {}
	#}

	date {
    		match => ["@timestamp", "yyyy-MM-dd HH:mm:ss"]
		locale => "cn"	
  	}
 
}
 
output {
	elasticsearch {
		hosts => "192.168.10.14:9200"
    # 动态索引库
		index => "%{esIndex}"
  	# 动态类型
		#document_type => "%{esType}"
	}
	stdout {
		# JSON格式输出
		codec => json_lines
	}

}