Filter with select in logstash

[juuuhuuu (XJ)]

Hi Community,
I would like to have a filter to dissect my logs and aggregate them like this

SELECT COUNT(CustomerID), Country
FROM Customers
GROUP BY Country;

Could you please help me?

=> in the logname are countries for example like: user 123 in Germany logelevel: Info /path...

       filter {
dissect {mapping => {"message" => "[%{logdate}] %{application}.%{loglevel}: %{logname} /%{path}"}}
       
 if "info" in [loglevel] {
        
          aggregate {
               task_id => "%{logname}"
               code => "
                 map['logname'] ||= 0; map['logname'] += 1;
                 map['logname'] ||= event.get('logname')
                 map['loglevel'] ||= []
                 map['loglevel'] << {'loglevel' => event.get('loglevel')}
                 event.cancel()
               "
               push_previous_map_as_event => true
               timeout => 3
             }
         }else{
         dissect { mapping => { "message" => "[%{logdate}] %{exception}.%{errorlogname}.%{errorCode}.%{loglevel}: %{errorMessage} /%{path}" } }
        }

[wangqinghuan (Wang)]

Why don't do it in Elasticsearch?

[Badger]

What does the data you are trying to aggregate look like?

[juuuhuuu (XJ)]

[2020-11-02 12:44:04] application.INFO: Today is beautiful weather /path/....
[2020-11-02 12:44:05] application.INFO: Today is beautiful weather /path/....
[2020-11-02 12:44:06] application.INFO: today rainbow /path/....
[2020-11-02 12:44:06] Chttpexception.ERROR: 34 rainy day /path/....

[juuuhuuu (XJ)]

How

[juuuhuuu (XJ)]

please help

[Badger]

You could try

    dissect {mapping => {"message" => "[%{logdate}] %{application}.%{loglevel}: %{}"}}
    if "INFO" in [loglevel] {
        dissect {mapping => {"message" => "[%{logdate}] %{application}.%{loglevel}: %{logname} /%{path}"}}
        aggregate {
           task_id => "%{logname}"
           code => "
             map['logname'] ||= 0; map['logname'] += 1;
             map['logname'] ||= event.get('logname')
             map['loglevel'] ||= []
             map['loglevel'] << {'loglevel' => event.get('loglevel')}
             event.cancel()
           "
           push_previous_map_as_event => true
           timeout => 3
         }
    }else{
        dissect { mapping => { "message" => "[%{logdate}] %{application}.%{loglevel}: %{errorCode} %{errorMessage} /%{path}" } }
    }

Note that the

map['logname'] ||= event.get('logname')

will never perform the assignment, because the previous line has ensured that map['logname'] is never nil.

[juuuhuuu (XJ)]

thank you very much, Ill try it and let you know !!
I see so ..

[juuuhuuu (XJ)]

Hello,

i tried but the problem is, that the logstash is not going in to the if condition.