如何在Logstash过滤器插件中处理banana字段Null值并保留数据？

问题背景 #

我通过Logstash HTTP插件接收数据源，传入的数据模型包含myArray数组，结构如下：

{
  "myArray": [
    {
      "myGrocery": {
        "myId": "aString",
        "apple": "aString",
        "banana": aNumber,
        "vegetable": {
          "garlic": "aString",
          "onion": "aString",
          "mushroom": "aString"    
        }
      }
    }
  ]
}

当前使用的Logstash过滤器配置如下：

filter {

    split {
        field => "myArray"
        add_field => {
            "my_id" => "%{[myArray][myGrocery][myId]}"
            "apple" => "%{[myArray][myGrocery][apple]}"
            "banana" => "%{[myArray][myGrocery][banana]}"           
            "vegetable" => "%{[myArray][myGrocery][vegetable]}"
        }
        remove_field => "myArray"
    }

    json {
        source => "[vegetable]"
        target => "[vegetable]"
    }

    mutate {
        rename => { "[vegetable][garlic]" => "[vegetable][myNewGarlic]" }
        rename => { "[vegetable][onion]" => "[vegetable][myNewOnion]" }
        rename => { "[vegetable][mushroom]" => "[vegetable][myNewMushroom]" }
    }

    mutate {
        remove_field => [ "host", "headers", "@version", "url", "event", "user_agent", "http" ]
    }

}

当前配置运行正常，但需要实现以下需求：

• 当banana字段传入为null时，将其转换为字符串"null"
• 同时记录日志"banana is null"
• 不丢弃整条数据

期望输出格式如下：

"myId": "aString",
"apple": "aString",
"banana": "null",
"myNewGarlic" : "aString",
"myNewOnion": "aString",
"myNewMushroom": "aString"

实现方法 #

完全可以实现，通过Logstash的条件判断结合基础过滤器就能完成，以下是修改后的完整配置：

filter {

    split {
        field => "myArray"
        add_field => {
            "my_id" => "%{[myArray][myGrocery][myId]}"
            "apple" => "%{[myArray][myGrocery][apple]}"
            "banana" => "%{[myArray][myGrocery][banana]}"           
            "vegetable" => "%{[myArray][myGrocery][vegetable]}"
        }
        remove_field => "myArray"
    }

    json {
        source => "[vegetable]"
        target => "[vegetable]"
    }

    # 处理banana空值逻辑
    if [banana] == "" or [banana] == null {
        # 将空值替换为"null"字符串
        mutate {
            replace => { "banana" => "null" }
        }
        # 标记该事件并记录日志
        mutate {
            add_tag => ["banana_null"]
        }
        log {
            message => "banana is null"
            level => "info"
            tags => ["banana_null"]
        }
    }

    mutate {
        rename => { "[vegetable][garlic]" => "[vegetable][myNewGarlic]" }
        rename => { "[vegetable][onion]" => "[vegetable][myNewOnion]" }
        rename => { "[vegetable][mushroom]" => "[vegetable][myNewMushroom]" }
    }

    mutate {
        remove_field => [ "host", "headers", "@version", "url", "event", "user_agent", "http" ]
    }

}

关键逻辑说明 #

1. 空值判断：用if [banana] == "" or [banana] == null覆盖两种空值场景——字段显式为null，或字段不存在导致的空字符串
2. 字段替换：通过mutate的replace动作，将空值的banana字段强制设置为字符串"null"
3. 日志记录：先给事件添加banana_null标签，再用log过滤器针对带该标签的事件输出指定日志，避免无差别打印

如果需要在日志里带上文档ID方便排查，可以把log部分替换为Ruby代码实现更灵活的日志内容：

ruby {
    code => 'logger.info("banana is null - document ID: #{event.get("[@metadata][_id]")}")'
}

内容的提问来源于stack exchange，提问作者Murat K.