Logstash解析嵌套KV格式审计日志的问题咨询

原始日志示例 #

<183>Apr 26 12:53:53 host.example tag_audit_log: type=USER_ACCT msg=audit(1989212.965:15124458): pid=20982 uid=0 auid=427890281 ses=4294967295 subj=system_u:system_r:init_t:s0 msg='op=PAM:accounting grantors=pam_unix acct="root" exe="/usr/lib/systemd/systemd" hostname=? addr=? terminal=? res=success'

当前使用的Logstash配置 #

filter {
  grok {
    match => { "message" => "%{SYSLOG5424PRI}%{SYSLOGTIMESTAMP:date} %{NOTSPACE:hostname} %{NOTSPACE:location}: %{GREEDYDATA:kv_message}" }
    overwrite => [ message ]
  }

  kv {
    source => [ "kv_message" ]
  }
}

遇到的问题 #

大部分字段解析正常，但msg被KV解析为数组：

"msg" => [
[0] "audit(1989212.965:15124458):",
[1] "'op=PAM:accounting"
]

需要进一步解析msg[1]中的内容，同时兼容不含这类嵌套字段的日志。

解决方案 #

1. 拆分第一个msg字段，提取审计元数据 #

修改grok模式，单独提取msg=audit(...):部分的内容，避免和后续重复的msg字段冲突：

grok {
  match => { "message" => "%{SYSLOG5424PRI}%{SYSLOGTIMESTAMP:date} %{NOTSPACE:hostname} %{NOTSPACE:location}: type=%{NOTSPACE:audit_type} msg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_seq}\): %{GREEDYDATA:remaining_kv}" }
  overwrite => [ "message" ]
  tag_on_failure => [ "_grok_parse_failure_audit" ]
}

2. 重命名嵌套msg字段，避免KV冲突 #

用mutate将第二个msg='...'替换为新字段pam_details='...'，再进行KV解析：

if !("_grok_parse_failure_audit" in [tags]) {
  mutate {
    gsub => [
      "remaining_kv", "msg='", "pam_details='"
    ]
  }

  kv {
    source => "remaining_kv"
    field_split => " "
    value_split => "="
    remove_field => [ "remaining_kv" ]
  }
}

3. 解析嵌套的pam_details内容 #

去掉pam_details末尾的单引号，再用KV解析内部的键值对：

if [pam_details] {
  mutate {
    gsub => [
      "pam_details", "'$", ""
    ]
  }
  kv {
    source => "pam_details"
    field_split => " "
    value_split => "="
    remove_field => [ "pam_details" ]
  }
}

4. 兼容无嵌套字段的日志 #

通过标签判断，对解析失败的日志回退到原始KV逻辑：

if "_grok_parse_failure_audit" in [tags] {
  kv {
    source => "kv_message"
    remove_field => [ "kv_message" ]
  }
}

完整配置 #

filter {
  # 尝试解析带嵌套msg的审计日志
  grok {
    match => { "message" => "%{SYSLOG5424PRI}%{SYSLOGTIMESTAMP:date} %{NOTSPACE:hostname} %{NOTSPACE:location}: type=%{NOTSPACE:audit_type} msg=audit\(%{NUMBER:audit_epoch}:%{NUMBER:audit_seq}\): %{GREEDYDATA:remaining_kv}" }
    overwrite => [ "message" ]
    tag_on_failure => [ "_grok_parse_failure_audit" ]
  }

  # 处理嵌套字段逻辑
  if !("_grok_parse_failure_audit" in [tags]) {
    mutate {
      gsub => [
        "remaining_kv", "msg='", "pam_details='"
      ]
    }

    kv {
      source => "remaining_kv"
      field_split => " "
      value_split => "="
      remove_field => [ "remaining_kv" ]
    }

    if [pam_details] {
      mutate {
        gsub => [
          "pam_details", "'$", ""
        ]
      }
      kv {
        source => "pam_details"
        field_split => " "
        value_split => "="
        remove_field => [ "pam_details" ]
      }
    }
  }

  # 回退处理无嵌套字段的日志
  if "_grok_parse_failure_audit" in [tags] {
    kv {
      source => "kv_message"
      remove_field => [ "kv_message" ]
    }
  }
}

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