Store version in .es_version and updated tests

docs/source/recipes/writing_filters.rst

@@ -39,7 +39,7 @@ See http://lucene.apache.org/core/2_9_4/queryparsersyntax.html for more informat
           query: "field: value OR otherfield: othervalue"
     - query:
         query_string:
-          query: "this: that AND these: those"
+           query: "this: that AND these: those"
 
 term
 ****

docs/source/running_elastalert.rst

@@ -6,7 +6,7 @@ Running ElastAlert for the First Time
 Requirements
 ------------
 
-- Elasticsearch 1.* or 2.*
+- Elasticsearch
 - ISO8601 or Unix timestamped data
 - Python 2.6 or 2.7
 - pip, see requirements.txt
@@ -23,6 +23,16 @@ Install the module::
     $ python setup.py install
     $ pip install -r requirements.txt
 
+Depending on the version of Elasticsearch, you may need to manually install the correct version of elasticsearch-py.
+
+Elasticsearch 5.0+::
+
+    $ pip install elasticsearch>=5.0.0
+
+Elasticsearch 2.X::
+
+    $ pip install elasticsearch<3.0.0
+
 Next, open up config.yaml.example. In it, you will find several configuration options. ElastAlert may be run without changing any of these settings.
 
 ``rules_folder`` is where ElastAlert will load rule configuration files from. It will attempt to load every .yaml file in the folder. Without any valid rules, ElastAlert will not start. ElastAlert will also load new rules, stop running missing rules, and restart modified rules as the files in this folder change. For this tutorial, we will use the example_rules folder.

elastalert/elastalert.py

@@ -115,17 +115,20 @@ def __init__(self, args):
         self.replace_dots_in_field_names = self.conf.get('replace_dots_in_field_names', False)
 
         self.writeback_es = elasticsearch_client(self.conf)
-        self.five = self.is_five()
+        self.es_version = self.get_version()
 
         for rule in self.rules:
             self.init_rule(rule)
 
         if self.args.silence:
             self.silence()
 
-    def is_five(self):
+    def get_version(self):
         info = self.writeback_es.info()
-        return info['version']['number'].startswith('5')
+        return info['version']['number']
+
+    def is_five(self):
+        return self.es_version.startswith('5')
 
     @staticmethod
     def get_index(rule, starttime=None, endtime=None):
@@ -176,7 +179,7 @@ def get_terms_query(self, query, size, field):
         query_element = query['query']
         if 'sort' in query_element:
             query_element.pop('sort')
-        if not self.five:
+        if not self.is_five():
             query_element['filtered'].update({'aggs': {'counts': {'terms': {'field': field, 'size': size}}}})
             aggs_query = {'aggs': query_element}
         else:
@@ -253,11 +256,11 @@ def get_hits(self, rule, starttime, endtime, index, scroll=False):
         :param endtime: The latest time to query.
         :return: A list of hits, bounded by rule['max_query_size'].
         """
-        query = self.get_query(rule['filter'], starttime, endtime, timestamp_field=rule['timestamp_field'], to_ts_func=rule['dt_to_ts'], five=self.five)
+        query = self.get_query(rule['filter'], starttime, endtime, timestamp_field=rule['timestamp_field'], to_ts_func=rule['dt_to_ts'], five=self.is_five())
         extra_args = {'_source_include': rule['include']}
         scroll_keepalive = rule.get('scroll_keepalive', self.scroll_keepalive)
         if not rule.get('_source_enabled'):
-            if self.five:
+            if self.is_five():
                 query['stored_fields'] = rule['include']
             else:
                 query['fields'] = rule['include']
@@ -305,7 +308,7 @@ def get_hits_count(self, rule, starttime, endtime, index):
         :param endtime: The latest time to query.
         :return: A dictionary mapping timestamps to number of hits for that time period.
         """
-        query = self.get_query(rule['filter'], starttime, endtime, timestamp_field=rule['timestamp_field'], sort=False, to_ts_func=rule['dt_to_ts'], five=self.five)
+        query = self.get_query(rule['filter'], starttime, endtime, timestamp_field=rule['timestamp_field'], sort=False, to_ts_func=rule['dt_to_ts'], five=self.is_five())
 
         try:
             res = self.current_es.count(index=index, doc_type=rule['doc_type'], body=query, ignore_unavailable=True)
