logstash.conf

input {
  udp {
    port => 5300
    codec => 'json'
  }
}
## the policy_router SQL filter
filter {
  if ([type] == "policy_router") {
    grok {
       patterns_dir => ["/etc/logstash/pattern"]
       match => { "message" => "%{ARSQL:sql}" }
       add_field => {"grok" => "true"} 
    }

   if ([grok] == 'true') {
      mutate {
        replace => {"type" => "pr-sql"}
        remove_field => ["grok"]
      }
    }
  }
}


output {
  elasticsearch {
    hosts => ["localhost:9200"]
    sniffing => true
    manage_template => false
    index => "%{[type]}-%{+YYYY.MM.dd}"
    document_type => "%{[type]}"
  }
}

## SQL regex for policy router.
ARSQL ((.+?) \(\d+\.\d+ms\)  (BEGIN|SELECT|INSERT|UPDATE|DELETE|COMMIT|ROLLBACK) .+)