Support for multi es instances

CHANGELOG.md

@@ -20,6 +20,7 @@
 - Added support for shortening Kibana Discover URLs using Kibana Shorten URL API - [#512](https://github.com/jertel/elastalert2/pull/512) - @JeffAshton
 - Added new alerter `HTTP Post 2` which allow more flexibility to build the body/headers of the request. - [#530](https://github.com/jertel/elastalert2/pull/530) - @lepouletsuisse
 - [Slack] Added new option to include url to jira ticket if it is created in the same pipeline. - [#547](https://github.com/jertel/elastalert2/pull/547) - @hugefarsen
+- Added support for multi ElasticSearch instances. - [#548](https://github.com/jertel/elastalert2/pull/548) - @buratinopy
 
 ## Other changes
 - [Docs] Add exposed metrics documentation - [#498](https://github.com/jertel/elastalert2/pull/498) - @thisisxgp

docs/source/ruletypes.rst

@@ -225,9 +225,24 @@ Required Settings
 es_host
 ^^^^^^^
 
-``es_host``: The hostname of the Elasticsearch cluster the rule will use to query. (Required, string, no default)
+``es_host``: The hostname (or comma separated list of hostnames) of the Elasticsearch cluster the rule will use to query. (Required, string, no default)
 The environment variable ``ES_HOST`` will override this field.
 
+Example of working with one host::
+
+    es_host: hostname
+
+Example of working with multiple hosts::
+
+    hostname0, hostname1, hostname2
+
+Example of working with multiple hosts and specific ports::
+
+    hostname0:9200, hostname1, hostname2:9201
+
+Nodes with a specified port will be used as is, nodes without a port will be used with the port from es_port
+
+
 es_port
 ^^^^^^^
 

elastalert/__init__.py

@@ -16,7 +16,7 @@ def __init__(self, conf):
         """
         :arg conf: es_conn_config dictionary. Ref. :func:`~util.build_es_conn_config`
         """
-        super(ElasticSearchClient, self).__init__(host=conf['es_host'],
+        super(ElasticSearchClient, self).__init__(hosts=conf['es_host'],
                                                   port=conf['es_port'],
                                                   url_prefix=conf['es_url_prefix'],
                                                   use_ssl=conf['use_ssl'],

elastalert/util.py

@@ -361,8 +361,11 @@ def build_es_conn_config(conf):
     parsed_conf['aws_region'] = None
     parsed_conf['profile'] = None
     parsed_conf['headers'] = None
-    parsed_conf['es_host'] = os.environ.get('ES_HOST', conf['es_host'])
-    parsed_conf['es_port'] = int(os.environ.get('ES_PORT', conf['es_port']))
+    es_host = os.environ.get('ES_HOST', conf['es_host'])
+    es_port = int(os.environ.get('ES_PORT', conf['es_port']))
+    parsed_conf['es_host'] = parse_host(es_host, es_port)
+    parsed_conf['es_port'] = es_port
+
     parsed_conf['es_url_prefix'] = ''
     parsed_conf['es_conn_timeout'] = conf.get('es_conn_timeout', 20)
     parsed_conf['send_get_body_as'] = conf.get('es_send_get_body_as', 'GET')
@@ -486,34 +489,34 @@ def should_scrolling_continue(rule_conf):
     return not stop_the_scroll
 
 
-def _expand_string_into_dict(string, value,  sep='.'):
+def _expand_string_into_dict(string, value, sep='.'):
     """
     Converts a encapsulated string-dict to a sequence of dict. Use separator (default '.') to split the string.
-    Example: 
+    Example:
         string1.string2.stringN : value  -> {string1: {string2: {string3: value}}
- 
+
     :param string: The encapsulated "string-dict"
     :param value: Value associated to the last field of the "string-dict"
     :param sep: Separator character. Default: '.'
     :rtype: dict
     """
     if sep not in string:
-        return {string : value}
+        return {string: value}
     key, val = string.split(sep, 1)
     return {key: _expand_string_into_dict(val, value)}
- 
- 
-def expand_string_into_dict(dictionary, string , value, sep='.'):
+
+
+def expand_string_into_dict(dictionary, string, value, sep='.'):
     """
     Useful function to "compile" a string-dict string used in metric and percentage rules into a dictionary sequence.
- 
+
     :param dictionary: The dictionary dict
-    :param string:  String Key 
+    :param string:  String Key
     :param value: String Value
     :param sep: Separator character. Default: '.'
     :rtype: dict
     """
- 
+
     if sep not in string:
         dictionary[string] = value
         return dictionary
@@ -526,7 +529,7 @@ def expand_string_into_dict(dictionary, string , value, sep='.'):
 def format_string(format_config, target_value):
     """
     Formats number, supporting %-format and str.format() syntax.
- 
+
     :param format_config: string format syntax, for example '{:.2%}' or '%.2f'
     :param target_value: number to format
     :rtype: string
@@ -536,3 +539,19 @@ def format_string(format_config, target_value):
     else:
         return format_config % (target_value)
 
+
+def parse_host(host, port=9200):
+    """
+    Convet host str like "host1:port1, host2:port2" to list
+    :param host str: hostnames (separated with comma ) or single host name
+    :param port: default to 9200
+    :return: list of hosts
+
+    """
+    if "," in host:
+        host_list = host.split(",")
+        host_list = [("{host}:{port}".format(host=x.strip(), port=port)
+                      if ":" not in x else x.strip()) for x in host_list]
+        return host_list
+    else:
+        return ["{host}:{port}".format(host=host, port=port)]

tests/util_test.py

@@ -37,6 +37,7 @@
 from elastalert.util import unixms_to_dt
 from elastalert.util import format_string
 from elastalert.util import pretty_ts
+from elastalert.util import parse_host
 
 
 @pytest.mark.parametrize('spec, expected_delta', [
@@ -337,7 +338,7 @@ def test_ts_utc_to_tz():
             'aws_region': None,
             'profile': None,
             'headers': None,
-            'es_host': 'localhost',
+            'es_host': ['localhost:9200'],
             'es_port': 9200,
             'es_url_prefix': '',
             'es_conn_timeout': 20,
@@ -360,7 +361,7 @@ def test_ts_utc_to_tz():
             'aws_region': 'us-east-1',
             'profile': 'default',
             'headers': None,
-            'es_host': 'localhost',
+            'es_host': ['localhost:9200'],
             'es_port': 9200,
             'es_url_prefix': 'elasticsearch',
             'es_conn_timeout': 30,
@@ -435,7 +436,7 @@ def test_build_es_conn_config2():
         'aws_region': None,
         'profile': None,
         'headers': None,
-        'es_host': 'localhost',
+        'es_host': ['localhost:9200'],
         'es_port': 9200,
         'es_url_prefix': '',
         'es_conn_timeout': 20,
@@ -519,3 +520,32 @@ def test_pretty_ts():
     assert '2021-08-16 16:35 UTC' == pretty_ts(ts)
     assert '2021-08-16 16:35 ' == pretty_ts(ts, False)
     assert '2021-08-16 16:35 +0000' == pretty_ts(ts, ts_format='%Y-%m-%d %H:%M %z')
+
+
+def test_parse_host():
+    assert parse_host("localhost", port=9200) == ["localhost:9200"]
+    assert parse_host("host1:9200, host2:9200, host3:9300") == ["host1:9200",
+                                                                "host2:9200",
+                                                                "host3:9300"]
+
+
+def test_build_cofig_for_multi():
+    assert build_es_conn_config({
+        "es_host": "localhost",
+        "es_port": 9200
+    })['es_host'] == ['localhost:9200']
+
+    assert build_es_conn_config({
+        "es_host": "host1:9200, host2:9200, host3:9300",
+        "es_port": 9200
+    })['es_host'] == ["host1:9200", "host2:9200", "host3:9300"]
+
+    assert build_es_conn_config({
+        "es_host": "host1, host2, host3",
+        "es_port": 9200
+    })['es_host'] == ["host1:9200", "host2:9200", "host3:9200"]
+
+    assert build_es_conn_config({
+        "es_host": "host1, host2:9200, host3:9300",
+        "es_port": 9200
+    })['es_host'] == ["host1:9200", "host2:9200", "host3:9300"]