Grok中复杂的logstash模式

Question

我在日志中有以下三行，需要通过logstash进行ElasticSearch的查找。

2020-01-27 13:30:43,536 INFO  com.test.bestmatch.streamer.function.BestMatchProcessor  - Best match for ID: COi0620200110450BAD5CB723457A9B4747F1727 Total Batch Processing time: 3942
2020-01-27 13:30:43,581 INFO  HTTPConnection                                              - COi0620200110450BAD5CB723457A9B4747F1727 | People: 51 | Addresses: 5935 | HTTP Query Time: 24
2020-01-27 13:30:43,698 INFO  bestRoute                                                    - COi0620200110450BAD5CB723457A9B4747F1727 | Touch Points: 117 | Best Match Time 3943

我尝试了各种古怪的模式，但无法达到任何具体的模式。

根据请求编辑

在特定日志条目的上下文中，我在ES中需要以下内容

第一行

ID: COi0620200110450BAD5CB723457A9B4747F1727 
Total Batch Processing time: 3942 

第二行

ID: COi0620200110450BAD5CB723457A9B4747F1727 
People: 51 
Addresses: 5935 
HTTP Query Time: 24

第三行

Touch Points 117 
Best Match Time: 3943.

输出来自Flink日志。如果那里有flink模式，请告诉我。

Answer 1

第一行：

^%{TIMESTAMP_ISO8601:time}\s*%{LOGLEVEL:loglevel}.*ID: (?<ID>[\w\d]*).*time: (?<total_time>[\d]*)$

第二行：

^%{TIMESTAMP_ISO8601:time}\s*%{LOGLEVEL:loglevel}.* - (?<ID>[\w]*).*People: (?<people>[\w]*).*Addresses: (?<addresses>[\d]*).*HTTP Query Time: (?<query_time>[\d]*)$

第三行：

^%{TIMESTAMP_ISO8601:time}\s*%{LOGLEVEL:loglevel}.* - (?<ID>[\w]*).*Touch Points: (?<touch_points>[\d]*).*Best Match Time (?<best_match_time>[\d]*)$

有很多解析方法，这只是一种方法。我建议您调整用于新ECS的字段名称。 https://www.elastic.co/guide/en/ecs/current/index.html