如何编写Logstash Grok正则提取多个BLOCK块内的START和END值

问题原因 #

你之前的配置无法生效主要有两个核心原因：

• 若日志按行逐行处理，即使添加多行匹配符(?m)也无法跨行匹配BLOCK块内容
• Grok默认仅执行单次匹配，无法捕获重复出现的同模式内容

推荐方案（适配不定数量BLOCK块） #

优先使用 grok + ruby过滤器 组合实现，灵活性更高，支持任意数量的BLOCK块提取。

步骤1：配置input合并多行日志 #

首先需要将同一份日志的所有行合并为单个事件字段，避免逐行处理：

input {
  file {
    path => "/你的日志文件路径.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    # 合并所有行为单条事件
    codec => multiline {
      pattern => "^__NEVER_MATCH__"
      negate => true
      what => "previous"
      max_lines => 10000
    }
  }
}

步骤2：配置filter提取BLOCK内容 #

用ruby正则全局匹配所有BLOCK块，直接构造你需要的输出结构：

filter {
  ruby {
    code => "
      # 全局匹配所有BLOCK块的START和END值
      block_matches = event.get('message').scan(/\[BLOCK\]\s*START=(?<start>\d+)\s*END=(?<end>\d+)\s*\[\/BLOCK\]/m)
      block_matches.each_with_index do |item, index|
        event.set(\"BLOCK#{index+1}\", {
          \"START\" => item[0],
          \"END\" => item[1]
        })
      end
    "
  }
  # 可选：清理不需要的原始字段
  # mutate {
  #   remove_field => ["message"]
  # }
}

输出结果和你要求的结构完全一致：

{ "BLOCK1": { "START":"1", "END":"2"}, "BLOCK2": { "START":"3", "END":"4" } }

可选方案（固定数量BLOCK块用纯Grok实现） #

如果确定BLOCK块数量固定，也可以纯用Grok实现，以最多2个BLOCK为例：

Grok匹配模式 #

(?m).*?\[BLOCK\]\s*START=(?<BLOCK1_START>\d+)\s*END=(?<BLOCK1_END>\d+)\s*\[\/BLOCK\].*?\[BLOCK\]\s*START=(?<BLOCK2_START>\d+)\s*END=(?<BLOCK2_END>\d+)\s*\[\/BLOCK\].*

补充mutate调整结构 #

filter {
  grok {
    match => { "message" => "(?m).*?\[BLOCK\]\s*START=(?<BLOCK1_START>\d+)\s*END=(?<BLOCK1_END>\d+)\s*\[\/BLOCK\].*?\[BLOCK\]\s*START=(?<BLOCK2_START>\d+)\s*END=(?<BLOCK2_END>\d+)\s*\[\/BLOCK\].*" }
  }
  mutate {
    add_field => {
      "[BLOCK1][START]" => "%{BLOCK1_START}"
      "[BLOCK1][END]" => "%{BLOCK1_END}"
      "[BLOCK2][START]" => "%{BLOCK2_START}"
      "[BLOCK2][END]" => "%{BLOCK2_END}"
    }
    remove_field => ["BLOCK1_START", "BLOCK1_END", "BLOCK2_START", "BLOCK2_END", "message"]
  }
}

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