Filebeat 关键字多行匹配日志采集(multiline与include_lines)
很多同事认为filebeat采集日志不能做到多行处理,今天这里讨论下filebeat的multiline与include_lines。 先来个案例,以下日志,我们只要求采集error的字段, 1 2 3 2017 /06/22 11:26:30[error]26067 #0:*17918connect()failed(111:Connectionrefused)whileconnectingtoupstream,client:192.168.32.17,server:localhost,request:"GET/wss/HTTP/1.1",upstream:"http://192.168.12.106:8010/",host:"192.168.12.106" 2017 /06/22 11:26:30[info]26067 #0: 2017 /06/22 12:05:10[error]26067 #0:*17922open()"/data/programs/nginx/html/ws"failed(2:Nosuchfileordirectory),client:192.168.32.17,server:localhost,request:"GET/wsHTTP/1.1",host:"192.168.12.106" filebeat.yml文件配置如下: 1 2 3 4 5 6 7 8 9 filebeat.prospectors: -input_type:log paths: - /tmp/test .log include_lines:[ 'error' ] output.kafka: enabled: true hosts:[ "192.168.12.105:9092" ] topic:logstash-errors-log 查看下kafka队列 果然只有“error”关键字的日志被采集了 1 2 { "@timestamp" : "2017-06-23T08:57:25.227Z" , "beat" :{ "name" : "192.168.12.106" }, "input_type" : "log" , "message" : "2017/06/2212:05:10[error]26067#0:*17922open()/data/programs/nginx/html/wsfailed(2:Nosuchfileordirectory),client:192.168.32.17,server:localhost,request:GET/wsHTTP/1.1,host:192.168.12.106" , "offset" :30926, "source" : "/tmp/test.log" , "type" : "log" } { "@timestamp" : "2017-06-23T08:57:32.228Z" , "beat" :{ "name" : "192.168.12.106" }, "input_type" : "log" , "message" : "2017/06/2212:05:10[error]26067#0:*17922open()/data/programs/nginx/html/wsfailed(2:Nosuchfileordirectory),client:192.168.32.17,server:localhost,request:GET/wsHTTP/1.1,host:192.168.12.106" , "offset" :31342, "source" : "/tmp/test.log" , "type" : "log" } 再来多行案例: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 [2016-05-2512:39:04,744][DEBUG][action.bulk][Set][***][3]failedtoexecutebulkitem(index)index{[***][***][***], source [{***}} MapperParsingException[Fieldname[events.created]cannotcontain '.' ] atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseProperties(ObjectMapper.java:273) atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseObjectOrDocumentTypeProperties(ObjectMapper.java:218) atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parse(ObjectMapper.java:193) atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseProperties(ObjectMapper.java:305) atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseObjectOrDocumentTypeProperties(ObjectMapper.java:218) atorg.elasticsearch.index.mapper.object.RootObjectMapper$TypeParser.parse(RootObjectMapper.java:139) atorg.elasticsearch.index.mapper.DocumentMapperParser.parse(DocumentMapperParser.java:118) atorg.elasticsearch.index.mapper.DocumentMapperParser.parse(DocumentMapperParser.java:99) atorg.elasticsearch.index.mapper.MapperService.parse(MapperService.java:498) atorg.elasticsearch.cluster.metadata.MetaDataMappingService$PutMappingExecutor.applyRequest(MetaDataMappingService.java:257) atorg.elasticsearch.cluster.metadata.MetaDataMappingService$PutMappingExecutor.execute(MetaDataMappingService.java:230) atorg.elasticsearch.cluster.service.InternalClusterService.runTasksForExecutor(InternalClusterService.java:468) atorg.elasticsearch.cluster.service.InternalClusterService$UpdateTask.run(InternalClusterService.java:772) atorg.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.runAndClean(PrioritizedEsThreadPoolExecutor.java:231) atorg.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.run(PrioritizedEsThreadPoolExecutor.java:194) atjava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) atjava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) atjava.lang.Thread.run(Thread.java:745) filebeat.yml文件配置如下: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 filebeat.prospectors: -input_type:log paths: - /tmp/test .log multiline: pattern: '^\[' negate: true match:after fields: beat.name:192.168.12.106 fields_under_root: true output.kafka: enabled: true hosts:[ "192.168.12.105:9092" ] topic:logstash-errors-log 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 kafka队列如下: { "@timestamp" : "2017-06-23T09:09:02.887Z" , "beat" :{ "name" : "192.168.12.106" }, "input_type" : "log" , "message" :"[2016-05-2512:39:04,744][DEBUG][action.bulk][Set][***][3]failedtoexecutebulkitem(index)index{[***][***][***], source [{***}}\n MapperParsingException[Fieldname[events.created]cannotcontain '.' ]\natorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseProperties(ObjectMapper.java:273)\n atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseObjectOrDocumentTypeProperties(ObjectMapper.java:218)\n atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parse(ObjectMapper.java:193)\n atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseProperties(ObjectMapper.java:305)\n atorg.elasticsearch.index.mapper.object.ObjectMapper$TypeParser.parseObjectOrDocumentTypeProperties(ObjectMapper.java:218)\n atorg.elasticsearch.index.mapper.object.RootObjectMapper$TypeParser.parse(RootObjectMapper.java:139)\n atorg.elasticsearch.index.mapper.DocumentMapperParser.parse(DocumentMapperParser.java:118)\n atorg.elasticsearch.index.mapper.DocumentMapperParser.parse(DocumentMapperParser.java:99)\n atorg.elasticsearch.index.mapper.MapperService.parse(MapperService.java:498)\n atorg.elasticsearch.cluster.metadata.MetaDataMappingService$PutMappingExecutor.applyRequest(MetaDataMappingService.java:257)\n atorg.elasticsearch.cluster.metadata.MetaDataMappingService$PutMappingExecutor.execute(MetaDataMappingService.java:230)\n atorg.elasticsearch.cluster.service.InternalClusterService.runTasksForExecutor(InternalClusterService.java:468)\n atorg.elasticsearch.cluster.service.InternalClusterService$UpdateTask.run(InternalClusterService.java:772)\n atorg.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.runAndClean(PrioritizedEsThreadPoolExecutor.java:231)\n atorg.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor$TieBreakingPrioritizedRunnable.run(PrioritizedEsThreadPoolExecutor.java:194)\n atjava.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\n atjava.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\n atjava.lang.Thread.run(Thread.java:745)\n\n\n\n "," offset ":35737," source ":" /tmp/test .log "," type ":" log"} 可以看出multiline将多行日志汇总。 multiline与include_lines,结合使用。 filebeat.yml文件配置如下: 1 2 3 4 5 6 7 8 9 10 11 12 13 filebeat.prospectors: -input_type:log paths: - /tmp/test .log include_lines:[ 'error' ] multiline: pattern: '^\[' negate: true match:after output.kafka: enabled: true hosts:[ "192.168.12.105:9092" ] topic:logstash-errors-log 即日志中如果有"error"关键字的日志,进行多行合并,发送至kafka. 经验证,在日志不断输入的情况,会把不含"error"的行也进行合并,日志有间隔的情况输入,过滤效果比较好,具体结合业务情况实用吧。 总之一句话,filebeat可以多行合并和进行关键字日志采集。 本文转自 jackjiaxiong 51CTO博客,原文链接:http://blog.51cto.com/xiangcun168/1941401