[New Term] Add persist_new_terms option to survive restarts

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

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 ``<writeback_index>_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
 ~~~~~~~~~~~
 

elastalert/es_mappings/7/past_elastalert.json

@@ -1,5 +1,8 @@
 {
   "properties": {
+    "persistence_type": {
+      "type": "keyword"
+    },
     "rule_name": {
       "type": "keyword"
     },

elastalert/es_mappings/8/past_elastalert.json

@@ -1,5 +1,8 @@
 {
   "properties": {
+    "persistence_type": {
+      "type": "keyword"
+    },
     "rule_name": {
       "type": "keyword"
     },

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):

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]

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 = {