input {
  beats {
    port => 5044
    ssl => false
    client_inactivity_timeout => "${LOGSTASH_CLIENT_INACTIVITY_TIMEOUT:60}"
  }
}

filter {
  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 {}
  }
  fingerprint {
    source => [ "aws.s3.object.key", "log.offset", "request_id" ]
    concatenate_sources => false
    target => "[@metadata][fingerprint]"
    key => "s3_access_metrics"
    method => "SHA512"
  }
}

output {
  elasticsearch {
    document_id => "[@metadata][fingerprint]"
    hosts => "${METRICS_ELASTICSEARCH_HOST}"
    sniffing => false
    manage_template => false
    index => "s3_access_metrics-%{+YYYY.MM.dd}"
    document_type => "s3-access-metric"
  }
}
