logstash2elastic.conf

logstash2elastic.conf

input 
{
  kafka 
  {
    zk_connect => "10.2.3.240:2181/stream/kafka/de"
    codec => plain
    topic_id => "mmbb"
    consumer_threads => 1
    reset_beginning => true
  }
}
filter { 
  grok {
    match => {message => "^(?<local_ip>[^\s]+?) (?<hostname>[^\s]+?) (?<private_ip>.*?),?\s*(?<forwarded>[^\s]+?) - (?<customdate>\d[^\s]+) \"(?<method>\w+) (?<url_path>[^\?]+)(?:\?(?<query>[^\s]*?)\s+)?HTTP\/(?<http_version>[^\s]+?)\" (?<code>\d+) (-|(?<size>\d+)) (-|(?<response_time>\d+))\s+(?<page_id>[^\s]+?)\s+(-|(?<db>\d+))\s+(-|(?<java>\d+))\s+(-|(?<memcache>\d+))\s+(-|(?<xcache>\d+))\s+(-|(?<ftp>\d+))\s+(-|(?<solr>\d+))\s+(-|(?<redis>\d+))\s+\"-?(?:(http|https):\/\/(?<referer_host>[^\/]+)\/?\??(?<referer_query>[^\s]*?))?\" \"(?<unique_ident>[^\s]+?)\" \"(?<user_agent>.+)\"$" }
  }
  if "_grokparsefailure" in [tags] {
    drop { }
  }
  date {
    match => ["customdate", "yyyyMMddHHmmss"]
  }
  mutate {
  } 
}
output {
  elasticsearch {
    protocol => "http"
    host => "elastic05"
    index => "accesslog"
    index_type => "httplog_v5"
  }
}