netflow for logstash

gistfile1.txt

input {

lumberjack {
    port => 5043
    type => "squid"
    ssl_certificate => "/home/ec2-user/logstash-forwarder/logstash-forwarder.crt"
    ssl_key => "/home/ec2-user/logstash-forwarder/logstash-forwarder.key"
  }

    udp {
      port => 1234
      codec => netflow {
        definitions => "/home/ec2-user/logstash-1.4.2/lib/logstash/codecs/netflow/netflow.yaml"
        versions => [5]
      }
    }
  }

filter {

 #mutate {
 # rename => [ "[netflow][ipv4_dst_addr]" , "geosrc" ]
 # type => "logs"
 #}

 geoip {
  source => "netflow.ipv_dst_addr"
  type => "logs"
 }

 grok {
  type => "squid"
  pattern => "%{NUMBER:timestamp}\s+%{NUMBER:request_msec:float} %{IPORHOST:src_ip} %{WORD:cache_result}/%{NUMBER:response_status:int} %{NUMBER:response_size:int} %{WORD:http_method} (%{URIPROTO:http_proto}://)?%{IPORHOST:dst_host}(?::%{POSINT:port})?(?:%{URIPATHPARAM:uri_param})? %{USERNAME:cache_user} %{WORD:request_route}/(%{IPORHOST:forwarded_to}|-) %{GREEDYDATA:content_type}"
  add_tag => "squid"
 }

 geoip {
  source => "dst_host"
  type => "squid"
 }

 date {
  tags => "squid"
  match => [ "timestamp", "UNIX" ]
 }
}

output {
    elasticsearch_http {
       index => "logstash_netflow5-%{+YYYY.MM.dd}"
       host => "localhost"
       port => 9200
       type => "logs"
     }

    elasticsearch_http {
       index => "logstash-%{+YYYY.MM.dd}"
       host => "localhost"
       port => 9200
       type => "squid"
     }
}