input {
  beats {
    port => 5044
    ssl => false
  }
}

filter {
  if [type] == "kraken-metrics" {
    grok {
      match => { "message" => "INFO: %{GREEDYDATA:kraken-message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }

    json {
      source => "kraken-message"
    }

    mutate {
      remove_field => ["kraken-message", "message"]
    }
  }
}

output {
  if [type] == "kraken-metrics" {
    elasticsearch {
      hosts => ["metrics-es.zooxlabs.com:80"]
      sniffing => false
      manage_template => false
      index => "kraken-metrics-%{+YYYY.MM.dd}"
      document_type => "kraken-metrics"
    }
  }
}
