feat(quickwit): add quickwit integration

Signed-off-by: Idriss Neumann <idriss.neumann@comwork.io>

README.md

@@ -1,6 +1,6 @@
 # ElastAlert 2
 
-ElastAlert 2 is a standalone software tool for alerting on anomalies, spikes, or other patterns of interest from data in [Elasticsearch][10] and [OpenSearch][9].
+ElastAlert 2 is a standalone software tool for alerting on anomalies, spikes, or other patterns of interest from data in [Elasticsearch][10], [OpenSearch][9] and [Quickwit][13].
 
 ElastAlert 2 is backwards compatible with the original [ElastAlert][0] rules.
 
@@ -43,3 +43,4 @@ ElastAlert 2 is licensed under the [Apache License, Version 2.0][5].
 [10]: https://github.com/elastic/elasticsearch
 [11]: https://github.com/jertel/elastalert2/pkgs/container/elastalert2%2Felastalert2
 [12]: https://elastalert2.readthedocs.io/en/latest/recipes/faq.html#does-elastalert-2-support-elasticsearch-8
+[13]: https://quickwit.io

elastalert/create_index.py

@@ -5,9 +5,10 @@
 import json
 import os
 import time
+import yaml
 
 import elasticsearch.helpers
-import yaml
+
 from elasticsearch import RequestsHttpConnection
 from elasticsearch.client import Elasticsearch
 from elasticsearch.client import IndicesClient
@@ -19,11 +20,20 @@
 
 env = Env(ES_USE_SSL=bool)
 
+def create_quickwit_mappings(es_client, ea_index, recreate=False, old_ea_index=None):
+    # TODO
+    return
 
 def create_index_mappings(es_client, ea_index, recreate=False, old_ea_index=None):
-    esversion = get_version_from_cluster_info(es_client)
+    esversion, distribution = get_version_from_cluster_info(es_client)
 
     es_index_mappings = {}
+
+    if distribution == "quickwit":
+        create_quickwit_mappings(es_client, ea_index, recreate, old_ea_index)
+        print('Done!')
+        return
+
     if is_atleasteight(esversion):
         es_index_mappings = read_es_index_mappings()
     elif is_atleastseven(esversion):

elastalert/util.py

@@ -19,6 +19,9 @@
 logging.captureWarnings(True)
 elastalert_logger = logging.getLogger('elastalert')
 
+_elastic7_version = "7.10.2"
+_elastic8_version = "8.2.0"
+
 
 def get_module(module_name):
     """ Loads a module and returns a specific object.
@@ -349,6 +352,36 @@ def replace_dots_in_field_names(document):
     return document
 
 
+def is_not_empty (var):
+    """ test if a value is empty, taking care of string that can be parsed as empty values """
+    if isinstance(var, bool):
+        return var
+    elif isinstance(var, int):
+        return not var == 0
+    elif isinstance(var, list):
+        return len(var) > 0
+    empty_chars = ["", "null", "nil", "false", "none"]
+    return var is not None and not any(c == "{}".format(var).lower() for c in empty_chars)
+
+
+def is_true (var):
+    """ test if a value is true, taking care of string that can be parsed as boolean values """
+    if isinstance(var, bool):
+        return var
+    false_char = ["false", "ko", "no", "off"]
+    return is_not_empty(var) and not any(c == "{}".format(var).lower() for c in false_char)
+
+
+def parse_boolean_conf(key, conf):
+    """ return a boolean parsed configuration. 
+    It will also check if the config exists in the environment variables """
+    qw_enable = os.getenv(key.upper(), False)
+    if is_true(qw_enable):
+        return True
+
+    if key in conf:
+        return is_true(conf['qw_enable'])
+
 def elasticsearch_client(conf):
     """ returns an :class:`ElasticSearchClient` instance configured using an es_conn_config """
     es_conn_conf = build_es_conn_config(conf)
@@ -358,6 +391,9 @@ def elasticsearch_client(conf):
     if es_conn_conf['es_bearer'] or es_conn_conf['es_api_key']:
         username = None
         password = None
+
+    if 'qw_enable' in es_conn_conf and ['qw_enable']
+
     es_conn_conf['http_auth'] = auth(host=es_conn_conf['es_host'],
                                      username=username,
                                      password=password,
@@ -390,6 +426,7 @@ def build_es_conn_config(conf):
     parsed_conf['aws_region'] = None
     parsed_conf['profile'] = None
     parsed_conf['headers'] = None
+    parsed_conf['qw_enable'] = parse_boolean_conf('qw_enable', conf)
     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']))
 
@@ -592,23 +629,29 @@ def parse_hosts(host, port=9200):
 
 def get_version_from_cluster_info(client):
     esversion = None
+    distribution = "elasticsearch"
     for retry in range(3):
         try:
             esinfo = client.info()['version']
             esversion = esinfo['number']
-            if esinfo.get('distribution') == "opensearch":
+            distribution = esinfo.get('distribution')
+            if distribution == "opensearch":
                 # https://opensearch.org/
                 if esversion[0] == "1":
                     # OpenSearch 1.x is based on Elasticsearch 7.10.2
-                    esversion = "7.10.2"
+                    esversion = _elastic7_version
                 else:
                     # OpenSearch 2.x has qualities similar to 8.2.0
-                    esversion = "8.2.0"
+                    esversion = _elastic8_version
+            elif distribution == "quickwit":
+                # Quickwit is aligned with the last opensearch version
+                # which has qualities similar to 8.2.0
+                esversion = _elastic8_version
             break
         except TransportError:
             if retry == 2:
                 raise
             elastalert_logger.warning('Failed to retrieve cluster version information, retrying in 3 seconds')
             time.sleep(3)
 
-    return esversion
+    return distribution, esversion