Fix/data stream delete support

CHANGELOG.md

@@ -7,6 +7,7 @@
 - [Datadog] Add optional `datadog_api_url` parameter to support regional Datadog sites (US3, US5, EU1, AP1, etc.) - [#1754](https://github.com/jertel/elastalert2/pull/1754) - @BillyWeans
 
 ## Other changes
+- Replace `es.delete()` with `delete_by_query()` in aggregation cleanup to support Elasticsearch data streams as the writeback index - [#1755](https://github.com/jertel/elastalert2/pull/1755) - @peterlehot
 - Remove MS Teams Alerter - [#1757](https://github.com/jertel/elastalert2/pull/1757) - @nsano-rururu
 
 # 2.29.0

elastalert/elastalert.py

@@ -1606,7 +1606,7 @@ def send_pending_alerts(self):
 
                     # Delete it from the index
                     try:
-                        self.writeback_es.delete(index=self.writeback_index, id=_id)
+                        self.writeback_es.delete_by_query(index=self.writeback_index, body={'query': {'term': {'_id': _id}}})
                     except ElasticsearchException:  # TODO: Give this a more relevant exception, try:except: is evil.
                         self.handle_error("Failed to delete alert %s at %s" % (_id, alert_time))
 
@@ -1640,7 +1640,7 @@ def get_aggregated_matches(self, _id):
                                            size=self.max_aggregation)
             for match in res['hits']['hits']:
                 matches.append(match['_source'])
-                self.writeback_es.delete(index=self.writeback_index, id=match['_id'])
+                self.writeback_es.delete_by_query(index=self.writeback_index, body={'query': {'term': {'_id': match['_id']}}})
         except (KeyError, ElasticsearchException) as e:
             self.handle_error("Error fetching aggregated matches: %s" % (e), {'id': _id})
         return matches

tests/conftest.py

@@ -66,6 +66,7 @@ def __init__(self, host='es', port=14900):
         self.create = mock.Mock()
         self.index = mock.Mock()
         self.delete = mock.Mock()
+        self.delete_by_query = mock.Mock()
         self.info = mock.Mock(return_value={'status': 200, 'name': 'foo', 'version': {'number': '2.0'}})
         self.ping = mock.Mock(return_value=True)
         self.indices = mock_es_indices_client()