Merge pull request #83 from eladamitpxi/statsd_and_multi_imports

Statsd and multi imports

Author: jertel

config.yaml.example

@@ -30,6 +30,12 @@ es_port: 9200
 # Optional URL prefix for Elasticsearch
 #es_url_prefix: elasticsearch
 
+# Optional prefix for statsd metrics
+#statsd_instance_tag: elastalert
+
+# Optional statsd host
+#statsd_host: dogstatsd
+
 # Connect with TLS to Elasticsearch
 #use_ssl: True
 

docs/source/ruletypes.rst

@@ -40,6 +40,10 @@ Rule Configuration Cheat Sheet
 +--------------------------------------------------------------+           |
 | ``es_url_prefix`` (string, no default)                       |           |
 +--------------------------------------------------------------+           |
+| ``statsd_instance_tag`` (string, no default)                 |           |
++--------------------------------------------------------------+           |
+| ``statsd_host`` (string, no default)                         |           |
++--------------------------------------------------------------+           |
 | ``es_send_get_body_as`` (string, default "GET")              |           |
 +--------------------------------------------------------------+           |
 | ``aggregation`` (time, no default)                           |           |
@@ -247,8 +251,8 @@ import
 
 ``import``: If specified includes all the settings from this yaml file. This allows common config options to be shared. Note that imported files that aren't
 complete rules should not have a ``.yml`` or ``.yaml`` suffix so that ElastAlert doesn't treat them as rules. Filters in imported files are merged (ANDed)
-with any filters in the rule. You can only have one import per rule, though the imported file can import another file, recursively. The filename
-can be an absolute path or relative to the rules directory. (Optional, string, no default)
+with any filters in the rule. You can only have one import per rule, though the imported file can import another file or multiple files, recursively.
+The filename can be an absolute path or relative to the rules directory. (Optional, string or array of strings, no default)
 
 use_ssl
 ^^^^^^^
@@ -291,6 +295,17 @@ es_url_prefix
 
 ``es_url_prefix``: URL prefix for the Elasticsearch endpoint. (Optional, string, no default)
 
+statsd_instance_tag
+^^^^^^^^^^^^^^^^^^^
+
+``statsd_instance_tag``: prefix for statsd metrics. (Optional, string, no default)
+
+
+statsd_host
+^^^^^^^^^^^^^
+
+``statsd_host``: statsd host. (Optional, string, no default)
+
 es_send_get_body_as
 ^^^^^^^^^^^^^^^^^^^
 

docs/source/running_elastalert.rst

@@ -69,6 +69,10 @@ 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_instance_tag``: Optional; prefix for statsd metrics.
+
+``statsd_host``: Optional; statsd host.
+
 ``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

@@ -20,7 +20,9 @@
                 '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_INSTANCE_TAG': 'statsd_instance_tag',
+                'STATSD_HOST': 'statsd_host'}
 
 env = Env(ES_USE_SSL=bool)
 

elastalert/elastalert.py

@@ -16,6 +16,8 @@
 from smtplib import SMTP
 from smtplib import SMTPException
 from socket import error
+import statsd
+
 
 import dateutil.tz
 import pytz
@@ -172,6 +174,12 @@ def __init__(self, args):
         self.thread_data.num_dupes = 0
         self.scheduler = BackgroundScheduler()
         self.string_multi_field_name = self.conf.get('string_multi_field_name', False)
+        self.statsd_instance_tag = self.conf.get('statsd_instance_tag', '')
+        self.statsd_host = self.conf.get('statsd_host', '')
+        if self.statsd_host and len(self.statsd_host) > 0:
+            self.statsd = statsd.StatsClient(host=self.statsd_host, port=8125)
+        else:
+            self.statsd = None
         self.add_metadata_alert = self.conf.get('add_metadata_alert', False)
         self.prometheus_port = self.args.prometheus_port
         self.show_disabled_rules = self.conf.get('show_disabled_rules', True)
