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

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  } else if [type] == "pipedream_log" {
    grok {
      match => { "message" => "\A\[%{TIMESTAMP_ISO8601:logged_time}] %{GREEDYDATA:pipedream_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }

    grok {
      match => { "source" => "\A/var/log/pipedream/(?<pipeline_name>[a-zA-Z0-9]+)_(?<stage_name>[a-zA-Z0-9]*)_%{UUID:pipe_id}" }
    }

    json {
      source => "pipedream_message"
    }

    date {
      match => [ "logged_time", "ISO8601" ]
    }
  } else if [type] == "ingest_log" {
    json {
      source => "ingest_message"
    }
  } else if [type] == "s3_access_log" {
    grok {
      match => { "message" => "%{S3_ACCESS_LOG}" }
      remove_field => [ "message", "owner", "requester", "http_version" ]
    }
    grok {
      match => { "request" => "%{URIPATH:path}(?:%{URIPARAM:params})?" }
    }
    kv {
      source => "params"
      field_split => "&?"
      include_keys => ["x-zoox-application"]
      target => "application_vars"
    }
    urldecode {
      field => "application_vars"
    }
    kv {
      source => "[application_vars][x-zoox-application]"
      field_split => ";"
      value_split => "="
    }
    if ![application_vars] or ![application_vars][x-zoox-application] {
      drop {}
    }
  }
}

output {
  if [type] == "syslog" {
    elasticsearch {
      hosts => ["localhost:9200"]
      sniffing => true
      manage_template => true
      index => "log-%{+YYYY.MM.dd}"
      document_type => "%{[@metadata][type]}"
    }
  } else if [type] == "pipedream_log" {
    elasticsearch {
      hosts => ["localhost:9200"]
      sniffing => true
      manage_template => false
      index => "pipedream-%{pipeline_name}"
      document_type => "log"
    }
  } else if [type] == "ingest_log" {
    elasticsearch {
      hosts => ["localhost:9200"]
      sniffing => true
      manage_template => false
      index => "ingest-log"
      document_type => "log"
    }
  } else if [type] == "s3_access_log" {
    elasticsearch {
      hosts => ["localhost:9200"]
      sniffing => true
      manage_template => false
      index => "s3_access_metrics"
      document_type => "log"
    }
  }
}
