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

Question

我的日志有两个数据来源。一个是节拍，一个是kafka，我想根据源创建ES索引。如果kafka ->使用kafka作为前缀index_name，如果使用节拍作为索引名的前缀。

input {
  beats {
    port => 9300
  }
}

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["my-topic"]
    codec => json
  }
}

output {
    # if kafka
      elasticsearch {
        hosts => "http://localhost:9200"
        user => "elastic"
        password => "password"
        index  => "[kafka-topic]-my-index"
     }
    # else if beat
    elasticsearch {
        hosts => "http://localhost:9200"
        user => "elastic"
        password => "password"
        index  => "[filebeat]-my-index"
     }
}

Answer 1

在输入中添加标记，并使用它们来过滤输出。

input {
  beats {
    port => 9300
    tags => ["beats"]
  }
}

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["my-topic"]
    codec => json
    tags => ["kafka"]
  }
}

output {
    if "beats" in [tags] { 
        output for beats 
    }
    if "kafka" in [tags] { 
        output for kafka 
    }
}