input {
  beats {
    port => 5044
  }
}
filter {
    if "beats_input_codec_plain_applied" in [tags] {
        mutate {
               remove_tag => ["beats_input_codec_plain_applied"]
        }
    }
    if [fileset][name] == "access" {

      grok {
        match => {
            "message" => ['^\"%{HTTPDATE:[nginx][access][time]}\" client=%{IPORHOST:[nginx][access][remote_ip]} method=%{WORD:[nginx][access][method]} request=\"%{WORD:verb} %{URIPATHPARAM:[nginx][access][request_path]} HTTP/%{NUMBER:[nginx][access][http_version]}\" request_length=(?:%{NUMBER:[nginx][access][request_length]}|-) status=%{NUMBER:[nginx][access][response_code]} bytes_sent=(?:%{NUMBER:[nginx][access][bytes_sent]}|-) body_bytes_sent=%{NUMBER:[nginx][access][body_sent][bytes]} referer=((?:%{URI:referrer}|-)|%{QS:referrer}) user_agent=%{QS:[nginx][access][agent]} upstream_addr=%{GREEDYDATA:[nginx][access][upstream_addr]} upstream_status=(%{NUMBER:[upstream][access][status_code]}|-) request_time=(?:%{NUMBER:[nginx][access][request_time]}|-) ssl_session_reused=(?:%{WORD:ssl_session_reused}|\.|-) upstream_response_time=(?:%{NUMBER:[upstream][access][response_time]}|-) upstream_connect_time=(?:%{NUMBER:[upstream][access][connect_time]}|-) upstream_header_time=(?:%{NUMBER:[upstream][access][header_time]}|-)', '^\"%{HTTPDATE:[nginx][access][time]}\" client=%{IPORHOST:[nginx][access][remote_ip]} method=- request=\"%{WORD:[nginx][access][method]} %{URIHOST:[nginx][access][request_path]} HTTP/%{NUMBER:[nginx][access][http_version]}\" request_length=(?:%{NUMBER:[nginx][access][request_length]}|-) status=%{NUMBER:[nginx][access][response_code]} bytes_sent=(?:%{NUMBER:[nginx][access][bytes_sent]}|-) body_bytes_sent=%{NUMBER:[nginx][access][body_sent][bytes]} referer=((?:%{URI:referrer}|-)|%{QS:referrer}) user_agent=%{QS:[nginx][access][agent]} upstream_addr=%{GREEDYDATA:[nginx][access][upstream_addr]} upstream_status=(%{NUMBER:[upstream][access][status_code]}|-) request_time=(?:%{NUMBER:[nginx][access][request_time]}|-) ssl_session_reused=(?:%{WORD:ssl_session_reused}|\.|-) upstream_response_time=(?:%{NUMBER:[upstream][access][response_time]}|-) upstream_connect_time=(?:%{NUMBER:[upstream][access][connect_time]}|-) upstream_header_time=(?:%{NUMBER:[upstream][access][header_time]}|-)']
        }
        remove_field => "message"
      }
      if [nginx][access][status_code] != ''{
        mutate {
            convert => {
                "[nginx][access][status_code]" => "integer"
            }
        }
      }


      if [nginx][access][bytes_sent] != '' {
        mutate {
        convert => {
                "[nginx][access][bytes_sent]" => "integer"
            }
        }
      }
      if [nginx][access][body_sent][bytes] != '' {
        mutate {
            convert => {
                "[nginx][access][body_sent][bytes]" => "integer"
            }
        }
      }
      if [upstream][access][status_code] != '' {
        mutate {
            convert => {
                "[upstream][access][status_code]" => "integer"
            }
        }
      }
      if [nginx][access][request_length] != '' {
        mutate {
            convert => {
                "[nginx][access][request_length]" => "integer"
            }
        }
      }
      if [nginx][access][request_time] != '' {
        mutate {
            convert => {
                "[nginx][access][request_time]" => "float"
            }
        }
      }
      if [upstream][access][response_time] != '' {
        mutate {
            convert => {
                "[upstream][access][response_time]" => "float"
            }
        }
      }
      if [upstream][access][connect_time] != '' {
        mutate {
            convert => {
                "[upstream][access][connect_time]" => "float"
            }
        }
      }
      if [upstream][access][header_time] != '' {
        mutate {
            convert => {
                "[upstream][access][header_time]" => "float"
            }
        }
      }
      mutate {
        add_field => { "read_timestamp" => "%{@timestamp}" }
        convert => {
        "[nginx][access][http_version]" => "float"
        }

      }
      date {
        match => [ "[nginx][access][time]", "dd/MMM/YYYY:H:m:s Z" ]
        remove_field => "[nginx][access][time]"
      }
      useragent {
        source => "[nginx][access][agent]"
        target => "[nginx][access][user_agent]"
        remove_field => "[nginx][access][agent]"
      }
      geoip {
        source => "[nginx][access][remote_ip]"
        target => "[nginx][access][geoip]"
      }
    }
    else if [fileset][name] == "error" {
      grok {
        match => { "message" => ["%{DATA:[nginx][error][time]} \[%{DATA:[nginx][error][level]}\] %{NUMBER:[nginx][error][pid]}#%{NUMBER:[nginx][error][tid]}: (\*%{NUMBER:[nginx][error][connection_id]} )?%{GREEDYDATA:[nginx][error][message]}"] }
        remove_field => "message"
      }
      mutate {
        rename => { "@timestamp" => "read_timestamp" }
      }
      date {
        match => [ "[nginx][error][time]", "YYYY/MM/dd H:m:s" ]
        remove_field => "[nginx][error][time]"
      }
    }
}
output {
    if "_grokparsefailure" in [tags] {
        file { path => "/tmp/grok_failures.txt" }
    }
    else {
      elasticsearch {
        hosts => "https://UNIQUE_HOST.eu-west-1.aws.found.io:9243"
        user => "elastic"
        password => "password"
        index => "filebeat-2020-07-07-15-01"
    }
  }
  # stdout { codec => rubydebug }
}