如何为Filebeat自定义解析规则？特殊表格型日志解析求助

针对你这种竖线分隔的表格日志，可通过Filebeat内置处理器组合实现解析，具体步骤如下：

1. 过滤无效行 #

先过滤掉表头、分隔线、汇总行这类非数据内容，使用drop_event处理器：

processors:
  - drop_event:
      when:
        or:
          - contains:
              message: "==============="
          - contains:
              message: "--------------------------------------------------------------------"
          - contains:
              message: "ALL"
          - contains:
              message: "TOTAL|"

2. 解析数据字段 #

根据日志的竖线分隔结构，推荐两种解析方式：

方式一：使用dissect处理器（适合固定格式） #

利用竖线作为分隔符定义匹配模板，同时去除字段前后空格：

processors:
  # ... 前置drop_event处理器
  - dissect:
      tokenizer: "%{name}|%{TOTAL}|%{FAIL}|%{FailNor}|%{FailErr}|%{SUMVAL}|%{AVG_ms}|%{MAX_ms}|%{MIN_ms}|%{gt_2ms}|%{gt_10ms}|%{gt_50ms}|%{gt_100ms}|%{MAX_RECORD}|"
      field: "message"
      target_prefix: ""
  - trim:
      fields: ["name", "TOTAL", "FAIL", "FailNor", "FailErr", "SUMVAL", "AVG_ms", "MAX_ms", "MIN_ms", "gt_2ms", "gt_10ms", "gt_50ms", "gt_100ms", "MAX_RECORD"]

注：将原字段名AVG(ms)改为AVG_ms，避免特殊字符引发配置问题。

方式二：使用grok处理器（适合格式略有波动的情况） #

通过正则表达式匹配字段，兼容性更强，且可直接指定字段类型：

processors:
  # ... 前置drop_event处理器
  - grok:
      match:
        message: "^%{DATA:name}\\s+\\|\\s+%{NUMBER:TOTAL:int}\\s+\\|\\s+%{NUMBER:FAIL:int}\\s+\\|\\s+%{NUMBER:FailNor:int}\\s+\\|\\s+%{NUMBER:FailErr:int}\\s+\\|\\s+%{NUMBER:SUMVAL:int}\\s+\\|\\s+%{NUMBER:AVG_ms:float}\\s+\\|\\s+%{NUMBER:MAX_ms:float}\\s+\\|\\s+%{NUMBER:MIN_ms:float}\\s+\\|\\s+%{NUMBER:gt_2ms:int}\\s+\\|\\s+%{NUMBER:gt_10ms:int}\\s+\\|\\s+%{NUMBER:gt_50ms:int}\\s+\\|\\s+%{NUMBER:gt_100ms:int}\\s+\\|\\s+%{DATA:MAX_RECORD}\\s+\\|"

3. 字段类型转换（仅dissect方式需要） #

若使用dissect解析，需将字符串类型的数值转换为对应类型：

processors:
  # ... 前置drop_event、dissect、trim处理器
  - convert:
      fields:
        - {from: "TOTAL", to: "integer"}
        - {from: "FAIL", to: "integer"}
        - {from: "FailNor", to: "integer"}
        - {from: "FailErr", to: "integer"}
        - {from: "SUMVAL", to: "integer"}
        - {from: "AVG_ms", to: "float"}
        - {from: "MAX_ms", to: "float"}
        - {from: "MIN_ms", to: "float"}
        - {from: "gt_2ms", to: "integer"}
        - {from: "gt_10ms", to: "integer"}
        - {from: "gt_50ms", to: "integer"}
        - {from: "gt_100ms", to: "integer"}

完整配置示例 #

整合以上步骤的Filebeat配置片段：

filebeat.inputs:
  - type: log
    paths:
      - /path/to/your/log/file.log

processors:
  - drop_event:
      when:
        or:
          - contains:
              message: "==============="
          - contains:
              message: "--------------------------------------------------------------------"
          - contains:
              message: "ALL"
          - contains:
              message: "TOTAL|"
  - grok:
      match:
        message: "^%{DATA:name}\\s+\\|\\s+%{NUMBER:TOTAL:int}\\s+\\|\\s+%{NUMBER:FAIL:int}\\s+\\|\\s+%{NUMBER:FailNor:int}\\s+\\|\\s+%{NUMBER:FailErr:int}\\s+\\|\\s+%{NUMBER:SUMVAL:int}\\s+\\|\\s+%{NUMBER:AVG_ms:float}\\s+\\|\\s+%{NUMBER:MAX_ms:float}\\s+\\|\\s+%{NUMBER:MIN_ms:float}\\s+\\|\\s+%{NUMBER:gt_2ms:int}\\s+\\|\\s+%{NUMBER:gt_10ms:int}\\s+\\|\\s+%{NUMBER:gt_50ms:int}\\s+\\|\\s+%{NUMBER:gt_100ms:int}\\s+\\|\\s+%{DATA:MAX_RECORD}\\s+\\|"

output.elasticsearch:
  hosts: ["localhost:9200"]
  index: "table-logs-%{+yyyy.MM.dd}"

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