如何在logstash中检查来源是kafka还是beat？

在logstash中检查消息来源是kafka还是beat，可以通过以下方式进行：

使用Logstash的input插件：通过配置Logstash的input插件来指定消息来源，可以使用kafka input插件或者beats input插件，具体根据消息来源选择相应的input插件。配置示例如下：

Kafka input插件：

input {
  kafka {
    bootstrap_servers => "kafka_server:9092"
    topics => ["your_topic"]
  }
}

Beats input插件：

input {
  beats {
    port => "5044"
  }
}

使用Logstash的filter插件：在处理消息的过程中，可以使用filter插件来检查消息的来源。可以使用if条件语句结合消息的特征来判断消息来源。例如：

filter {
  if [source] =~ "kafka" {
    # 处理kafka来源的消息
  } else if [source] =~ "beat" {
    # 处理beat来源的消息
  }
}

使用Logstash的output插件：根据消息的来源，可以使用相应的output插件将消息发送到目标位置。根据业务需求选择合适的output插件。示例配置如下：

发送到Elasticsearch：

output {
  if [source] =~ "kafka" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "kafka_index"
    }
  } else if [source] =~ "beat" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "beat_index"
    }
  }
}