Merge pull request Yelp#2593 from Yelp/fix_497

Allow run_every to be unique per rule

Author: Qmando

elastalert/elastalert.py

@@ -27,10 +27,10 @@
 from elasticsearch.exceptions import TransportError
 
 from . import kibana
-from .kibana_discover import generate_kibana_discover_url
 from .alerts import DebugAlerter
 from .config import load_conf
 from .enhancements import DropMatchException
+from .kibana_discover import generate_kibana_discover_url
 from .ruletypes import FlatlineRule
 from .util import add_raw_postfix
 from .util import cronite_datetime_to_timestamp
@@ -1022,8 +1022,7 @@ def init_rule(self, new_rule, new=True):
                            'processed_hits',
                            'starttime',
                            'minimum_starttime',
-                           'has_run_once',
-                           'run_every']
+                           'has_run_once']
         for prop in copy_properties:
             if prop not in rule:
                 continue
@@ -1467,8 +1466,8 @@ def send_alert(self, matches, rule, alert_time=None, retried=False):
         # Compute top count keys
         if rule.get('top_count_keys'):
             for match in matches:
-                if 'query_key' in rule and rule['query_key'] in match:
-                    qk = match[rule['query_key']]
+                if 'query_key' in rule:
+                    qk = lookup_es_key(match, rule['query_key'])
                 else:
                     qk = None
 

setup.py

@@ -33,7 +33,7 @@
         'boto3>=1.4.4',
         'configparser>=3.5.0',
         'croniter>=0.3.16',
-        'elasticsearch>=7.0.0',
+        'elasticsearch==7.0.0',
         'envparse>=0.2.0',
         'exotel>=0.1.3',
         'jira>=2.0.0',

tests/base_test.py

@@ -75,7 +75,7 @@ def test_init_rule(ea):
     ea.rules[0]['starttime'] = '2014-01-02T00:11:22'
     ea.rules[0]['processed_hits'] = ['abcdefg']
     new_rule = ea.init_rule(new_rule, False)
-    for prop in ['starttime', 'agg_matches', 'current_aggregate_id', 'processed_hits', 'minimum_starttime']:
+    for prop in ['starttime', 'agg_matches', 'current_aggregate_id', 'processed_hits', 'minimum_starttime', 'run_every']:
         assert new_rule[prop] == ea.rules[0][prop]
 
     # Properties are fresh
@@ -84,6 +84,11 @@ def test_init_rule(ea):
     assert 'starttime' not in new_rule
     assert new_rule['processed_hits'] == {}
 
+    # Assert run_every is unique
+    new_rule['run_every'] = datetime.timedelta(seconds=17)
+    new_rule = ea.init_rule(new_rule, True)
+    assert new_rule['run_every'] == datetime.timedelta(seconds=17)
+
 
 def test_query(ea):
     ea.thread_data.current_es.search.return_value = {'hits': {'total': 0, 'hits': []}}
@@ -989,17 +994,20 @@ def test_kibana_dashboard(ea):
 def test_rule_changes(ea):
     ea.rule_hashes = {'rules/rule1.yaml': 'ABC',
                       'rules/rule2.yaml': 'DEF'}
-    ea.rules = [ea.init_rule(rule, True) for rule in [{'rule_file': 'rules/rule1.yaml', 'name': 'rule1', 'filter': []},
-                                                      {'rule_file': 'rules/rule2.yaml', 'name': 'rule2', 'filter': []}]]
+    run_every = datetime.timedelta(seconds=1)
+    ea.rules = [ea.init_rule(rule, True) for rule in [{'rule_file': 'rules/rule1.yaml', 'name': 'rule1', 'filter': [],
+                                                       'run_every': run_every},
+                                                      {'rule_file': 'rules/rule2.yaml', 'name': 'rule2', 'filter': [],
+                                                       'run_every': run_every}]]
     ea.rules[1]['processed_hits'] = ['save me']
     new_hashes = {'rules/rule1.yaml': 'ABC',
                   'rules/rule3.yaml': 'XXX',
                   'rules/rule2.yaml': '!@#$'}
 
     with mock.patch.object(ea.conf['rules_loader'], 'get_hashes') as mock_hashes:
         with mock.patch.object(ea.conf['rules_loader'], 'load_configuration') as mock_load:
