Aggregation not working when run on past events

[nicolasnovelli]

Elastalert version: 2.17.0

Hello everyone!
I experienced a potential problem with the aggregation feature.
The problem occurred because a rule remained disabled for a few days and was configured to use aggregation to group events into a single alert.

The aggregation configuration section was the following:

...

aggregation:
  hours: 4

aggregation_key: ["fields.host_guid", "winlog.event_data.SubjectUserName"]
aggregate_by_match_time: True

...

As soon as I restarted the service, which led to the rule reload, it started sending a lot of alerts (~2000 emails in a few hours).

Analyzing the alerts and the service logs, I noticed that the rule was running on past events without applying aggregation.
All alerts generated contained a single event, and the logs clearly showed that a new aggregation was created for each event, even if the aggregation key was the same and the match timestamp was only a few milliseconds apart.

Oct 7, 2024 @ 13:10:55.239 - INFO - elastalert New aggregation for rule_1234, aggregation_key: 188a6875-4e70-4434-8b37-4a694d08fe4d, user_pippo. next alert at 2024-10-05 23:19:57.900000+00:00.
Oct 7, 2024 @ 13:10:55.214 - INFO - elastalert New aggregation for rule_1234, aggregation_key: 188a6875-4e70-4434-8b37-4a694d08fe4d, user_pippo. next alert at 2024-10-05 23:19:57.895000+00:00.
Oct 7, 2024 @ 13:10:55.189 - INFO - elastalert New aggregation for rule_1234, aggregation_key: 188a6875-4e70-4434-8b37-4a694d08fe4d, user_pippo. next alert at 2024-10-05 23:19:57.091000+00:00.
Oct 7, 2024 @ 13:10:55.168 - INFO - elastalert New aggregation for rule_1234, aggregation_key: 188a6875-4e70-4434-8b37-4a694d08fe4d, user_pippo. next alert at 2024-10-05 23:19:57.091000+00:00.

I found the following section in the code that could be the cause of the problem (#L1649):

    def add_aggregated_alert(self, match, rule):
        """ Save a match as a pending aggregate alert to Elasticsearch. """

        ...
        
            # This is a fallback option in case this change to using ts_now() interferes with the behavior current
            # users are accustomed to. It is not documented because it likely won't be needed. If no one reports 
            # a problem we can remove this fallback option in a future release.
            if rule.get('aggregation_alert_time_compared_with_timestamp_field', False):
                compare_dt = lookup_es_key(match, rule['timestamp_field'])
            else:
                compare_dt = ts_now()

            if (not rule['current_aggregate_id'].get(aggregation_key_value) or  
                    ('aggregate_alert_time' in rule and aggregation_key_value in rule['aggregate_alert_time'] and rule[
                        'aggregate_alert_time'].get(aggregation_key_value) < ts_to_dt(compare_dt))):
                # ElastAlert may have restarted while pending alerts exist
                pending_alert = self.find_pending_aggregate_alert(rule, aggregation_key_value)
                ...

From what I understand, the code is always creating a new aggregation for each event, since the condition rule['aggregate_alert_time'].get(aggregation_key_value) < ts_to_dt(compare_dt)) is always false when processing past events in combination with the aggregate_by_match_time configuration set to True.
In the if code block, find_pending_aggregate_alert is called, which returns None because it checks only for alerts with an alert_time in the future.

As mentioned in the comment, the problem might be resolved using the aggregation_alert_time_compared_with_timestamp_field configuration, but I haven't tested it yet in similar conditions.
I wanted to ask if I understood the problem correctly and, if so, to ask to not remove the fallback option in a future releases, or to bind it to aggregate_by_match_time configuration.

I'm not sure whether aggregation is intended to work in such cases, since an alert waiting for aggregation would be sent anyway by the next execution of the method send_pending_alerts (called by the pending alerts job).

Finally, a little off-topic about the aggregation code. In order to identify the problem, I read all the code that handles aggregations and found something that I think might be a bug.

1. In line #L1577, I found a .pop(...) on a dictionary while iterating over it. I'm not 100% sure that this is a problem when iterating using .items(). Probably adding a list( ... ) as a wrapper should fix it.

2. In line #L1667, rule['current_aggregate_id'] is entirely overwritten with the new aggregation data, losing other aggregations.
   I don't think this impacts the correct functioning of the service since, if not found, data of past aggregations would be reloaded anyway.

If you agree on the points I raised, I can open a PR to fix the code.

Thanks for your attention and for the great work you are doing with Elastalert!

[jertel]

Thanks for reviewing the code in depth. I personally don't use aggregation so I don't run into these problems. So it's certainly possible what you described is correct.

I wanted to ask if I understood the problem correctly and, if so, to ask to not remove the fallback option in a future releases, or to bind it to aggregate_by_match_time configuration.

I am reading this to say "please do not remove the aggregation_alert_time_compared_with_timestamp_field config option from future releases." Is that what you meant? If so, that's fine, and a PR that describes this option in the docs would be welcome.

For off-topic bug #1, I don't think this is a bug since the loop "breaks" upon the execution of the pop(). So even though the underlying dictionary just changed, it doesn't impact the loop since it's finished at that point.

For off-topic bug #2, I think I agree with you. I'd have to leave it to you to verify that the functionality works as expected after changing this to add the key to the existing map (and create it if there is no existing map).

[nicolasnovelli]

Thank you for the quick response.

I am reading this to say "please do not remove the aggregation_alert_time_compared_with_timestamp_field config option from future releases." Is that what you meant?

Yes, that's correct. I'll be happy to update the documentation as soon as I can, and I hope to find time to do so in the coming weeks.

Regarding the off-topic point, I somehow missed the break inside the for loop. In light of that, I agree it doesn't impact the loop. As for the second point, I will test it and get back to you.