input {
    kafka {
        bootstrap_servers => "10.201.1.46:9092"
        topics => ['linklog_common_error']
        group_id => "linklog_error"
        consumer_threads => 4
        decorate_events => true
        codec => "json"
        auto_offset_reset => "latest"
        type => "linklog_error"
    }

    kafka {
        bootstrap_servers => "10.201.1.46:9092"
        topics => ['linklog_std_log']
        group_id => "linklog_std"
        consumer_threads => 4
        decorate_events => true
        codec => "json"
        auto_offset_reset => "latest"
        type => "linklog_std"
    }

}

output {

    if [type] == "linklog_std" {
        elasticsearch {
            hosts => ["10.201.1.46:9200"]
            index => "linklog_std_log_%{+YYYY-MM-dd}"
        }
    }

    if [type] == "linklog_error" {
        elasticsearch {
            hosts => ["10.201.1.46:9200"]
            index => "linklog_common_error_%{+YYYY-MM-dd}"
        }
    }
}

filter {
        json {
           source => "%{message}"
        }
       
  # mutate {
  #      add_field => {"@message" => "%{message}"}
  #     }
  # json {
  #      source => "@message"
  #    }

  # mutate {
  #      remove_field => ["@message"]
  #  }

}