3_filter.app.logstash.conf

filter {

    # apply filters only on 'app-log'
    if [type] == "app-log" {

        # parse app-log line
        grok {
          patterns_dir => "/etc/logstash/patterns"
          match => ["message","%{LOG_DATE:log_date} \[%{NOTSPACE:thread}\] %{LOGLEVEL:log_level} %{NOTSPACE:classname} - %{GREEDYDATA:msg}"]
        }

        # use log_date field as @timestamp
        date {
          match => ["log_date","dd-MM-YYYY HH:mm:ss.SSS"]
        }

        # split msg field with '&' separator
        kv {
          field_split => "&"
          source => "msg"
        }

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

        # remove msg field
        mutate {
          remove_field => [ "msg" ]
        }

        # convert id as integer
        mutate {
          convert => [ "id", "integer" ]
        }

        # convert price as float
        mutate {
          convert => [ "price", "float" ]
        }

        # transform options as an array
        mutate {
          split => [ "options", "|" ]
        }

        # geoloc ip with GeoLiteCity database
        geoip {
          source => "ip"
          #database => "./GeoLiteCity.dat"
        }

        # parse user agent
        useragent {
          source => "ua"
          target => "useragent"
        }

    }

}