Trouble for grab events with aggregate filter

[hummermania (Markevych Alexander)]

Hello experts!

I try to parse log which have field used for task_id, some other lines with same format - event details, and also have end marker for completelly ending event.

For example:

filter {
grok {
match => {
"message" => "...%{..:cps} %{GREEDYDATA:message}"
}
overwrite => [ "message" ]
tag_on_failure => ["failed to parse cps name"]
}
....
}

We have succesfulley parsed field named "cps" and tail string in the "message". It's all work normally.

After that I try to use aggregate:

if [message] =~ /^Pattern1.*$/ {
grok {
match => {
"message" => "^Pattern1(?[ var1].+)$"
}
}
aggregate {
task_id => "%{cps}"
map_action => "create"
code => "...."
}
}

We also succesfully match line beginning from Pattern1 and get contents for var1.

also I check some another pattern:

if [message] =~ /^Pattern1.$/ {
.....
aggregate {
task_id => "%{cps}"
map_action => "create"
code => "...."
}
}
else if [message] =~ /^Pattern2.$/ {
......
aggregate {
task_id => "%{cps}"
map_action => "update"
code => "...."
}
} else if [message] =~ /^Pattern3.*$/ {
aggregate {
task_id => "%{cps}"
map_action => "update"
code => "...."
end_of_task => true
}
} else {
aggregate {
task_id => "%{cps}"
map_action => "update"
code => "...."
}
}

But every line generate separate event, and can't aggregate, check end pattern etc....

Example lines:

cps1 Pattern1 ...... (<-- begin of first event from cps1)
cps1 Pattern2 ......
cps2 Pattern1 .......
cps1 Pattern2 ......
cps1 Pattern3 ...... (<-- env of event from cps1)
cps2 Pattern2 ......
cps1 Pattern1 ...... (<-- must be begin another event for cps1)
cps2 Pattern3 ...... (<-- end of event from cps2)
......

cpsX object is many - 50 - 70 and more in the future, and in the log their number are shuffled.

But now all line genereate divided event, and all config manipulation can't make it work as expected.

That can I done wrong?

[hummermania (Markevych Alexander)]

I find solution.

Main patern - is the end of event. And all previous lines and message values we can compile in some field.

And after event is ended we can parse this new field consist needed text.

...
mutate {
id => "filter.db_error.add_compiled_message"
add_field => { "compiled_message" => "" }
}
.....
if [message] =~ /^END_PATTERN$/ {
aggregate {
id => "filter.db_error.aggr_end_event"
task_id => "%{cps}"
code => "map['compiled_message'] ||= ''; event.set('compiled_message', map['compiled_message'])"
end_of_task => true
timeout => 120
}
mutate {
add_tag =>[ "end_of_event" ]
}

} else {
# compile oneline message with divider ' ||| '
aggregate {
id => "filter.db_error.aggr_all_message"
task_id => "%{cps}"
code => "map['compiled_message'] ||= ''; map['compiled_message'] += event.get('message') + '|||';"
}
# We don't need intermediate lines as event
drop {}
}

if "end_of_event" in [tags] {
.... Parse whole message
}