Logstash Config File for Parsing HTTP Logs

logstash.config

input {
  file {
    path => ["/opt/gen_logs/logs/access.log"]
    type => "apache_access"
  }
}
filter {
    grok {
       match => [
       "message" , "%{COMBINEDAPACHELOG}+%{GREEDYDATA:extra_fields}",
       "message" , "%{COMMONAPACHELOG}+%{GREEDYDATA:extra_fields}"
       ]
       overwrite => [ "message" ]
    }
    mutate {
       convert => ["response", "integer"]
       convert => ["bytes", "integer"]
       convert => ["responsetime", "float"]
       add_field => [ "rounded_timestamp", "%{@timestamp}" ]
       convert => ["rounded_timestamp", "string"]
    }
    geoip {
       source => "clientip"
       target => "geoip"
       add_tag => [ "apache-geoip" ]
    }
    ruby {
      code => "
       timestamp_stripped = event.get('rounded_timestamp')[0..16] + '00.000Z'
       event.set('rounded_timestamp',timestamp_stripped)
      "
    }
    useragent {
       source => "agent"
    }
}
output {
  kafka {
    codec => json
    topic_id => "retail_logs"
    bootstrap_servers => "mapr02.itversity.com:9092,mapr03.itversity.com:9092,mapr04.itversity.com:9092"
  }
}