如何使用Grok捕获日志中全部消息并发送至Kibana？

嗨，我完全懂你的烦恼——作为Grok新手，碰到数组格式的日志确实容易卡壳，毕竟Grok默认只会匹配第一个符合规则的内容，没法直接处理数组里的多条消息。不过别慌，咱们结合Logstash的几个插件就能搞定，一步步来：

第一步：先解析整个JSON结构 #

你的原始日志是JSON格式，所以第一步得先用json过滤器把整个结构解析出来，把包含6条消息的数组提取到单独的字段里。配置示例：

filter {
  json {
    source => "message"  # 这里填你原始日志中包含JSON的字段名
    target => "parsed_log"  # 解析后的内容会放到这个字段下
  }
}

执行这一步后，parsed_log.message就会变成那个包含所有6条消息的数组，parsed_log.seq就是对应的序列号信息。

第二步：用Split插件拆分数组 #

接下来用split插件把数组里的每一条消息拆成独立的Logstash事件——这样每条消息都会单独成为一个待处理的事件，而不是挤在同一个数组里。配置示例：

split {
  field => "[parsed_log][message]"  # 指定要拆分的数组字段
}

如果想把序列号（seq的w和c）保留到每个拆分后的事件里，可以在split之前加个mutate插件复制字段：

mutate {
  add_field => {
    "seq_w" => "%{[parsed_log][seq][w]}"
    "seq_c" => "%{[parsed_log][seq][c]}"
  }
}

第三步：用Grok匹配单条消息 #

现在每个事件里的[parsed_log][message]就是单独的一条日志了，这时候再用Grok去匹配提取字段就没问题了。根据你的日志格式，写个对应的Grok表达式：

grok {
  match => { "[parsed_log][message]" => "%{TIME:timestamp} %{DATA:task_id} %{DATA:code}%{DATE:date} %{TIME:log_time} %{WORD:level} %{DATA:user} %{GREEDYDATA:log_content}" }
  remove_field => "[parsed_log][message]"  # 匹配完成后可以删掉原始消息字段，让Kibana里的字段更整洁
}

注意：你日志里的奇怪字符（比如\u0000）可以不用特意处理，GREEDYDATA会把剩下的内容全匹配上；如果需要更精准的匹配，可以调整Grok模式里的占位符（比如把DATA换成更具体的模式）。

测试配置 #

你可以用Logstash的命令行测试功能快速验证配置是否生效，比如：

bin/logstash -e 'input { stdin {} } filter { json { source => "message" target => "parsed_log" } mutate { add_field => { "seq_w" => "%{[parsed_log][seq][w]}" "seq_c" => "%{[parsed_log][seq][c]}" } } split { field => "[parsed_log][message]" } grok { match => { "[parsed_log][message]" => "%{TIME:timestamp} %{DATA:task_id} %{DATA:code}%{DATE:date} %{TIME:log_time} %{WORD:level} %{DATA:user} %{GREEDYDATA:log_content}" } } output { stdout { codec => rubydebug } }'

把你的JSON日志粘贴到控制台，就能看到每条消息都被拆分成独立事件，并且Grok正确提取了字段。

总结一下：核心思路是先解析JSON拆出数组，再拆分数组为单条事件，最后用Grok处理单条消息——别想着用Grok直接处理数组，它本来就不是干这个的~

内容的提问来源于stack exchange，提问作者Joie