Logstash Kafka输出插件是否支持条件语句？配置报错求助

核心结论 #

Logstash的Kafka输出插件不支持在插件配置块内部编写if/else条件语句，你碰到的语法错误正是因为把条件判断放在了kafka{}内部导致的。Logstash的条件逻辑必须放在插件层级之外，通过条件分支来配置不同的输出实例。

错误原因 #

你的配置将if/else嵌套在了kafka{}块里，但Logstash的配置语法规定，条件语句只能作用于整个输入、过滤或输出插件，不能在单个插件的配置项里嵌套条件，所以编译时触发了语法错误。

正确配置写法 #

把条件判断移到kafka插件外部，针对不同标签情况分别配置Kafka输出：

output {
  if "null-value" in [tags] {
    kafka {
      id => "plugin_SharedAlarmCreated_null"
      bootstrap_servers => "kafka-1:9092,kafka-2:9092,kafka-3:9092"
      codec => json
      topic_id => "Shared.Event.AlarmCreated"
      message_key => "%{Source}+%{Id}+NULL"
    }
  } else {
    kafka {
      id => "plugin_SharedAlarmCreated_normal"
      bootstrap_servers => "kafka-1:9092,kafka-2:9092,kafka-3:9092"
      codec => json
      topic_id => "Shared.Event.AlarmCreated"
      message_key => "%{Source}+%{Id}+%{SourceId}"
    }
  }
}

如果要减少重复代码，可以用变量定义公共配置：

# 定义公共配置变量
var.bootstrap_servers = "kafka-1:9092,kafka-2:9092,kafka-3:9092"
var.common_topic = "Shared.Event.AlarmCreated"

output {
  if "null-value" in [tags] {
    kafka {
      id => "plugin_SharedAlarmCreated_null"
      bootstrap_servers => "${bootstrap_servers}"
      codec => json
      topic_id => "${common_topic}"
      message_key => "%{Source}+%{Id}+NULL"
    }
  } else {
    kafka {
      id => "plugin_SharedAlarmCreated_normal"
      bootstrap_servers => "${bootstrap_servers}"
      codec => json
      topic_id => "${common_topic}"
      message_key => "%{Source}+%{Id}+%{SourceId}"
    }
  }
}

额外说明 #

Logstash的if/else条件仅能用于输入、过滤、输出这三个顶级阶段，用来控制整个插件是否启用，所有插件的内部配置项都不支持嵌套条件语法。

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