Merge pull request #1193 from Goggin/feat-fields-query

Add fields option to rules to allow for runtime fields and others

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 ## New features
 - Add initial support for EQL - [#1189](https://github.com/jertel/elastalert2/pull/1189) - @jertel
+- Add `fields` parameter to rules to be able to pull in runtimes fields, and more. [#1193](https://github.com/jertel/elastalert2/pull/1193) - @Goggin
 - Add EQL support to elastalert-test-rule utility - [#1195](https://github.com/jertel/elastalert2/pull/1195) - @jertel
 
 ## Other changes

docs/source/ruletypes.rst

@@ -104,6 +104,8 @@ Rule Configuration Cheat Sheet
 +--------------------------------------------------------------+           |
 | ``include`` (list of strs, default ["*"])                    |           |
 +--------------------------------------------------------------+           |
+| ``fields`` (list of strs, no default)                        |           |
++--------------------------------------------------------------+           |
 | ``filter`` (ES filter DSL, no default)                       |           |
 +--------------------------------------------------------------+           |
 | ``max_query_size`` (int, default global max_query_size)      |           |
@@ -584,6 +586,13 @@ include
 fields, along with '@timestamp', ``query_key``, ``compare_key``, and ``top_count_keys``  are included, if present.
 (Optional, list of strings, default all fields)
 
+fields
+^^^^^^
+
+``fields``: A list of fields that should be included in query results and passed to rule types and alerts. If ``_source_enabled`` is False,
+only these fields and those from ``include`` are included.  When ``_source_enabled`` is True, these are in addition to source.  This is used
+for runtime fields, script fields, etc.  This only works with Elasticsearch version 7.11 and newer.  (Optional, list of strings, no default)
+
 top_count_keys
 ^^^^^^^^^^^^^^
 

elastalert/elastalert.py

@@ -363,6 +363,9 @@ def get_hits(self, rule, starttime, endtime, index, scroll=False):
             query['stored_fields'] = rule['include']
             extra_args = {}
 
+        if rule.get('fields', None) is not None:
+            query['fields'] = rule['fields']
+
         try:
             if scroll:
                 res = self.thread_data.current_es.scroll(scroll_id=rule['scroll_id'], scroll=scroll_keepalive)

elastalert/loaders.py

@@ -347,6 +347,7 @@ def load_options(self, rule, conf, filename, args=None):
         rule.setdefault('description', "")
         rule.setdefault('jinja_root_name', "_data")
         rule.setdefault('query_timezone', "")
+        rule.setdefault('fields', None)
 
         # Set timestamp_type conversion function, used when generating queries and processing hits
         rule['timestamp_type'] = rule['timestamp_type'].strip().lower()
@@ -393,6 +394,9 @@ def _dt_to_ts_with_format(dt):
         if 'include' in rule and type(rule['include']) != list:
             raise EAException('include option must be a list')
 
+        if 'fields' in rule and rule['fields'] is not None and type(rule['fields']) != list:
+            raise EAException('fields option must be a list')
+
         raw_query_key = rule.get('query_key')
         if isinstance(raw_query_key, list):
             if len(raw_query_key) > 1:

elastalert/schema.yaml

@@ -255,6 +255,7 @@ properties:
     items: *filter
 
   include: {type: array, items: {type: string}}
+  fields: {type: array, item: {type: string}}
   top_count_keys: {type: array, items: {type: string}}
   top_count_number: {type: integer}
   raw_count_keys: {type: boolean}

tests/base_test.py

@@ -101,7 +101,7 @@ def test_query(ea):
         size=ea.rules[0]['max_query_size'], scroll=ea.conf['scroll_keepalive'])
 
 
-def test_query_with_fields(ea):
+def test_query_with_stored_fields(ea):
     ea.rules[0]['_source_enabled'] = False
     ea.thread_data.current_es.search.return_value = {'hits': {'total': {'value': 0}, 'hits': []}}
     ea.run_query(ea.rules[0], START, END)
@@ -112,6 +112,18 @@ def test_query_with_fields(ea):
         size=ea.rules[0]['max_query_size'], scroll=ea.conf['scroll_keepalive'])
 
 
+def test_query_with_fields(ea):
+    ea.rules[0]['fields'] = ['test_runtime_field']
+    ea.thread_data.current_es.search.return_value = {'hits': {'total': {'value': 0}, 'hits': []}}
+    ea.run_query(ea.rules[0], START, END)
+    ea.thread_data.current_es.search.assert_called_with(body={
+        'query': {'bool': {
+            'filter': {'bool': {'must': [{'range': {'@timestamp': {'lte': END_TIMESTAMP, 'gt': START_TIMESTAMP}}}]}}}},
+        'sort': [{'@timestamp': {'order': 'asc'}}], 'fields': ['test_runtime_field']},
+        index='idx', ignore_unavailable=True, _source_includes=['@timestamp'],
+        size=ea.rules[0]['max_query_size'], scroll=ea.conf['scroll_keepalive'])
+
+
 def test_query_with_unix(ea):
     ea.rules[0]['timestamp_type'] = 'unix'
     ea.rules[0]['dt_to_ts'] = dt_to_unix

tests/loaders_test.py

@@ -45,7 +45,8 @@
              'use_count_query': True,
              'email': '[email protected]',
              'aggregation': {'hours': 2},
-             'include': ['comparekey', '@timestamp']}
+             'include': ['comparekey', '@timestamp'],
+             'fields': ['test_runtime_field']}
 
 test_args = mock.Mock()
 test_args.config = 'test_config'
@@ -275,6 +276,8 @@ def test_load_rules():
                 assert isinstance(rules['rules'][0]['alert'][0], elastalert.alerts.Alerter)
                 assert isinstance(rules['rules'][0]['timeframe'], datetime.timedelta)
                 assert isinstance(rules['run_every'], datetime.timedelta)
+                assert isinstance(rules['rules'][0]['fields'], list)
+                assert 'test_runtime_field' in rules['rules'][0]['fields']
                 for included_key in ['comparekey', 'testkey', '@timestamp']:
                     assert included_key in rules['rules'][0]['include']
 
@@ -363,7 +366,10 @@ def test_load_disabled_rules():
 
 
 def test_raises_on_missing_config():
-    optional_keys = ('aggregation', 'use_count_query', 'query_key', 'compare_key', 'filter', 'include', 'es_host', 'es_port', 'name')
+    optional_keys = (
+        'aggregation', 'use_count_query', 'query_key', 'compare_key', 'filter', 'include', 'es_host', 'es_port',
+        'name', 'fields'
+    )
     test_rule_copy = copy.deepcopy(test_rule)
     for key in list(test_rule_copy.keys()):
         test_rule_copy = copy.deepcopy(test_rule)