diff --git a/CHANGELOG.md b/CHANGELOG.md index bf49f467b..14ac5e437 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 - [#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 diff --git a/docs/source/ruletypes.rst b/docs/source/ruletypes.rst index 9da5edcef..25c440d80 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 ~~~~~~~~~~~ diff --git a/elastalert/es_mappings/7/past_elastalert.json b/elastalert/es_mappings/7/past_elastalert.json index 0cf2c67db..a57f00ddf 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 be8ef80f0..de5e0d3f5 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 43ccf2b1d..8fedd95e6 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 @@ -662,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 = {} @@ -691,6 +696,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 +898,67 @@ 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 + 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='%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. """ + # 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: + 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 +984,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 +998,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 bb64031e9..a739efc83 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 785198547..af747c577 100644 --- a/tests/rules_test.py +++ b/tests/rules_test.py @@ -810,6 +810,170 @@ 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': {'persistence_type': 'new_terms', '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']['persistence_type'] == 'new_terms' + assert call[1]['body']['rule_name'] == 'test-persist' + assert call[1]['body']['match_body'] == {'field': 'a', 'value': 'key2'} + # 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()) + 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, 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'] == {'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(): + 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 = {