@@ -1306,6 +1314,25 @@ def handle_rule_execution(self, rule):
                                    " %s alerts sent" % (rule['name'], old_starttime, pretty_ts(endtime, rule.get('use_local_time')),
                                                         self.thread_data.num_hits, self.thread_data.num_dupes, num_matches,
                                                         self.thread_data.alerts_sent))
+            rule_duration = seconds(endtime - rule.get('original_starttime'))
+            elastalert_logger.info("%s range %s" % (rule['name'], rule_duration))
+            if self.statsd:
+                try:
+                    self.statsd.gauge(
+                        'query.hits', self.thread_data.num_hits,
+                        tags={"elastalert_instance": self.statsd_instance_tag, "rule_name": rule['name']})
+                    self.statsd.gauge(
+                        'already_seen.hits', self.thread_data.num_dupes,
+                        tags={"elastalert_instance": self.statsd_instance_tag, "rule_name": rule['name']})
+                    self.statsd.gauge(
+                        'query.matches', num_matches,
+                        tags={"elastalert_instance": self.statsd_instance_tag, "rule_name": rule['name']})
+                    self.statsd.gauge(
+                        'query.alerts_sent', self.thread_data.alerts_sent,
+                        tags={"elastalert_instance": self.statsd_instance_tag, "rule_name": rule['name']})
+                except BaseException as e:
+                    elastalert_logger.error("unable to send metrics:\n%s" % str(e))
+
             self.thread_data.alerts_sent = 0
 
             if next_run < datetime.datetime.utcnow():

elastalert/loaders.py

@@ -201,6 +201,7 @@ def load_yaml(self, filename):
         }
 
         self.import_rules.pop(filename, None)  # clear `filename` dependency
+        files_to_import = []
         while True:
             loaded = self.get_yaml(filename)
 
@@ -211,14 +212,16 @@ def load_yaml(self, filename):
             loaded.update(rule)
             rule = loaded
             if 'import' in rule:
-                # Find the path of the next file.
-                import_filename = self.get_import_rule(rule)
-                # set dependencies
+                # add all of the files to load into the load queue
+                files_to_import += self.get_import_rule(rule)
+                del (rule['import'])  # or we could go on forever!
+            if len(files_to_import) > 0:
+                # set the next file to load
+                next_file_to_import = files_to_import.pop()
                 rules = self.import_rules.get(filename, [])
-                rules.append(import_filename)
+                rules.append(next_file_to_import)
                 self.import_rules[filename] = rules
-                filename = import_filename
-                del (rule['import'])  # or we could go on forever!
+                filename = next_file_to_import
             else:
                 break
 
@@ -545,10 +548,16 @@ def get_import_rule(self, rule):
         :return: Path the import rule
         :rtype: str
         """
-        if os.path.isabs(rule['import']):
-            return rule['import']
-        else:
-            return os.path.join(os.path.dirname(rule['rule_file']), rule['import'])
+        rule_imports = rule['import']
+        if type(rule_imports) is str:
+            rule_imports = [rule_imports]
+        expanded_imports = []
+        for rule_import in rule_imports:
+            if os.path.isabs(rule_import):
+                expanded_imports.append(rule_import)
+            else:
+                expanded_imports.append(os.path.join(os.path.dirname(rule['rule_file']), rule_import))
+        return expanded_imports
 
     def get_rule_file_hash(self, rule_file):
         rule_file_hash = ''

elastalert/schema.yaml

@@ -183,7 +183,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

requirements.txt

@@ -21,5 +21,6 @@ PyYAML>=5.1
 requests>=2.10.0
 stomp.py>=4.1.17
 texttable>=0.8.8
+statsd-tags==3.2.1.post1
 twilio>=6.0.0,<6.1
 tzlocal<3.0

setup.py

@@ -49,6 +49,7 @@
         'texttable>=0.8.8',
         'twilio>=6.0.0,<6.1',
         'cffi>=1.11.5',
+        'statsd-tags==3.2.1.post1',
         'tzlocal<3.0'
     ]
 )