-            mock_load.side_effect = [{'filter': [], 'name': 'rule2', 'rule_file': 'rules/rule2.yaml'},
-                                     {'filter': [], 'name': 'rule3', 'rule_file': 'rules/rule3.yaml'}]
+            mock_load.side_effect = [{'filter': [], 'name': 'rule2', 'rule_file': 'rules/rule2.yaml', 'run_every': run_every},
+                                     {'filter': [], 'name': 'rule3', 'rule_file': 'rules/rule3.yaml', 'run_every': run_every}]
             mock_hashes.return_value = new_hashes
             ea.load_rule_changes()
 
@@ -1021,7 +1029,7 @@ def test_rule_changes(ea):
         with mock.patch.object(ea.conf['rules_loader'], 'load_configuration') as mock_load:
             with mock.patch.object(ea, 'send_notification_email') as mock_send:
                 mock_load.return_value = {'filter': [], 'name': 'rule3', 'new': 'stuff',
-                                          'rule_file': 'rules/rule4.yaml'}
+                                          'rule_file': 'rules/rule4.yaml', 'run_every': run_every}
                 mock_hashes.return_value = new_hashes
                 ea.load_rule_changes()
                 mock_send.assert_called_once_with(exception=mock.ANY, rule_file='rules/rule4.yaml')
@@ -1033,7 +1041,8 @@ def test_rule_changes(ea):
     new_hashes.update({'rules/rule4.yaml': 'asdf'})
     with mock.patch.object(ea.conf['rules_loader'], 'get_hashes') as mock_hashes:
         with mock.patch.object(ea.conf['rules_loader'], 'load_configuration') as mock_load:
-            mock_load.return_value = {'filter': [], 'name': 'rule4', 'new': 'stuff', 'is_enabled': False, 'rule_file': 'rules/rule4.yaml'}
+            mock_load.return_value = {'filter': [], 'name': 'rule4', 'new': 'stuff', 'is_enabled': False,
+                                      'rule_file': 'rules/rule4.yaml', 'run_every': run_every}
             mock_hashes.return_value = new_hashes
             ea.load_rule_changes()
     assert len(ea.rules) == 3
@@ -1044,7 +1053,8 @@ def test_rule_changes(ea):
     new_hashes['rules/rule4.yaml'] = 'qwerty'
     with mock.patch.object(ea.conf['rules_loader'], 'get_hashes') as mock_hashes:
         with mock.patch.object(ea.conf['rules_loader'], 'load_configuration') as mock_load:
-            mock_load.return_value = {'filter': [], 'name': 'rule4', 'new': 'stuff', 'rule_file': 'rules/rule4.yaml'}
+            mock_load.return_value = {'filter': [], 'name': 'rule4', 'new': 'stuff', 'rule_file': 'rules/rule4.yaml',
+                                      'run_every': run_every}
             mock_hashes.return_value = new_hashes
             ea.load_rule_changes()
     assert len(ea.rules) == 4
@@ -1053,7 +1063,8 @@ def test_rule_changes(ea):
     new_hashes.pop('rules/rule4.yaml')
     with mock.patch.object(ea.conf['rules_loader'], 'get_hashes') as mock_hashes:
         with mock.patch.object(ea.conf['rules_loader'], 'load_configuration') as mock_load:
-            mock_load.return_value = {'filter': [], 'name': 'rule4', 'new': 'stuff', 'rule_file': 'rules/rule4.yaml'}
+            mock_load.return_value = {'filter': [], 'name': 'rule4', 'new': 'stuff', 'rule_file': 'rules/rule4.yaml',
+                                      'run_every': run_every}
             mock_hashes.return_value = new_hashes
             ea.load_rule_changes()
     ea.scheduler.remove_job.assert_called_with(job_id='rule4')

tests/conftest.py

@@ -202,6 +202,7 @@ def ea_sixsix():
               'index': 'idx',
               'filter': [],
               'include': ['@timestamp'],
+              'run_every': datetime.timedelta(seconds=1),
               'aggregation': datetime.timedelta(0),
               'realert': datetime.timedelta(0),
               'processed_hits': {},