如何用Benthos Grok解析CheckPoint日志并提取指定字段？

你的日志属于带结构化键值对的Syslog格式，核心问题是原配置仅匹配了日志头部，未处理方括号内的业务字段。以下是完整的解析配置，可直接提取你需要的5个字段并输出JSON：

完整Benthos配置 #

input:
  type: file
  file:
    paths: [./input.txt]
    codec: lines

pipeline:
  processors:
    - grok:
        expressions:
          - '%{NGFWLOGFILE}'
        pattern_definitions:
          # 匹配Syslog头部
          SYSLOG_HEADER: '<%{POSINT:syslog_pri}>%{POSINT:syslog_ver} %{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:hostname} %{NOTSPACE:product} %{POSINT:pid} -'
          # 匹配单个key:"value"格式的键值对
          CHECKPOINT_KEYVAL: '%{NOTSPACE:key}:"%{DATA:value}"'
          # 完整日志匹配规则，捕获需要的字段
          NGFWLOGFILE: '%{SYSLOG_HEADER} \[%{CHECKPOINT_KEYVAL}(; %{CHECKPOINT_KEYVAL})*\]'
    # 整理输出字段，重命名并过滤不需要的内容
    - mapping: |
        root.timestamp = this.timestamp
        root.direction = this.key_value.ifdir
        root.src_ip = this.key_value.src
        root.dst_ip = this.key_value.dst
        root.src_port = this.key_value.s_port
        # 清理临时生成的冗余字段
        del(this.key_value)
        del(this.syslog_pri, this.syslog_ver, this.hostname, this.product, this.pid)

output:
  stdout:
    codec: json_lines

配置说明 #

1. Grok Pattern定义：

   • SYSLOG_HEADER：精准匹配日志开头的Syslog标准格式，直接提取出ISO8601格式的timestamp
   • CHECKPOINT_KEYVAL：匹配方括号内单个key:"value"结构的字段，Benthos会自动将所有键值对整理到key_value映射中
   • NGFWLOGFILE：组合头部和键值对规则，覆盖完整日志内容

2. Mapping处理器：

   • 从key_value映射中提取目标字段，将ifdir重命名为更直观的direction
   • 删除临时生成的冗余字段，确保输出JSON仅包含你需要的5个核心字段

输出示例 #

执行配置后，会输出如下JSON格式内容：

{"timestamp":"2023-01-21T17:18:05Z","direction":"outbound","src_ip":"10.21.9.1","dst_ip":"10.21.10.2","src_port":"46540"}

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