@@ -329,13 +332,13 @@ def get_hits_terms(self, rule, starttime, endtime, index, key, qk=None, size=Non
             if rule.get('raw_count_keys', True) and not rule['query_key'].endswith('.raw'):
                 filter_key = add_raw_postfix(filter_key)
             rule_filter.extend([{'term': {filter_key: qk}}])
-        base_query = self.get_query(rule_filter, starttime, endtime, timestamp_field=rule['timestamp_field'], sort=False, to_ts_func=rule['dt_to_ts'], five=self.five)
+        base_query = self.get_query(rule_filter, starttime, endtime, timestamp_field=rule['timestamp_field'], sort=False, to_ts_func=rule['dt_to_ts'], five=self.is_five())
         if size is None:
             size = rule.get('terms_size', 50)
         query = self.get_terms_query(base_query, size, key)
 
         try:
-            if not self.five:
+            if not self.is_five():
                 res = self.current_es.search(index=index, doc_type=rule['doc_type'], body=query, search_type='count', ignore_unavailable=True)
             else:
                 res = self.current_es.search(index=index, doc_type=rule['doc_type'], body=query, size=0, ignore_unavailable=True)
@@ -349,7 +352,7 @@ def get_hits_terms(self, rule, starttime, endtime, index, key, qk=None, size=Non
 
         if 'aggregations' not in res:
             return {}
-        if not self.five:
+        if not self.is_five():
             buckets = res['aggregations']['filtered']['counts']['buckets']
         else:
             buckets = res['aggregations']['counts']['buckets']
@@ -438,7 +441,7 @@ def get_starttime(self, rule):
         """
         sort = {'sort': {'@timestamp': {'order': 'desc'}}}
         query = {'filter': {'term': {'rule_name': '%s' % (rule['name'])}}}
-        if self.five:
+        if self.is_five():
             query = {'query': {'bool': query}}
         query.update(sort)
 
@@ -657,7 +660,7 @@ def init_rule(self, new_rule, new=True):
             new_rule[prop] = rule[prop]
 
         # In ES5, filters starting with 'query' should have the top wrapper removed
-        if self.five:
+        if self.is_five():
             for es_filter in new_rule.get('filter', []):
                 if es_filter.get('query'):
                     new_filter = es_filter['query']
@@ -1085,7 +1088,7 @@ def find_recent_pending_alerts(self, time_limit):
         time_filter = {'range': {'alert_time': {'from': dt_to_ts(ts_now() - time_limit),
                                                 'to': dt_to_ts(ts_now())}}}
         sort = {'sort': {'alert_time': {'order': 'asc'}}}
-        if self.five:
+        if self.is_five():
             query = {'query': {'bool': {'must': inner_query, 'filter': time_filter}}}
         else:
             query = {'query': inner_query, 'filter': time_filter}
@@ -1313,7 +1316,7 @@ def is_silenced(self, rule_name):
 
         query = {'term': {'rule_name': rule_name}}
         sort = {'sort': {'until': {'order': 'desc'}}}
-        if self.five:
+        if self.is_five():
             query = {'query': query}
         else:
             query = {'filter': query}

tests/base_test.py

@@ -293,7 +293,7 @@ def test_agg(ea):
     call1 = ea.writeback_es.index.call_args_list[0][1]['body']
     call2 = ea.writeback_es.index.call_args_list[1][1]['body']
     call3 = ea.writeback_es.index.call_args_list[2][1]['body']
-    assert call1['match_body'] == {'@timestamp': '2014-09-26T12:34:45'}
+    assert call1['match_body']['@timestamp'] == '2014-09-26T12:34:45'
     assert not call1['alert_sent']
     assert 'aggregate_id' not in call1
     assert call1['alert_time'] == alerttime1
@@ -353,7 +353,7 @@ def test_agg_cron(ea):
     call2 = ea.writeback_es.index.call_args_list[1][1]['body']
     call3 = ea.writeback_es.index.call_args_list[2][1]['body']
 
-    assert call1['match_body'] == {'@timestamp': '2014-09-26T12:34:45'}
+    assert call1['match_body']['@timestamp'] == '2014-09-26T12:34:45'
     assert not call1['alert_sent']
     assert 'aggregate_id' not in call1
     assert call1['alert_time'] == alerttime1
@@ -419,7 +419,7 @@ def test_agg_with_aggregation_key(ea):
     call1 = ea.writeback_es.index.call_args_list[0][1]['body']
     call2 = ea.writeback_es.index.call_args_list[1][1]['body']
     call3 = ea.writeback_es.index.call_args_list[2][1]['body']
-    assert call1['match_body'] == {'@timestamp': '2014-09-26T12:34:45', 'key': 'Key Value 1'}
+    assert call1['match_body']['key'] == 'Key Value 1'
     assert not call1['alert_sent']
     assert 'aggregate_id' not in call1
     assert 'aggregate_key' in call1