如何在Logstash Grok模式中编写条件判断处理空值字段

嘿，我来帮你搞定这个头疼的问题！你的核心痛点是：当Y或age字段为空、缺失时，Grok的严格匹配会直接失败，导致日志无法被解析，自然也就没法写入Elasticsearch。我们分两步来解决：

第一步：让Grok支持可选字段匹配 #

Grok默认是严格匹配的，只要某个字段的模式没匹配上，整个解析就会失败。我们需要把Y和age改成可选匹配，用非捕获组(?: ... )?来包裹这些字段的匹配规则，这样即使字段为空，整个Grok也能正常完成匹配：

grok {
  match => { 
    "message" => "%{TIMESTAMP_ISO8601:formattedDate}.* X: %{DATA:X} (?:Y: %{NUMBER:Y})? (?:Z: %{DATA:Z})? (?:age: %{NUMBER:age:int})?" 
  }
  # 关闭默认的匹配失败标签，避免这些日志被误过滤
  tag_on_failure => []
}

这里的(?: ... )?表示这个匹配段是可选的——能匹配到字段就解析，匹配不到也不会触发失败。同时去掉_grokparsefailure标签，确保这些带空值的日志能进入后续过滤流程。

第二步：自动将空的age设为0 #

接下来我们需要处理age的空值情况，这里推荐用ruby过滤器，逻辑更直观：

ruby {
  code => '
    # 如果age是null或者空字符串，就设为0
    event.set("age", 0) if event.get("age").nil? || event.get("age").to_s.strip.empty?
  '
}

如果你更习惯用Logstash内置的mutate过滤器，也可以用coalesce来实现：

mutate {
  # 先尝试把age转成整数，空值会变成null
  convert => { "age" => "integer" }
  # 新增一个临时字段，值为0
  add_field => { "[age_temp]" => "0" }
}
mutate {
  # 用coalesce取第一个非空值：如果age为空，就用age_temp的0填充
  coalesce => { "age" => "%{age_temp}" }
  # 删掉临时字段
  remove_field => ["age_temp"]
}

修改后的完整配置 #

把这些修改整合到你的配置里，我还帮你修正了date过滤器里的一个小错误（原来的TimeStamp应该是formattedDate，因为你的Grok解析的字段是这个）：

input { 
  file { 
    path => ["C:/log/*.log"] 
    start_position => "beginning" 
    discover_interval => 10 
    stat_interval => 10 
    sincedb_write_interval => 10 
    close_older => 10 
    codec => multiline { 
      pattern => "^%{TIMESTAMP_ISO8601}\|" 
      negate => true 
      what => "previous" 
    } 
  } 
} 

filter { 
  grok { 
    match => { 
      "message" => "%{TIMESTAMP_ISO8601:formattedDate}.* X: %{DATA:X} (?:Y: %{NUMBER:Y})? (?:Z: %{DATA:Z})? (?:age: %{NUMBER:age:int})?" 
    }
    tag_on_failure => []
  } 

  # 处理age空值，自动设为0
  ruby {
    code => '
      event.set("age", 0) if event.get("age").nil? || event.get("age").to_s.strip.empty?
    '
  }

  date { 
    timezone => "Europe/Istanbul" 
    match => ["formattedDate", "ISO8601"] 
  } 

  json{ 
    source => "request" 
    target => "parsedJson" 
  } 

  mutate { 
    remove_field => [ "path","message","tags","@version"] 
  } 
} 

output { 
  stdout { codec => rubydebug } 
  elasticsearch { 
    hosts => [ "http://localhost:9200" ] 
    index => "logstash-%{+YYYY.MM}" 
  } 
}

这样调整后，不管Y或age是空值还是缺失，Grok都能正常解析日志，age字段也会被自动填充为0，日志就能顺利写入Elasticsearch啦！

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