logstash.conf

input {
  gelf {}
}
filter {
  if [message] {
    grok {
      match => { "message" => "%{IPORHOST:remote_ip} - %{DATA:user_name} \[%{HTTPDATE:access_time}\] \"%{WORD:http_method} %{DATA:url} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:body_sent_bytes} \"%{DATA:referrer}\" \"%{DATA:agent}\"" }
    }
    if "_grokparsefailure" in [tags] {
    }
    else {
      mutate {
          convert => { "bytes" => "integer" }
      }
    }
  }
  geoip {
    source => "remote_ip"
    target => "geoip"
    database => "/var/opt/GeoLite2-City/GeoLite2-City.mmdb"
    fields => ["country_name", "country_code3", "region_name", "city_name", "location"]
  }
}
output {
  elasticsearch {
    hosts => ["elasticsearch"]
  }
  stdout {
    codec => rubydebug
  }
}