feat(quickwit): add quickwit integration

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

CHANGELOG.md

@@ -4,10 +4,10 @@
 - None
 
 ## New features
-- None
+- [Quicwit](https://quickwit.io/) integration
 
 ## Other changes
-- None
+- Remove unused import
 
 # 2.17.0
 

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

chart/elastalert2/README.md

@@ -61,6 +61,7 @@ The command removes all the Kubernetes components associated with the chart and
 | `replicaCount`                               | number of replicas to run                                                                                                     | 1                                                        |
 | `minReadySeconds`                            | # number of seconds for which a newly created Pod should be ready without any of its containers crashing, for it to be considered available                                                                                          | 5                                                        |
 | `rulesFolder`                                | Locaton of rules directory. Useful when you have one Docker image and different set of rules per environemnt.                 | /opt/elastalert/rules                                    |
+| `quickwit.enable`                            | enable quickwit interoperability                                                                                              | false                                                    |
 | `elasticsearch.host`                         | elasticsearch endpoint to use                                                                                                 | elasticsearch                                            |
 | `elasticsearch.port`                         | elasticsearch port to use                                                                                                     | 9200                                                     |
 | `elasticsearch.useSsl`                       | whether or not to connect to es_host using SSL                                                                                | False                                                    |

chart/elastalert2/templates/config.yaml

@@ -29,6 +29,9 @@ data:
 {{- end }}
 {{- if .Values.elasticsearch.password }}
     es_password: {{ .Values.elasticsearch.password }}
+{{- end }}
+{{- if .Values.quickwit.enable }}
+    qw_enable: "True"
 {{- end }}
     writeback_index: {{ .Values.writebackIndex }}
     use_ssl: {{ .Values.elasticsearch.useSsl }}

chart/elastalert2/values.yaml

@@ -61,6 +61,10 @@ deploymentAnnotations: {}
 # Annotations to be added to pods
 podAnnotations: {}
 
+quickwit:
+  # enable quickwit interoperability
+  enable: false
+
 elasticsearch:
   # elasticsearch endpoint e.g. (svc.namespace||svc)
   host: elasticsearch

docs/source/configuration.rst

@@ -22,6 +22,8 @@ be specified in the ``es_host`` parameter. The ``es_hosts`` parameter can be ove
 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.
 
+``qw_enable``: (``true`` or ``false``), it will indicate that we're using quickwit and add ``/api/v1/_elastic`` to the URL to ensure interoperability
+
 ``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.
 

docs/source/elastalert.rst

@@ -1,7 +1,7 @@
 Introduction
 ************
 
-ElastAlert 2 is a simple framework for alerting on anomalies, spikes, or other patterns of interest from data in `Elasticsearch <https://www.elastic.co/elasticsearch/>`_ and `OpenSearch <https://opensearch.org/>`_.
+ElastAlert 2 is a simple framework for alerting on anomalies, spikes, or other patterns of interest from data in `Elasticsearch <https://www.elastic.co/elasticsearch/>`_, `OpenSearch <https://opensearch.org/>`_ and `Quickwit <https://quickwit.io>`_.
 
 If you have data being written into Elasticsearch in near real time and want to be alerted when that data matches certain patterns, ElastAlert 2 is the tool for you.
 

elastalert/__init__.py

@@ -6,7 +6,6 @@
 from elasticsearch import RequestsHttpConnection
 from elasticsearch.client import _make_path
 from elasticsearch.client import query_params
-from elasticsearch.exceptions import TransportError
 
 
 class ElasticSearchClient(Elasticsearch):
@@ -47,7 +46,8 @@ def es_version(self):
         Returns the reported version from the Elasticsearch server.
         """
         if self._es_version is None:
-            self._es_version = util.get_version_from_cluster_info(self)
+            (_, version) = util.get_version_from_cluster_info(self)
+            self._es_version = version
 
         return self._es_version
 

elastalert/config.py

@@ -24,6 +24,7 @@
                 'ES_HOSTS': 'es_hosts',
                 'ES_PORT': 'es_port',
                 'ES_URL_PREFIX': 'es_url_prefix',
+                'QW_ENABLE': 'qw_enable',
                 'STATSD_INSTANCE_TAG': 'statsd_instance_tag',
                 'STATSD_HOST': 'statsd_host'}
 

elastalert/create_index.py

@@ -5,25 +5,53 @@
 import json
 import os
 import time
+import yaml
+import requests
 
 import elasticsearch.helpers
-import yaml
+
 from elasticsearch import RequestsHttpConnection
 from elasticsearch.client import Elasticsearch
 from elasticsearch.client import IndicesClient
 from elasticsearch.exceptions import NotFoundError
 from envparse import Env
 
 from elastalert.auth import Auth
-from elastalert.util import get_version_from_cluster_info
+from elastalert.util import _quickwit_url_prefix, get_version_from_cluster_info, is_true, is_empty, is_response_ok
 
 env = Env(ES_USE_SSL=bool)
 
+def create_quickwit_mappings(client_infos, recreate):
+    qw_index_mappings = read_qw_index_mappings()
+    for index_id in qw_index_mappings:
+        if not recreate:
+            r = requests.get("{}/api/v1/indexes/{}/describe".format(client_infos['url'], index_id), auth=client_infos['http_auth'], headers=client_infos['headers'])
+            if is_response_ok(r.status_code):
+                print('Index ' + index_id + ' already exists. Skipping index creation.')
+                continue
+
+        r = requests.post("{}/api/v1/indexes".format(client_infos['url']), json=qw_index_mappings[index_id], auth=client_infos['http_auth'], headers=client_infos['headers'])
+        if is_response_ok(r.status_code):
+            print('Index ' + index_id + ' successfully created.')
+        else:
+            print('Index ' + index_id + ' not created because of error. Attempting to recreate index...')
+            requests.delete("{}/api/v1/indexes/{}".format(client_infos['url'], index_id), auth=client_infos['http_auth'], headers=client_infos['headers'])
+            r = requests.post("{}/api/v1/indexes".format(client_infos['url']), json=qw_index_mappings[index_id], auth=client_infos['http_auth'], headers=client_infos['headers'])
+            if is_response_ok(r.status_code):
+                print('Index ' + index_id + ' successfully created.')
+            else:
+                print('Index ' + index_id + ' not created because of error. Skipping...')
 
-def create_index_mappings(es_client, ea_index, recreate=False, old_ea_index=None):
-    esversion = get_version_from_cluster_info(es_client)
+def create_index_mappings(es_client, ea_index, client_infos, recreate=False, old_ea_index=None):
+    (distribution, esversion) = get_version_from_cluster_info(es_client)
 
     es_index_mappings = {}
+
+    if distribution == "quickwit":
+        create_quickwit_mappings(client_infos, recreate)
+        print('Done!')
+        return
+
     if is_atleasteight(esversion):
         es_index_mappings = read_es_index_mappings()
     elif is_atleastseven(esversion):
@@ -94,6 +122,24 @@ def create_index_mappings(es_client, ea_index, recreate=False, old_ea_index=None
 
     print('Done!')
 
+def read_qw_index_mappings():
+    print('Reading Quickwit index mappings:')
+    return {
+        'silence': read_qw_index_mapping('silence'),
+        'elastalert_status': read_qw_index_mapping('elastalert_status'),
+        'elastalert': read_qw_index_mapping('elastalert'),
+        'past_elastalert': read_qw_index_mapping('past_elastalert'),
+        'elastalert_error': read_qw_index_mapping('elastalert_error')
+    }
+
+def read_qw_index_mapping(mapping):
+    base_path = os.path.abspath(os.path.dirname(__file__))
+    mapping_path = 'qw_mappings/{0}.json'.format(mapping)
+    path = os.path.join(base_path, mapping_path)
+    with open(path, 'r') as f:
+        print("Reading index mapping '{0}'".format(mapping_path))
+        return json.load(f)
+
 
 def read_es_index_mappings(es_version=8):
     print('Reading Elastic {0} index mappings:'.format(es_version))
@@ -105,7 +151,6 @@ def read_es_index_mappings(es_version=8):
         'elastalert_error': read_es_index_mapping('elastalert_error', es_version)
     }
 
-
 def read_es_index_mapping(mapping, es_version=7):
     base_path = os.path.abspath(os.path.dirname(__file__))
     mapping_path = 'es_mappings/{0}/{1}.json'.format(es_version, mapping)
@@ -128,6 +173,7 @@ def main():
     parser.add_argument('--password', default=os.environ.get('ES_PASSWORD', None), help='Elasticsearch password')
     parser.add_argument('--bearer', default=os.environ.get('ES_BEARER', None), help='Elasticsearch bearer token')
     parser.add_argument('--api-key', default=os.environ.get('ES_API_KEY', None), help='Elasticsearch api-key token')
+    parser.add_argument('--quickwit', default=os.environ.get('QW_ENABLE', False), type=bool, help='Quickwit interoperability')
     parser.add_argument('--url-prefix', help='Elasticsearch URL prefix')
     parser.add_argument('--no-auth', action='store_const', const=True, help='Suppress prompt for basic auth')
     parser.add_argument('--ssl', action='store_true', default=env('ES_USE_SSL', None), help='Use TLS')
@@ -165,6 +211,7 @@ def main():
             data = yaml.load(config_file, Loader=yaml.FullLoader)
         host = args.host if args.host else data.get('es_host')
         port = args.port if args.port else data.get('es_port')
+        qw_enable = is_true(args.quickwit) if is_true(args.quickwit) else is_true(data.get('qw_enable'))
         username = args.username if args.username else data.get('es_username')
         password = args.password if args.password else data.get('es_password')
         bearer = args.bearer if args.bearer else data.get('es_bearer')
@@ -187,6 +234,7 @@ def main():
         aws_region = args.aws_region
         host = args.host if args.host else input('Enter Elasticsearch host: ')
         port = args.port if args.port else int(input('Enter Elasticsearch port: '))
+        qw_enable = is_true(args.quickwit)
         use_ssl = (args.ssl if args.ssl is not None
                    else input('Use SSL? t/f: ').lower() in ('t', 'true'))
         if use_ssl:
@@ -224,6 +272,9 @@ def main():
     if api_key is not None:
         headers.update({'Authorization': f'ApiKey {api_key}'})
 
+    if is_true(qw_enable) and not is_empty(url_prefix):
+        url_prefix=_quickwit_url_prefix
+
     es = Elasticsearch(
         host=host,
         port=port,
@@ -238,8 +289,13 @@ def main():
         client_cert=client_cert,
         ca_certs=ca_certs,
         client_key=client_key)
-
-    create_index_mappings(es_client=es, ea_index=index, recreate=args.recreate, old_ea_index=old_index)
+
+    client_infos = {
+        'url': "{}://{}:{}".format("https" if use_ssl else "http", host, port),
+        'headers': headers,
+        'http_auth': http_auth
+    }
+    create_index_mappings(es_client=es, ea_index=index, client_infos=client_infos, recreate=args.recreate, old_ea_index=old_index)
 
 
 if __name__ == '__main__':

elastalert/qw_mappings/elastalert.json

@@ -0,0 +1,53 @@
+{
+  "doc_mapping": {
+    "dynamic_mapping": {
+      "description": "alerts from elastalert2",
+      "expand_dots": true,
+      "fast": true,
+      "indexed": true,
+      "record": "basic",
+      "stored": true,
+      "tokenizer": "raw"
+    },
+    "field_mappings": [
+      {
+        "name": "rule_name",
+        "type": "text"
+      },
+      {
+        "name": "aggregate_id",
+        "type": "text"
+      },
+      {
+        "name": "alert_time",
+        "type": "datetime"
+      },
+      {
+        "name": "match_time",
+        "type": "datetime"
+      },
+      {
+        "name": "match_body",
+        "type": "json"
+      },
+      {
+        "name": "@timestamp",
+        "type": "datetime",
+        "fast": true
+      }
+    ],
+    "store_source": true,
+    "timestamp_field": "@timestamp"
+  },
+  "index_id": "elastalert",
+  "search_settings": {
+    "default_search_fields": [
+      "rule_name",
+      "aggregate_id",
+      "alert_time",
+      "match_time",
+      "@timestamp"
+    ]
+  },
+  "version": "0.8"
+}

elastalert/qw_mappings/elastalert_error.json

@@ -0,0 +1,33 @@
+{
+  "doc_mapping": {
+    "dynamic_mapping": {
+      "description": "errors from elastalert2",
+      "expand_dots": true,
+      "fast": true,
+      "indexed": true,
+      "record": "basic",
+      "stored": true,
+      "tokenizer": "raw"
+    },
+    "field_mappings": [
+      {
+        "name": "data",
+        "type": "json"
+      },
+      {
+        "name": "@timestamp",
+        "type": "datetime",
+        "fast": true
+      }
+    ],
+    "store_source": true,
+    "timestamp_field": "@timestamp"
+  },
+  "index_id": "elastalert_error",
+  "search_settings": {
+    "default_search_fields": [
+      "@timestamp"
+    ]
+  },
+  "version": "0.8"
+}

elastalert/qw_mappings/elastalert_status.json

@@ -0,0 +1,34 @@
+{
+  "doc_mapping": {
+    "dynamic_mapping": {
+      "description": "status alerts from elastalert2",
+      "expand_dots": true,
+      "fast": true,
+      "indexed": true,
+      "record": "basic",
+      "stored": true,
+      "tokenizer": "raw"
+    },
+    "field_mappings": [
+      {
+        "name": "rule_name",
+        "type": "text"
+      },
+      {
+        "name": "@timestamp",
+        "type": "datetime",
+        "fast": true
+      }
+    ],
+    "store_source": true,
+    "timestamp_field": "@timestamp"
+  },
+  "index_id": "elastalert_status",
+  "search_settings": {
+    "default_search_fields": [
+      "rule_name",
+      "@timestamp"
+    ]
+  },
+  "version": "0.8"
+}