Logstash parser config for nginx logs

logstash-nginx.cnf

input {
    kafka {
        bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
        group_id => "central-logging"
        topics =>  ["nginx-logs"]
    }
}

filter {
    json {
	    source => "message"
    }
  grok {
            match => { 'message' => '%{IPORHOST:proxyIP} \[%{HTTPDATE:ACTIVITY_DATE}\] %{DATA:DOMAIN} %{NUMBER:TIME_TAKEN} \"%{WORD:REQUEST_TYPE} %{DATA:REQUEST_URI} HTTP/%{NUMBER}\" %{NUMBER:RCODE} %{NUMBER:RESPONSE_SIZE} %{NOTSPACE:UPSTREAM_CACHE_STATUS}%{SPACE}%{NOTSPACE:TRANSACTION_ID}%{SPACE}%{NOTSPACE:REQUEST_ID} %{NOTSPACE:XForwardedIP} \"%{DATA:ACTUAL_REFERER}\" \"%{DATA:ACTUAL_UA}\" %{DATA:UPSTREAM} \"%{DATA:scheme}\" \"%{DATA:REQUESTED_WITH}\"%{SPACE}%{GREEDYDATA:ADDITIONAL_INFO}' }
        }
        date {
			match => ["Msec","dd/MMM/yyyy:HH:mm:ss Z","UNIX_MS","UNIX"]
			target => "@timestamp"
      }
      
      // creating summary of URL
        translate {
            field => "REQUEST_URI"
            destination => "URL_SUMMARY"
            exact => true
            regex => true
            override => true
            refresh_interval => 3600
            dictionary_path => '/mnt/logstash/rules/url-translate-rules.json'
        }
     // tranforming to multiple indices depending on domain
        translate {
            field => "DOMAIN"
            destination => "INDEX"
            exact => true
            regex => true
            fallback => "default-index"
            refresh_interval => 3600
            dictionary_path => '/mnt/logstash/rules/domain-index-mapping.json'
        }

        if [ACTUAL_UA] != "-" and [ACTUAL_UA] != "" {
            useragent {
                source => "ACTUAL_UA"
                lru_cache_size => 10000
                remove_field => [ "patch","build", "os_minor", "minor", "os_name"]
            }
            mutate {
                rename => { "name" => "Browser" }
                rename => { "major" => "Browser_Version" }
                rename => { "os_major" => "OS_Version" }
                rename => { "os" => "OS" }
            }
        } 
        // adding geo location information if the origin IP is not a local
        if [XForwardedIP] and [XForwardedIP] != "-" {
            cidr {
                address => [ "%{[XForwardedIP]}" ]
                network => [ "0.0.0.0/32", "10.10.0.0/16", "172.10.0.0/16", "192.168.0.0/16", "fc00::/7", "127.0.0.0/8", "::1/128","169.254.0.0/16", "fe80::/10","224.0.0.0/4", "ff00::/8","255.255.255.255/32" ]
                add_field => { "origin_type" => "local" }
            }
            if [origin_type] != "local" {
                geoip {
                    database => "/mnt/logstash/GeoIP2-City.mmdb"
                    source => "XForwardedIP"
                    target => "origin"
                    cache_size => 10000
                }
            }
        }

        ruby {
            code => "event.set('index_name' , event.get('INDEX')+'-nginx-access-'+event.get('@timestamp').time.localtime.strftime('%Y-%m-%d'))"
            remove_field => [ "INDEX" ]
        }
    }
}

output {
  elasticsearch {
    hosts => ["elastic1", "elastic2", "elastic3"]
    manage_template => false
    index => "%{index_name}"
    document_type => "nginx-logs"
  }

}