Dev multiple imports statsd8

.travis.yml

@@ -5,7 +5,7 @@ env:
 - TOXENV=docs
 - TOXENV=py27
 install:
-- pip install urllib3 tox
+- pip install urllib3 tox statsd statsd-tags
 script: make test
 deploy:
   provider: pypi

Dockerfile-test

@@ -1,7 +1,7 @@
 FROM ubuntu:latest
 
 RUN apt-get update && apt-get upgrade -y
-RUN apt-get -y install build-essential python-setuptools python2.7 python2.7-dev libssl-dev git tox urllib3
+RUN apt-get -y install build-essential python-setuptools python2.7 python2.7-dev libssl-dev git tox
 
 RUN easy_install pip
 

config.yaml.example

@@ -30,6 +30,9 @@ es_port: 9200
 # Optional URL prefix for Elasticsearch
 #es_url_prefix: elasticsearch
 
+# Optional prefix for statsd metrics
+#statsd_metrics_prefix: cops
+
 # Connect with TLS to Elasticsearch
 #use_ssl: True
 

docs/source/ruletypes.rst

@@ -40,6 +40,8 @@ Rule Configuration Cheat Sheet
 +--------------------------------------------------------------+           |
 | ``es_url_prefix`` (string, no default)                       |           |
 +--------------------------------------------------------------+           |
+| ``statsd_metrics_prefix`` (string, no default)               |           |
++--------------------------------------------------------------+           |
 | ``es_send_get_body_as`` (string, default "GET")              |           |
 +--------------------------------------------------------------+           |
 | ``aggregation`` (time, no default)                           |           |
@@ -271,6 +273,11 @@ es_url_prefix
 
 ``es_url_prefix``: URL prefix for the Elasticsearch endpoint. (Optional, string, no default)
 
+statsd_metrics_prefix
+^^^^^^^^^^^^^
+
+``statsd_metrics_prefix``: prefix for statsd metrics. (Optional, string, no default)
+
 es_send_get_body_as
 ^^^^^^^^^^^^^^^^^^^
 

docs/source/running_elastalert.rst

@@ -66,6 +66,8 @@ Next, open up config.yaml.example. In it, you will find several configuration op
 
 ``es_url_prefix``: Optional; URL prefix for the Elasticsearch endpoint.
 
+``statsd_metrics_prefix``: Optional; prefix for statsd metrics.
+
 ``es_send_get_body_as``: Optional; Method for querying Elasticsearch - ``GET``, ``POST`` or ``source``. The default is ``GET``
 
 ``writeback_index`` is the name of the index in which ElastAlert will store data. We will create this index later.

elastalert/config.py

@@ -38,7 +38,8 @@
                 'ES_USERNAME': 'es_username',
                 'ES_HOST': 'es_host',
                 'ES_PORT': 'es_port',
-                'ES_URL_PREFIX': 'es_url_prefix'}
+                'ES_URL_PREFIX': 'es_url_prefix',
+                'STATSD_METRICS_PREFIX': 'statsd_metrics_prefix'}
 
 env = Env(ES_USE_SSL=bool)
 
@@ -120,39 +121,49 @@ def load_configuration(filename, conf, args=None):
 
 
 def load_rule_yaml(filename):
-    rule = {
-        'rule_file': filename,
-    }
+    return add_rule_yaml(filename, {'rule_file': filename})
 
-    import_rules.pop(filename, None)  # clear `filename` dependency
-    while True:
-        try:
-            loaded = yaml_loader(filename)
-        except yaml.scanner.ScannerError as e:
-            raise EAException('Could not parse file %s: %s' % (filename, e))
-
-        # Special case for merging filters - if both files specify a filter merge (AND) them
-        if 'filter' in rule and 'filter' in loaded:
-            rule['filter'] = loaded['filter'] + rule['filter']
-
-        loaded.update(rule)
-        rule = loaded
-        if 'import' in rule:
-            # Find the path of the next file.
-            if os.path.isabs(rule['import']):
-                import_filename = rule['import']
-            else:
-                import_filename = os.path.join(os.path.dirname(filename), rule['import'])
-            # set dependencies
-            rules = import_rules.get(filename, [])
-            rules.append(import_filename)
-            import_rules[filename] = rules
-            filename = import_filename
-            del(rule['import'])  # or we could go on forever!
-        else:
-            break
 
