jertel · jertel · Nov 14, 2021 · Nov 7, 2021 · Nov 8, 2021 · Nov 8, 2021
diff --git a/CHANGELOG.md b/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

diff --git a/docs/source/elastalert.rst b/docs/source/elastalert.rst
@@ -127,9 +127,16 @@ is set to true. Note that back filled data may not always trigger count based al
 When ElastAlert 2 is started, it will query for information about the time that it was last run. This way,
 even if ElastAlert 2 is stopped and restarted, it will never miss data or look at the same events twice. It will also specify the default cluster for each rule to run on.
 The environment variable ``ES_HOST`` will override this field.
+For multiple host Elasticsearch clusters see ``es_hosts`` parameter.
 
 ``es_port``: The port corresponding to ``es_host``. The environment variable ``ES_PORT`` will override this field.
 
+``es_hosts`` is the list of addresses of the nodes of the Elasticsearch cluster. This
+parameter can be used for high availability purposes, but the primary host must also
+be specified in the ``es_host`` parameter. The ``es_hosts`` parameter can be overridden 
+within each rule. This value can be specified as ``host:port`` if overriding the default port.
+The environment variable ``ES_HOSTS`` will override this field, and can be specified as a comma-separated value to denote multiple hosts.
+
 ``use_ssl``: Optional; whether or not to connect to ``es_host`` using TLS; set to ``True`` or ``False``.
 The environment variable ``ES_USE_SSL`` will override this field.
 

diff --git a/docs/source/recipes/faq-md.md b/docs/source/recipes/faq-md.md
@@ -374,13 +374,15 @@ alert_text: |
 Does the alert notification destination support Alertmanager?
 ==========
 
-Not supported.
+Now supported as of ElastAlert 2.2.3.
 
 The es_host parameter seems to use only one host. Is it possible to specify multiple nodes?
 ==========
 
-Only one can be set in es_host.
-Please use haproxy in front of elasticsearch to support multiple hosts.
+There are two options:
+
+1. Use haproxy in front of elasticsearch to support multiple hosts.
+2. Use the new ``es_hosts`` parameter introduced in ElastAlert 2.2.3. See :ref:`Configuration <configuration>`.
 
 Is there any plan to implement a REST API into this project?
 ==========

diff --git a/docs/source/ruletypes.rst b/docs/source/ruletypes.rst
@@ -26,6 +26,8 @@ Rule Configuration Cheat Sheet
 +--------------------------------------------------------------+           |
 | ``alert`` (string or list)                                   |           |
 +--------------------------------------------------------------+-----------+
+| ``es_hosts`` (list, no default)                              |           |
++--------------------------------------------------------------+           |
 | ``name`` (string, defaults to the filename)                  |           |
 +--------------------------------------------------------------+           |
 | ``use_strftime_index`` (boolean, default False)              |  Optional |
@@ -227,6 +229,7 @@ es_host
 
 ``es_host``: The hostname of the Elasticsearch cluster the rule will use to query. (Required, string, no default)
 The environment variable ``ES_HOST`` will override this field.
+For multiple host Elasticsearch clusters see ``es_hosts`` parameter.
 
 es_port
 ^^^^^^^
@@ -261,6 +264,11 @@ or loaded from a module. For loading from a module, the alert should be specifie
 
 Optional Settings
 ~~~~~~~~~~~~~~~~~
+es_hosts
+^^^^^^^^
+
+``es_hosts``: The list of nodes of the Elasticsearch cluster that the rule will use for the request. (Optional, list, default none). Values can be specified as ``host:port`` if overriding the default port.
+The environment variable ``ES_HOSTS`` will override this field, and can be specified as a comma-separated value. Note that the ``es_host`` parameter must still be specified in order to identify a primary Elasticsearch host. 
 
 import
 ^^^^^^

diff --git a/docs/source/running_elastalert.rst b/docs/source/running_elastalert.rst
@@ -165,12 +165,19 @@ For this tutorial, we will use the ``examples/rules`` folder.
 time each query is run. This value is ignored for rules where
 ``use_count_query`` or ``use_terms_query`` is set to true.
 
-``es_host`` is the address of an Elasticsearch cluster where ElastAlert 2 will
+``es_host`` is the primary address of an Elasticsearch cluster where ElastAlert 2 will
 store data about its state, queries run, alerts, and errors. Each rule may also
-use a different Elasticsearch host to query against.
+use a different Elasticsearch host to query against. For multiple host Elasticsearch 
+clusters see ``es_hosts`` parameter.
 
 ``es_port`` is the port corresponding to ``es_host``.
 
+``es_hosts`` is the list of addresses of the nodes of the Elasticsearch cluster. This
+parameter can be used for high availability purposes, but the primary host must also
+be specified in the ``es_host`` parameter. The ``es_hosts`` parameter can be overridden 
+within each rule. This value can be specified as ``host:port`` if overriding the default 
+port.
+
 ``use_ssl``: Optional; whether or not to connect to ``es_host`` using TLS; set
 to ``True`` or ``False``.
 

diff --git a/elastalert/__init__.py b/elastalert/__init__.py
@@ -16,7 +16,8 @@ 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__(host=conf.get('es_host'),
+                                                  hosts=conf.get('es_hosts'),
                                                   port=conf['es_port'],
                                                   url_prefix=conf['es_url_prefix'],
                                                   use_ssl=conf['use_ssl'],

diff --git a/elastalert/config.py b/elastalert/config.py
@@ -21,14 +21,14 @@
                 'ES_USERNAME': 'es_username',
                 'ES_API_KEY': 'es_api_key',
                 'ES_HOST': 'es_host',
+                'ES_HOSTS': 'es_hosts',
                 'ES_PORT': 'es_port',
                 'ES_URL_PREFIX': 'es_url_prefix',
                 'STATSD_INSTANCE_TAG': 'statsd_instance_tag',
                 'STATSD_HOST': 'statsd_host'}
 
 env = Env(ES_USE_SSL=bool)
 
