From 9b358c18ea901756341ad2afa282ab0e6265b296 Mon Sep 17 00:00:00 2001 From: test Date: Wed, 15 Jul 2026 12:26:48 +0200 Subject: [PATCH 1/5] feat(new_term): add persist_new_terms option to survive restarts Newly seen terms are written into the existing _past index with deterministic ids and merged back into the baseline at startup, so terms alerted on longer ago than terms_window_size (or whose source documents were deleted) do not re-alert after a restart. Opt-in; persistence failures never interrupt matching. Co-Authored-By: Claude Fable 5 --- elastalert/ruletypes.py | 63 +++++++++++++++++ elastalert/schema.yaml | 1 + tests/rules_test.py | 145 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 209 insertions(+) diff --git a/elastalert/ruletypes.py b/elastalert/ruletypes.py index 43ccf2b1..605342ec 100644 --- a/elastalert/ruletypes.py +++ b/elastalert/ruletypes.py @@ -1,6 +1,8 @@ # -*- coding: utf-8 -*- import copy import datetime +import hashlib +import json import sys from sortedcontainers import SortedKeyList as sortedlist @@ -691,6 +693,9 @@ def __init__(self, rule, args=None): except Exception as e: # Refuse to start if we cannot get existing terms raise EAException('Error searching for existing terms: %s' % (repr(e))).with_traceback(sys.exc_info()[2]) + self.persist_index = self.get_persist_index() + if self.persist_index: + self.load_persisted_terms() def get_all_terms(self, args): """ Performs a terms aggregation for each field to get every existing term. """ @@ -890,6 +895,62 @@ def flatten_aggregation_hierarchy(self, root, hierarchy_tuple=()): results.append(hierarchy_tuple + (node['key'],)) return results + def get_persist_index(self): + """ Returns the index used to persist newly seen terms across restarts, + or None if persist_new_terms is disabled or the index is unavailable. """ + if not self.rules.get('persist_new_terms'): + return None + try: + index = self.rules['writeback_index'] + '_past' + if self.es.indices.exists(index=index): + return index + reason = 'index %s does not exist' % (index) + except Exception as e: + reason = repr(e) + elastalert_logger.warning('persist_new_terms disabled for rule %s: %s' % (self.rules['name'], reason)) + return None + + def persist_term(self, field, value): + """ Writes a newly seen term into the persist index with a deterministic id, + so that it survives restarts. Failures are logged and never interrupt matching. """ + if not self.persist_index: + return + field = list(field) if type(field) is tuple else field + value = list(value) if type(value) is tuple else value + doc_id = hashlib.sha1(json.dumps([self.rules['name'], field, value], default=str).encode('utf-8')).hexdigest() + body = {'rule_name': self.rules['name'], + '@timestamp': dt_to_ts(ts_now()), + 'match_body': {'field': field, 'value': value}} + try: + self.es.index(index=self.persist_index, id=doc_id, body=body) + except Exception as e: + elastalert_logger.warning('Failed to persist new term for rule %s: %s' % (self.rules['name'], repr(e))) + + def load_persisted_terms(self): + """ Merges terms persisted by persist_term into the baseline built by get_all_terms. + On failure the in-memory baseline is used as-is. """ + query = {'query': {'bool': {'filter': [{'term': {'rule_name': self.rules['name']}}]}}, 'size': 10000} + try: + hits = self.es.search(index=self.persist_index, body=query, ignore_unavailable=True)['hits']['hits'] + except Exception as e: + elastalert_logger.warning('Failed to load persisted terms for rule %s: %s' % (self.rules['name'], repr(e))) + return + if len(hits) >= 10000: + elastalert_logger.warning('More than 10000 persisted terms for rule %s, ' + 'restored baseline may be incomplete' % (self.rules['name'])) + loaded = 0 + for hit in hits: + match_body = hit['_source'].get('match_body') or {} + field, value = match_body.get('field'), match_body.get('value') + if not field or not value: + continue + if type(field) is list: + field, value = tuple(field), tuple(value) + if field in self.seen_values and value not in self.seen_values[field]: + self.seen_values[field].append(value) + loaded += 1 + elastalert_logger.info('Loaded %d persisted terms for rule %s' % (loaded, self.rules['name'])) + def add_data(self, data): for document in data: for field in self.fields: @@ -915,6 +976,7 @@ def add_data(self, data): document['new_field'] = lookup_field self.add_match(copy.deepcopy(document)) self.seen_values[lookup_field].append(value) + self.persist_term(lookup_field, value) def add_terms_data(self, terms): # With terms query, len(self.fields) is always 1 and the 0'th entry is always a string @@ -928,6 +990,7 @@ def add_terms_data(self, terms): 'new_field': field} self.add_match(match) self.seen_values[field].append(bucket['key']) + self.persist_term(field, bucket['key']) class CardinalityRule(RuleType): diff --git a/elastalert/schema.yaml b/elastalert/schema.yaml index bb64031e..a739efc8 100644 --- a/elastalert/schema.yaml +++ b/elastalert/schema.yaml @@ -197,6 +197,7 @@ oneOf: terms_size: {type: integer} window_step_size: *timeframe use_keyword_postfix: {type: boolean} + persist_new_terms: {type: boolean} - title: Cardinality required: [cardinality_field, timeframe] diff --git a/tests/rules_test.py b/tests/rules_test.py index 78519854..19afd25b 100644 --- a/tests/rules_test.py +++ b/tests/rules_test.py @@ -810,6 +810,151 @@ def test_new_term_with_composite_fields(): assert rule.matches[1]['missing_field'] == ('d', 'e.f') +def new_term_persist_rules(**extra): + rules = {'fields': ['a'], + 'name': 'test-persist', + 'timestamp_field': '@timestamp', + 'es_host': 'example.com', 'es_port': 10, 'index': 'logstash', + 'writeback_index': 'wb', + 'persist_new_terms': True, + 'ts_to_dt': ts_to_dt, 'dt_to_ts': dt_to_ts} + rules.update(extra) + return rules + + +def new_term_persist_mock_es(mock_es, persisted_hits=None, index_exists=True): + """ Mock ES that answers aggregation queries with a key1 baseline and the + persisted-terms query with the given hits. """ + def search(*args, **kwargs): + if 'aggs' in kwargs.get('body', {}): + return {'aggregations': {'filtered': {'values': {'buckets': [{'key': 'key1', 'doc_count': 1}]}}}} + return {'hits': {'hits': persisted_hits or []}} + + instance = mock.Mock() + instance.search.side_effect = search + instance.info.return_value = {'version': {'number': '8.0.0'}} + instance.indices.exists.return_value = index_exists + mock_es.return_value = instance + return instance + + +def new_term_persist_hit(field, value): + return {'_source': {'rule_name': 'test-persist', 'match_body': {'field': field, 'value': value}}} + + +def test_new_term_persist_init(): + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es) + rule = NewTermsRule(new_term_persist_rules()) + assert rule.persist_index == 'wb_past' + instance.indices.exists.assert_called_once_with(index='wb_past') + + +def test_new_term_persist_off_by_default(): + rules = new_term_persist_rules() + del rules['persist_new_terms'] + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es) + rule = NewTermsRule(rules) + assert rule.persist_index is None + assert instance.indices.exists.call_count == 0 + rule.add_data([{'@timestamp': ts_now(), 'a': 'key2'}]) + assert instance.index.call_count == 0 + + +def test_new_term_persist_missing_index_disables(): + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es, index_exists=False) + rule = NewTermsRule(new_term_persist_rules()) + assert rule.persist_index is None + rule.add_data([{'@timestamp': ts_now(), 'a': 'key2'}]) + assert len(rule.matches) == 1 + assert instance.index.call_count == 0 + + +def test_new_term_persist_writes_new_term(): + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es) + rule = NewTermsRule(new_term_persist_rules()) + rule.add_data([{'@timestamp': ts_now(), 'a': 'key2'}]) + assert len(rule.matches) == 1 + assert instance.index.call_count == 1 + call = instance.index.call_args + assert call[1]['index'] == 'wb_past' + assert call[1]['body']['rule_name'] == 'test-persist' + assert call[1]['body']['match_body'] == {'field': 'a', 'value': 'key2'} + # Deterministic id: same term yields the same document id + doc_id = call[1]['id'] + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance2 = new_term_persist_mock_es(mock_es) + rule2 = NewTermsRule(new_term_persist_rules()) + rule2.add_data([{'@timestamp': ts_now(), 'a': 'key2'}]) + assert instance2.index.call_args[1]['id'] == doc_id + # Already known terms are not persisted again + rule.add_data([{'@timestamp': ts_now(), 'a': 'key1'}]) + assert instance.index.call_count == 1 + + +def test_new_term_persist_load_merges_terms(): + hits = [new_term_persist_hit('a', 'key9'), new_term_persist_hit('stale_field', 'x')] + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es, persisted_hits=hits) + rule = NewTermsRule(new_term_persist_rules()) + # The persisted term is part of the baseline and does not re-alert + assert 'key9' in rule.seen_values['a'] + rule.add_data([{'@timestamp': ts_now(), 'a': 'key9'}]) + assert rule.matches == [] + # Terms of fields no longer configured are ignored + assert 'stale_field' not in rule.seen_values + # The load query targets the persist index, filtered by rule name + load_call = [c for c in instance.search.call_args_list if c[1].get('index') == 'wb_past'] + assert len(load_call) == 1 + assert load_call[0][1]['body']['query']['bool']['filter'] == [{'term': {'rule_name': 'test-persist'}}] + + +def test_new_term_persist_composite_roundtrip(): + hits = [new_term_persist_hit(['a', 'b'], ['key3', 'key4'])] + rules = new_term_persist_rules(fields=[['a', 'b']]) + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es, persisted_hits=hits) + + def search(*args, **kwargs): + if 'aggs' in kwargs.get('body', {}): + return {'aggregations': {'filtered': {'values': {'buckets': [ + {'key': 'key1', 'doc_count': 1, + 'values': {'buckets': [{'key': 'key2', 'doc_count': 1}]}}]}}}} + return {'hits': {'hits': hits}} + instance.search.side_effect = search + rule = NewTermsRule(rules) + # Loaded composite term is reconstructed as a tuple and does not re-alert + assert ('key3', 'key4') in rule.seen_values[('a', 'b')] + rule.add_data([{'@timestamp': ts_now(), 'a': 'key3', 'b': 'key4'}]) + assert rule.matches == [] + # A new composite term is persisted with lists in match_body + rule.add_data([{'@timestamp': ts_now(), 'a': 'key1', 'b': 'other'}]) + assert len(rule.matches) == 1 + assert instance.index.call_args[1]['body']['match_body'] == {'field': ['a', 'b'], 'value': ['key1', 'other']} + + +def test_new_term_persist_es_errors_ignored(): + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es) + + def search(*args, **kwargs): + if 'aggs' in kwargs.get('body', {}): + return {'aggregations': {'filtered': {'values': {'buckets': [{'key': 'key1', 'doc_count': 1}]}}}} + raise Exception('es down') + instance.search.side_effect = search + instance.index.side_effect = Exception('es down') + # Load failure at startup falls back to the aggregation baseline + rule = NewTermsRule(new_term_persist_rules()) + assert rule.seen_values['a'] == ['key1'] + # Write failure neither raises nor loses the match + rule.add_data([{'@timestamp': ts_now(), 'a': 'key2'}]) + assert len(rule.matches) == 1 + assert 'key2' in rule.seen_values['a'] + + def test_flatline(): events = hits(40) rules = { From 513f5e6d6d385ad6aa05b731230f18c9c6304d46 Mon Sep 17 00:00:00 2001 From: test Date: Wed, 15 Jul 2026 12:27:53 +0200 Subject: [PATCH 2/5] docs(new_term): document persist_new_terms option Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 1 + docs/source/ruletypes.rst | 10 ++++++++++ 2 files changed, 11 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf49f467..6df483e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ## New features - Add ES|QL (Elasticsearch Piped Query Language) support - [#1767](https://github.com/jertel/elastalert2/pull/1767) - @jertel +- [New Term] Add `persist_new_terms` option to persist and restore seen terms across restarts via the `_past` writeback index ## Other changes - [Kibana] Fix discover URLs when query_key values are empty - [#1769](https://github.com/jertel/elastalert2/pull/1769) - @thejohnrichard diff --git a/docs/source/ruletypes.rst b/docs/source/ruletypes.rst index 9da5edce..25c440d8 100644 --- a/docs/source/ruletypes.rst +++ b/docs/source/ruletypes.rst @@ -228,6 +228,8 @@ Rule Configuration Cheat Sheet +-------------------------------------------------------+--------+-----------+-----------+--------+-----------+-------+----------+--------+-----------+------------------+-----------------+----------------+ |``alert_on_missing_field`` (boolean, default False) | | | | | | | | Opt | | | | | +-------------------------------------------------------+--------+-----------+-----------+--------+-----------+-------+----------+--------+-----------+------------------+-----------------+----------------+ +|``persist_new_terms`` (boolean, default False) | | | | | | | | Opt | | | | | ++-------------------------------------------------------+--------+-----------+-----------+--------+-----------+-------+----------+--------+-----------+------------------+-----------------+----------------+ |``cardinality_field`` (string, no default) | | | | | | | | | Req | | | | +-------------------------------------------------------+--------+-----------+-----------+--------+-----------+-------+----------+--------+-----------+------------------+-----------------+----------------+ |``max_cardinality`` (boolean, default False) | | | | | | | | | Opt | | | | @@ -1554,6 +1556,14 @@ that if a new term appears but there are at least 50 terms which appear more fre initial query. These are non-analyzed fields added by Logstash. If the field used is analyzed, the initial query will return only the tokenized values, potentially causing false positives. Defaults to true. +``persist_new_terms``: If true, every newly seen term is also written into the ``_past`` index with a +deterministic document id, and merged back into the baseline when the rule starts. This makes the rule restart-safe: terms +that were alerted on longer ago than ``terms_window_size``, or whose source documents have since been deleted by index +lifecycle policies, will not alert again after a restart. The ``*_past`` index is created by ``elastalert-create-index`` for +new installations; for existing installations create it manually with the ``past_elastalert`` mapping (see +``elastalert/es_mappings/``), otherwise a warning is logged and persistence stays disabled. Persistence failures never stop +the rule; it continues with its in-memory baseline. Defaults to false. + Cardinality ~~~~~~~~~~~ From 3304a4529274a14b315d29e5024469012f53220f Mon Sep 17 00:00:00 2001 From: test Date: Wed, 15 Jul 2026 14:20:25 +0200 Subject: [PATCH 3/5] docs(new_term): add PR reference to changelog entry Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6df483e6..14ac5e43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ ## New features - Add ES|QL (Elasticsearch Piped Query Language) support - [#1767](https://github.com/jertel/elastalert2/pull/1767) - @jertel -- [New Term] Add `persist_new_terms` option to persist and restore seen terms across restarts via the `_past` writeback index +- [New Term] Add `persist_new_terms` option to persist and restore seen terms across restarts via the `_past` writeback index - [#1770](https://github.com/jertel/elastalert2/pull/1770) - @MarcellZamboHu ## Other changes - [Kibana] Fix discover URLs when query_key values are empty - [#1769](https://github.com/jertel/elastalert2/pull/1769) - @thejohnrichard From 1a36761e8be34d82e693d80aa32e8bbbe7549bfb Mon Sep 17 00:00:00 2001 From: test Date: Wed, 15 Jul 2026 15:19:33 +0200 Subject: [PATCH 4/5] fix(new_term): avoid bool/filter query shape that trips the eql/esql wrapper The ElasticSearchClient.search wrapper runs every query body through eql.format_request/esql.format_request, which assume query.bool.filter is a dict. A list-valued filter raised AttributeError at startup, disabling term loading. Use a plain term query instead. Found via live ES testing. Co-Authored-By: Claude Fable 5 --- elastalert/ruletypes.py | 2 +- tests/rules_test.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/elastalert/ruletypes.py b/elastalert/ruletypes.py index 605342ec..ce30893f 100644 --- a/elastalert/ruletypes.py +++ b/elastalert/ruletypes.py @@ -929,7 +929,7 @@ def persist_term(self, field, value): def load_persisted_terms(self): """ Merges terms persisted by persist_term into the baseline built by get_all_terms. On failure the in-memory baseline is used as-is. """ - query = {'query': {'bool': {'filter': [{'term': {'rule_name': self.rules['name']}}]}}, 'size': 10000} + query = {'query': {'term': {'rule_name': self.rules['name']}}, 'size': 10000} try: hits = self.es.search(index=self.persist_index, body=query, ignore_unavailable=True)['hits']['hits'] except Exception as e: diff --git a/tests/rules_test.py b/tests/rules_test.py index 19afd25b..bf8672f5 100644 --- a/tests/rules_test.py +++ b/tests/rules_test.py @@ -909,7 +909,7 @@ def test_new_term_persist_load_merges_terms(): # The load query targets the persist index, filtered by rule name load_call = [c for c in instance.search.call_args_list if c[1].get('index') == 'wb_past'] assert len(load_call) == 1 - assert load_call[0][1]['body']['query']['bool']['filter'] == [{'term': {'rule_name': 'test-persist'}}] + assert load_call[0][1]['body']['query'] == {'term': {'rule_name': 'test-persist'}} def test_new_term_persist_composite_roundtrip(): From e1cdda56142261945e8bf389542869a8e677217f Mon Sep 17 00:00:00 2001 From: test Date: Wed, 15 Jul 2026 21:36:34 +0200 Subject: [PATCH 5/5] feat(new_term): scope persisted terms with a persistence_type field The _past index may later hold persisted state for other rule types, so tag every document with persistence_type ('new_terms'), namespace the document id with it, and scope loads to persistence_type + rule_name. The scoped query uses constant_score rather than a top-level bool: the eql/esql request formatters in the ES client assume query.bool.filter is a dict, so a bool there raises AttributeError. A regression test pins the query shape against both formatters. Co-Authored-By: Claude Fable 5 --- elastalert/es_mappings/7/past_elastalert.json | 3 +++ elastalert/es_mappings/8/past_elastalert.json | 3 +++ elastalert/ruletypes.py | 16 ++++++++--- tests/rules_test.py | 27 ++++++++++++++++--- 4 files changed, 41 insertions(+), 8 deletions(-) diff --git a/elastalert/es_mappings/7/past_elastalert.json b/elastalert/es_mappings/7/past_elastalert.json index 0cf2c67d..a57f00dd 100644 --- a/elastalert/es_mappings/7/past_elastalert.json +++ b/elastalert/es_mappings/7/past_elastalert.json @@ -1,5 +1,8 @@ { "properties": { + "persistence_type": { + "type": "keyword" + }, "rule_name": { "type": "keyword" }, diff --git a/elastalert/es_mappings/8/past_elastalert.json b/elastalert/es_mappings/8/past_elastalert.json index be8ef80f..de5e0d3f 100644 --- a/elastalert/es_mappings/8/past_elastalert.json +++ b/elastalert/es_mappings/8/past_elastalert.json @@ -1,5 +1,8 @@ { "properties": { + "persistence_type": { + "type": "keyword" + }, "rule_name": { "type": "keyword" }, diff --git a/elastalert/ruletypes.py b/elastalert/ruletypes.py index ce30893f..8fedd95e 100644 --- a/elastalert/ruletypes.py +++ b/elastalert/ruletypes.py @@ -664,6 +664,9 @@ def garbage_collect(self, ts): class NewTermsRule(RuleType): """ Alerts on a new value in a list of fields. """ + # Identifies this rule type's documents in the shared persistence index + persistence_type = 'new_terms' + def __init__(self, rule, args=None): super(NewTermsRule, self).__init__(rule, args) self.seen_values = {} @@ -917,19 +920,24 @@ def persist_term(self, field, value): return field = list(field) if type(field) is tuple else field value = list(value) if type(value) is tuple else value - doc_id = hashlib.sha1(json.dumps([self.rules['name'], field, value], default=str).encode('utf-8')).hexdigest() - body = {'rule_name': self.rules['name'], + term_hash = hashlib.sha1(json.dumps([self.rules['name'], field, value], default=str).encode('utf-8')).hexdigest() + body = {'persistence_type': self.persistence_type, + 'rule_name': self.rules['name'], '@timestamp': dt_to_ts(ts_now()), 'match_body': {'field': field, 'value': value}} try: - self.es.index(index=self.persist_index, id=doc_id, body=body) + self.es.index(index=self.persist_index, id='%s:%s' % (self.persistence_type, term_hash), body=body) except Exception as e: elastalert_logger.warning('Failed to persist new term for rule %s: %s' % (self.rules['name'], repr(e))) def load_persisted_terms(self): """ Merges terms persisted by persist_term into the baseline built by get_all_terms. On failure the in-memory baseline is used as-is. """ - query = {'query': {'term': {'rule_name': self.rules['name']}}, 'size': 10000} + # constant_score keeps this in filter context; a bool query at the top level of + # 'query' would be misparsed by the eql/esql request formatters in the ES client. + query = {'query': {'constant_score': {'filter': {'bool': {'must': [ + {'term': {'persistence_type': self.persistence_type}}, + {'term': {'rule_name': self.rules['name']}}]}}}}, 'size': 10000} try: hits = self.es.search(index=self.persist_index, body=query, ignore_unavailable=True)['hits']['hits'] except Exception as e: diff --git a/tests/rules_test.py b/tests/rules_test.py index bf8672f5..af747c57 100644 --- a/tests/rules_test.py +++ b/tests/rules_test.py @@ -839,7 +839,8 @@ def search(*args, **kwargs): def new_term_persist_hit(field, value): - return {'_source': {'rule_name': 'test-persist', 'match_body': {'field': field, 'value': value}}} + return {'_source': {'persistence_type': 'new_terms', 'rule_name': 'test-persist', + 'match_body': {'field': field, 'value': value}}} def test_new_term_persist_init(): @@ -881,10 +882,12 @@ def test_new_term_persist_writes_new_term(): assert instance.index.call_count == 1 call = instance.index.call_args assert call[1]['index'] == 'wb_past' + assert call[1]['body']['persistence_type'] == 'new_terms' assert call[1]['body']['rule_name'] == 'test-persist' assert call[1]['body']['match_body'] == {'field': 'a', 'value': 'key2'} - # Deterministic id: same term yields the same document id + # The id is namespaced by persistence type so other rule types cannot collide doc_id = call[1]['id'] + assert doc_id.startswith('new_terms:') with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: instance2 = new_term_persist_mock_es(mock_es) rule2 = NewTermsRule(new_term_persist_rules()) @@ -906,10 +909,26 @@ def test_new_term_persist_load_merges_terms(): assert rule.matches == [] # Terms of fields no longer configured are ignored assert 'stale_field' not in rule.seen_values - # The load query targets the persist index, filtered by rule name + # The load query targets the persist index, scoped to this rule type and rule name, + # in filter context and without a bool at the top level of 'query' (see load_persisted_terms) load_call = [c for c in instance.search.call_args_list if c[1].get('index') == 'wb_past'] assert len(load_call) == 1 - assert load_call[0][1]['body']['query'] == {'term': {'rule_name': 'test-persist'}} + assert load_call[0][1]['body']['query'] == {'constant_score': {'filter': {'bool': {'must': [ + {'term': {'persistence_type': 'new_terms'}}, + {'term': {'rule_name': 'test-persist'}}]}}}} + + +def test_new_term_persist_query_survives_request_formatters(): + # ElasticSearchClient.search runs every body through the eql/esql request formatters, + # which assume query.bool.filter is a dict. A bool at the top level of 'query' would + # raise there and silently disable term loading, so guard the query shape here. + from elastalert import eql, esql + with mock.patch('elastalert.ruletypes.elasticsearch_client') as mock_es: + instance = new_term_persist_mock_es(mock_es) + NewTermsRule(new_term_persist_rules()) + body = [c for c in instance.search.call_args_list if c[1].get('index') == 'wb_past'][0][1]['body'] + assert eql.format_request(body) is None + assert esql.format_request(body) is None def test_new_term_persist_composite_roundtrip():