Как проверить, является ли источник кафкой или битом в logstash?

Question

У меня есть два источника данных для моих журналов. Один - это ритм, а другой - Кафка, и я хочу создать индексы ES на основе источника. if kafka -> префикс index_name с kafka, а if beat префикс индекса с помощью beat.

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 
    }
}