logstash.conf

input {
   file {
      path => "/var/log/log-generator/app.log"
   }
}

filter {

   grok {
      match => [ "message" , "(?<log_date>%{MONTHDAY}-%{MONTHNUM}-%{YEAR} %{HOUR}:%{MINUTE}:%{SECOND}.[0-9]{3}) \[%{NOTSPACE:thread}\] %{LOGLEVEL:log_level} %{NOTSPACE:classname} - %{GREEDYDATA:log_msg}" ]
   }

   date {
      match => ["log_date","dd-MM-YYYY HH:mm:ss.SSS"]
   }

   kv {
      field_split => "&"
      source => "log_msg"
   }

   if [classname] =~ /SellRequest$/ {
      mutate { add_tag => "sell" }
   } else if [classname] =~ /SearchRequest$/ {
      mutate { add_tag => "search" }
   }

   mutate {    
      remove_field => [ "log_msg" ]
   }  

   mutate {
      convert => [ "id", "integer" ]
   }
   mutate {
      convert => [ "price", "float" ]
   }

   mutate {
      split => [ "options", "|" ]
   }

   geoip {
      source => "ip"
   }

   useragent {
      source => "ua"
      target => "useragent"
      remove_field => [ "ua" ]
   }

}

output {
   elasticsearch {}
}