如何高效处理Logstash流水线的字段默认值设置问题

我有一段Logstash流水线的filter代码如下：

filter {
    if condition {
        prune {
            blacklist_names => ["^cat[1-8]$","^classifier.version$","^accessory_check$"]
        }
        http {
            url => aURL
            query =>  {
                // relevant parts
            }
            verb => GET
            body_format => "json"
            ecs_compatibility => "disabled"
             add_field => {
                 "cat1" => "%{[body][cat1]}"
                 "cat2" => "%{[body][cat2]}"
                 "cat3" => "%{[body][cat3]}"
                 "cat4" => "%{[body][cat4]}"
                 "cat5" => "%{[body][cat5]}"
                 "cat6" => "%{[body][cat6]}"
                 "cat7" => "%{[body][cat7]}"
                 "cat8" => "%{[body][cat8]}"
                 "accessory_check" => "%{[body][accessory_check]}"
                 "classifier.version" => "${CLASSIFIER_LATEST_VERSION}"
             }
            remove_field => ["body","headers","@timestamp","@version"]
        }
    }
}

问题在于：当%{[body][cat1]}对应的字段为空时，cat1的值会变成字符串"%{[body][cat1]}"，但我希望此时cat1的值是空字符串""。

我自己写了一个解决方法，但代码非常冗余：

filter {
    if condition {
        prune {
            blacklist_names => ["^cat[1-8]$","^classifier.version$","^accessory_check$"]
        }
        http {
            url => aURL
            query =>  {
              // relevant parts
            }
            verb => GET
            body_format => "json"
            ecs_compatibility => "disabled"
#             add_field => {
#                 "cat1" => "%{[body][cat1]}"
#                 "cat2" => "%{[body][cat2]}"
#                 "cat3" => "%{[body][cat3]}"
#                 "cat4" => "%{[body][cat4]}"
#                 "cat5" => "%{[body][cat5]}"
#                 "cat6" => "%{[body][cat6]}"
#                 "cat7" => "%{[body][cat7]}"
#                 "cat8" => "%{[body][cat8]}"
#                 "accessory_check" => "%{[body][accessory_check]}"
                  "classifier.version" => "${CLASSIFIER_LATEST_VERSION}"
#             }
            remove_field => ["body","headers","@timestamp","@version"]
        }
    }

    if [body][cat1] {
             mutate{    add_field => {"cat1" => "%{[body][cat1]}"}  }
        } else {
            mutate{    add_field => {"cat1" => ""}  }
        }

    if [body][cat2] {
         mutate{    add_field => {"cat2" => "%{[body][cat2]}"}  }
    } else {
        mutate{    add_field => {"cat2" => ""}  }
    }
  // 其余cat3到cat8以及accessory_check的判断
}

有没有更简洁的方案实现相同的效果？

优化方案 #

方案1：使用Ruby插件批量处理字段

利用Ruby脚本遍历需要处理的字段列表，一次性完成赋值逻辑，避免重复代码：

filter {
    if condition {
        prune {
            blacklist_names => ["^cat[1-8]$","^classifier.version$","^accessory_check$"]
        }
        http {
            url => aURL
            query =>  {
                // relevant parts
            }
            verb => GET
            body_format => "json"
            ecs_compatibility => "disabled"
            # 只保留不需要处理的固定字段赋值
            add_field => {
                "classifier.version" => "${CLASSIFIER_LATEST_VERSION}"
            }
            # 注意：这里暂时不要删除body字段，留到Ruby处理完再删
            remove_field => ["headers","@timestamp","@version"]
        }
    }

    # 批量处理目标字段
    ruby {
        code => '
            # 定义需要处理的字段映射：目标字段 => body中的源字段
            field_mappings = {
                "cat1" => "[body][cat1]",
                "cat2" => "[body][cat2]",
                "cat3" => "[body][cat3]",
                "cat4" => "[body][cat4]",
                "cat5" => "[body][cat5]",
                "cat6" => "[body][cat6]",
                "cat7" => "[body][cat7]",
                "cat8" => "[body][cat8]",
                "accessory_check" => "[body][accessory_check]"
            }

            field_mappings.each do |target_field, source_path|
                # 获取body中的字段值，不存在则设为空字符串
                source_value = event.get(source_path) || ""
                event.set(target_field, source_value)
            end

            # 处理完后删除body字段
            event.remove("body")
        '
    }
}

方案2：结合mutate插件的copy与条件判断（无需Ruby）

先将body中的字段复制到目标字段，再用条件判断把插值失败的字段替换为空字符串：

filter {
    if condition {
        prune {
            blacklist_names => ["^cat[1-8]$","^classifier.version$","^accessory_check$"]
        }
        http {
            url => aURL
            query =>  {
                // relevant parts
            }
            verb => GET
            body_format => "json"
            ecs_compatibility => "disabled"
            add_field => {
                "cat1" => "%{[body][cat1]}",
                "cat2" => "%{[body][cat2]}",
                "cat3" => "%{[body][cat3]}",
                "cat4" => "%{[body][cat4]}",
                "cat5" => "%{[body][cat5]}",
                "cat6" => "%{[body][cat6]}",
                "cat7" => "%{[body][cat7]}",
                "cat8" => "%{[body][cat8]}",
                "accessory_check" => "%{[body][accessory_check]}",
                "classifier.version" => "${CLASSIFIER_LATEST_VERSION}"
            }
            remove_field => ["body","headers","@timestamp","@version"]
        }
    }

    # 批量替换所有插值失败的字段为空字符串
    mutate {
        gsub => [
            # 匹配以%{[body][开头的插值字符串，替换为空
            "cat1", '^\%{\[body\]\[cat1\]}$', "",
            "cat2", '^\%{\[body\]\[cat2\]}$', "",
            "cat3", '^\%{\[body\]\[cat3\]}$', "",
            "cat4", '^\%{\[body\]\[cat4\]}$', "",
            "cat5", '^\%{\[body\]\[cat5\]}$', "",
            "cat6", '^\%{\[body\]\[cat6\]}$', "",
            "cat7", '^\%{\[body\]\[cat7\]}$', "",
            "cat8", '^\%{\[body\]\[cat8\]}$', "",
            "accessory_check", '^\%{\[body\]\[accessory_check\]}$', ""
        ]
    }
}

方案3：使用Logstash的%{+field:default}插值语法（仅支持Logstash 7.10+）

如果你的Logstash版本在7.10及以上，可以直接使用带默认值的插值语法，无需额外处理：

filter {
    if condition {
        prune {
            blacklist_names => ["^cat[1-8]$","^classifier.version$","^accessory_check$"]
        }
        http {
            url => aURL
            query =>  {
                // relevant parts
            }
            verb => GET
            body_format => "json"
            ecs_compatibility => "disabled"
            add_field => {
                "cat1" => "%{[body][cat1]:}",
                "cat2" => "%{[body][cat2]:}",
                "cat3" => "%{[body][cat3]:}",
                "cat4" => "%{[body][cat4]:}",
                "cat5" => "%{[body][cat5]:}",
                "cat6" => "%{[body][cat6]:}",
                "cat7" => "%{[body][cat7]:}",
                "cat8" => "%{[body][cat8]:}",
                "accessory_check" => "%{[body][accessory_check]:}",
                "classifier.version" => "${CLASSIFIER_LATEST_VERSION}"
            }
            remove_field => ["body","headers","@timestamp","@version"]
        }
    }
}

这里的%{[body][cat1]:}表示如果[body][cat1]不存在或为空，就用空字符串作为默认值，完美解决插值失败的问题，代码最简洁。

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