vim /etc/logstash/conf.d/daemonset-filebeat-kafka-to-es.conf
input {
  kafka {
    bootstrap_servers => "10.0.7.53:9092,10.0.7.54:9092,10.0.7.55:9092"
    topics => ["daemonset-pod-console-log"]
  }
}


filter {
 if [fields][log_topic] == "daemonset-pod-console-log"  {
    mutate {
     remove_field => ["@version","agent","ecs"]
    }
  }
}

output {
  if [fields][log_topic] == "daemonset-pod-console-log"  {
    elasticsearch {
      hosts => ["10.0.7.46:9200","10.0.7.47:9200","10.0.7.48:9200"]
      index => "daemonset-pod-console-log-%{+YYYY.MM.dd}"
    }
  }
}