-    return rule
+def add_rule_yaml(filename, parent_rule):
+    try:
+        loaded_rule = yaml_loader(filename)
+    except yaml.scanner.ScannerError as e:
+        raise EAException('Could not parse file %s: %s' % (filename, e))
+
+    if 'import' in loaded_rule:
+        current_import = loaded_rule['import']
+        del(loaded_rule['import'])  # or we could go on forever!
+
+        child_rules = {}
+        if isinstance(current_import, basestring):
+            child_rules = add_rule_yaml(rule_file_import_path_to_absolute_path(filename, current_import), child_rules)
+        elif isinstance(current_import, list):
+            for import_target in current_import:
+                child_rules = add_rule_yaml(rule_file_import_path_to_absolute_path(filename, import_target), child_rules)
+        loaded_rule = merge_rules(loaded_rule, child_rules)
+
+    loaded_rule = merge_rules(parent_rule, loaded_rule)
+    return loaded_rule
+
+
+def merge_rules(parent_rule, child_rule):
+    # Special case for merging filters - if both files specify a filter merge (AND) them
+    merged_filter = None
+    if 'filter' in parent_rule and 'filter' in child_rule:
+        merged_filter = child_rule['filter'] + parent_rule['filter']
+
+    child_rule.update(parent_rule)
+    if merged_filter is not None:
+        child_rule['filter'] = merged_filter
+
+    return child_rule
+
+
+def rule_file_import_path_to_absolute_path(currently_parsed_file_path, import_file_path):
+    if os.path.isabs(import_file_path):
+        return import_file_path
+    else:
+        return os.path.join(os.path.dirname(currently_parsed_file_path), import_file_path)
 
 
 def load_options(rule, conf, filename, args=None):

elastalert/elastalert.py

@@ -10,10 +10,11 @@
 import time
 import timeit
 import traceback
+import socket
+import statsd
 from email.mime.text import MIMEText
 from smtplib import SMTP
 from smtplib import SMTPException
-from socket import error
 
 import dateutil.tz
 import kibana
@@ -152,6 +153,11 @@ def __init__(self, args):
         self.disabled_rules = []
         self.replace_dots_in_field_names = self.conf.get('replace_dots_in_field_names', False)
         self.string_multi_field_name = self.conf.get('string_multi_field_name', False)
+        self.statsd_prefix = os.environ.get('statsd_metrics_prefix', '')
+        #self.statsd_prefix = socket.gethostname()
+        self.statsd = statsd.StatsClient(host='statsd_exporter',
+                        port=8125,
+                        prefix=self.statsd_prefix)
 
         self.writeback_es = elasticsearch_client(self.conf)
         self._es_version = None
@@ -1129,6 +1135,14 @@ def run_all_rules(self):
                 elastalert_logger.info("Ran %s from %s to %s: %s query hits (%s already seen), %s matches,"
                                        " %s alerts sent" % (rule['name'], old_starttime, pretty_ts(endtime, rule.get('use_local_time')),
                                                             total_hits, self.num_dupes, num_matches, self.alerts_sent))
+                rule_duration = seconds(endtime - rule.get('original_starttime'))
+                elastalert_logger.info("%s range %s" % (rule['name'], rule_duration))
+
+                self.statsd.gauge('query.hits', total_hits, tags={"rule_name": rule['name']})
+                self.statsd.gauge('already_seen.hits', self.num_dupes,tags={"rule_name": rule['name']})
+                self.statsd.gauge('query.matches', num_matches, tags={"rule_name": rule['name']})
+                self.statsd.gauge('query.alerts_sent', self.alerts_sent, tags={"rule_name": rule['name']})
+
                 self.alerts_sent = 0
 
                 if next_run < datetime.datetime.utcnow():

elastalert/schema.yaml

@@ -144,7 +144,12 @@ properties:
   use_strftime_index: {type: boolean}
 
   # Optional Settings
-  import: {type: string}
+  import:
+    anyOf:
+      - type: array
+        items:
+          type: string
+      - type: string
   aggregation: *timeframe
   realert: *timeframe
   exponential_realert: *timeframe

src/pip-delete-this-directory.txt

@@ -0,0 +1,5 @@
+This file is placed here by pip to indicate the source was put
+here by pip.
+
+Once this package is successfully installed this source code will be
+deleted (unless you remove this file).

tests/config_test.py

@@ -97,6 +97,32 @@ def test_import_import():
         assert import_rules == {'blah.yaml': ['importme.ymlt']}
 
 
+def test_multi_imports():
+    import_rule = copy.deepcopy(test_rule)
+    del(import_rule['es_host'])
+    del(import_rule['es_port'])
+    import_rule['import'] = [
+        'import_me_1.ymlt',
+        'import_me_2.ymlt',
+    ]
+    import_me_1 = {
+        'es_host': 'imported_host',
+    }
+    import_me_2 = {
+        'es_port': 12349,
+    }
+
+    with mock.patch('elastalert.config.yaml_loader') as mock_open:
+        mock_open.side_effect = [import_rule, import_me_1, import_me_2]
+        rules = load_configuration('blah.yaml', test_config)
+        assert mock_open.call_args_list[0][0] == ('blah.yaml',)
+        assert mock_open.call_args_list[1][0] == ('import_me_1.ymlt',)
+        assert mock_open.call_args_list[2][0] == ('import_me_2.ymlt',)
+        assert len(mock_open.call_args_list) == 3
+        assert rules['es_port'] == 12349
+        assert rules['es_host'] == 'imported_host'
+
+
 def test_import_absolute_import():
     import_rule = copy.deepcopy(test_rule)
     del(import_rule['es_host'])