# 
input {
  kafka {
    bootstrap_servers => "elk:9092,elk2:9092,elk3:9092"
    topics => ["elk_devlogs"]
    codec => "json" 
    consumer_threads => 1
    decorate_events => true
  }
}

filter{

    mutate{
        remove_field => ["ecs","@version","input","host"]
        remove_field => "agent"
        remove_field => "[json][time]"
        remove_field => "[json][stream]"
        remove_field => "[fields]"
    }

    mutate {
      add_field => { "log_path" => "%{log}" }
      rename => ["[json][log]", "message"]
    }

    grok {
      match => {"log_path" => "/var/log/containers/%{DATA:pod_name}_%{DATA:namespace}_%{GREEDYDATA:app}-%{DATA:container_id}.log"}
      remove_field  => "log_path"
      remove_field  => "log"
    }
}

output {
      elasticsearch {
          action   => "index"
          hosts    => ["elk:9200","elk2:9200","elk3:9200"]
          index    => "%{[namespace]}-%{[app]}-%{+YYYY.MM.dd}"
          user     => "elastic"
          password => "xlFnyMMyZiqjkzLIV5Kd"
      }
}