-
 # Used to map the names of rule loaders to their classes
 loader_mapping = {
     'file': loaders.FileRulesLoader,

diff --git a/elastalert/schema.yaml b/elastalert/schema.yaml
@@ -200,6 +200,7 @@ properties:
   use_strftime_index: {type: boolean}
 
   # Optional Settings
+  es_hosts: {type: array, items: {type: string}}
   import:
     anyOf:
       - type: array

diff --git a/elastalert/util.py b/elastalert/util.py
@@ -363,6 +363,11 @@ def build_es_conn_config(conf):
     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_hosts = os.environ.get('ES_HOSTS')
+    es_hosts = parse_hosts(es_hosts, parsed_conf.get('es_port')) if es_hosts else conf.get('es_hosts')
+    parsed_conf['es_hosts'] = es_hosts
+
     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 +491,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 +531,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 +541,22 @@ def format_string(format_config, target_value):
     else:
         return format_config % (target_value)
 
+
+def format_host_port(host, port):
+    host = host.strip()
+    if ":" not in host:
+        return "{host}:{port}".format(host=host, port=port)
+    return host
+
+
+def parse_hosts(host, port=9200):
+    """
+    Convert 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
+    """
+    host_list = host.split(",")
+    host_list = [format_host_port(x, port) for x in host_list]
+    return host_list
+
diff --git a/tests/util_test.py b/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_hosts
 
 
 @pytest.mark.parametrize('spec, expected_delta', [
@@ -338,6 +339,7 @@ def test_ts_utc_to_tz():
             'profile': None,
             'headers': None,
             'es_host': 'localhost',
+            'es_hosts': None,
             'es_port': 9200,
             'es_url_prefix': '',
             'es_conn_timeout': 20,
@@ -361,6 +363,7 @@ def test_ts_utc_to_tz():
             'profile': 'default',
             'headers': None,
             'es_host': 'localhost',
+            'es_hosts': None,
             'es_port': 9200,
             'es_url_prefix': 'elasticsearch',
             'es_conn_timeout': 30,
@@ -436,6 +439,77 @@ def test_build_es_conn_config2():
         'profile': None,
         'headers': None,
         'es_host': 'localhost',
+        'es_hosts': None,
+        'es_port': 9200,
+        'es_url_prefix': '',
+        'es_conn_timeout': 20,
+        'send_get_body_as': 'GET',
+        'ssl_show_warn': True
+    }
+    actual = build_es_conn_config(conf)
+    assert expected == actual
+
+
+@mock.patch.dict(os.environ, {'ES_USERNAME': 'USER',
+                              'ES_PASSWORD': 'PASS',
+                              'ES_API_KEY': 'KEY',
+                              'ES_BEARER': 'BEARE'})
+def test_build_es_conn_config_es_hosts_list():
+    conf = {}
+    conf['es_host'] = 'localhost'
+    conf['es_port'] = 9200
+    conf['es_hosts'] = ['host1:123', 'host2']
+    expected = {
+        'use_ssl': False,
+        'verify_certs': True,
+        'ca_certs': None,
+        'client_cert': None,
+        'client_key': None,
+        'http_auth': None,
+        'es_username': 'USER',
+        'es_password': 'PASS',
+        'es_api_key': 'KEY',
+        'es_bearer': 'BEARE',
+        'aws_region': None,
+        'profile': None,
+        'headers': None,
+        'es_host': 'localhost',
+        'es_hosts': ['host1:123', 'host2'],
+        'es_port': 9200,
+        'es_url_prefix': '',
+        'es_conn_timeout': 20,
+        'send_get_body_as': 'GET',
+        'ssl_show_warn': True
+    }
+    actual = build_es_conn_config(conf)
+    assert expected == actual
+
+
+@mock.patch.dict(os.environ, {'ES_USERNAME': 'USER',
+                              'ES_PASSWORD': 'PASS',
+                              'ES_API_KEY': 'KEY',
+                              'ES_HOSTS': 'host1:123,host2',
+                              'ES_BEARER': 'BEARE'})
+def test_build_es_conn_config_es_hosts_csv():
+    conf = {}
+    conf['es_host'] = 'localhost'
+    conf['es_port'] = 9200
+    expected = {
+        'use_ssl': False,
+        'verify_certs': True,
+        'ca_certs': None,
+        'client_cert': None,
+        'client_key': None,
+        'http_auth': None,
+        'es_username': 'USER',
+        'es_password': 'PASS',
+        'es_api_key': 'KEY',
+        'es_bearer': 'BEARE',
+        'aws_region': None,
+        'profile': None,
+        'headers': None,
+        'es_host': 'localhost',
+        'es_hosts': ['host1:123', 'host2:9200'],
         'es_port': 9200,
         'es_url_prefix': '',
         'es_conn_timeout': 20,
@@ -519,3 +593,14 @@ 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_hosts("localhost", port=9200) == ["localhost:9200"]
+    assert parse_hosts("localhost:9201", port=9200) == ["localhost:9201"]
+    assert parse_hosts("host1, host2, host3.foo") == ["host1:9200",
+                                                      "host2:9200",
+                                                      "host3.foo:9200"]
+    assert parse_hosts("host1, host2:9200, host3:9300") == ["host1:9200",
+                                                            "host2:9200",
+                                                            "host3:9300"]