Merge pull request #5698 from rupozzi/replace-ES

[integration] Replace Elasticsearch clients with OpenSearch, remove support for ES6 - resolves #5167

Author: fstagni

.github/workflows/integration.yml

@@ -35,8 +35,6 @@ jobs:
             ARGS: MYSQL_VER=mysql:5.7
           - TEST_NAME: "MariaDB 10.6, opensearch:1.0.0"
             ARGS: MYSQL_VER=mariadb:10.6.3 ES_VER=opensearchproject/opensearch:1.0.0
-          - TEST_NAME: "Elasticsearch 6.6.0"
-            ARGS: ES_VER=elasticsearch:6.6.0
           - TEST_NAME: "HTTPS"
             ARGS: TEST_HTTPS=Yes
 

docs/source/AdministratorGuide/ExternalsSupport/index.rst

@@ -57,7 +57,6 @@ ElasticSearch versions:
 
 ElasticSearch is an optional dependency for DIRAC servers installations. Supported versions:
 
-- 6.x
 - 7.x
 
 ElasticSearch server is not shipped with DIRAC. You are responsible of its administration.

docs/source/AdministratorGuide/Systems/MonitoringSystem/index.rst

@@ -23,8 +23,7 @@ Install Elasticsearch
 ======================
 
 This is not covered here, as installation and administration of ES are not part of DIRAC guide.
-Just a note on the ES versions supported: ES7 and ES6 are supported, the support for ES5 is not assured,
-and the one for ES6 will be dropped in a future release.
+Just a note on the ES versions supported: only ES7+ versions are currently supported, and are later to be replaced by OpenSearch services.
 
 Configure the MonitoringSystem
 ===============================

environment.yml

@@ -19,6 +19,8 @@ dependencies:
   - db12
   - elasticsearch <7.14
   - elasticsearch-dsl
+  - opensearch-py
+  - opensearch-dsl
   - fts3
   - future
   - gitpython >=2.1.0

src/DIRAC/Core/Utilities/ElasticSearchDB.py

@@ -3,12 +3,6 @@
 It is used to query Elasticsearch instances.
 """
 
-from __future__ import absolute_import
-from __future__ import division
-from __future__ import print_function
-
-__RCSID__ = "$Id$"
-
 from datetime import datetime
 from datetime import timedelta
 
@@ -17,10 +11,16 @@
 import functools
 import json
 
-from elasticsearch import Elasticsearch
-from elasticsearch_dsl import Search, Q, A
-from elasticsearch.exceptions import ConnectionError, TransportError, NotFoundError, RequestError
-from elasticsearch.helpers import BulkIndexError, bulk
+try:
+    from opensearchpy import OpenSearch as Elasticsearch
+    from opensearch_dsl import Search, Q, A
+    from opensearchpy.exceptions import ConnectionError, TransportError, NotFoundError, RequestError
+    from opensearchpy.helpers import BulkIndexError, bulk
+except ImportError:
+    from elasticsearch import Elasticsearch
+    from elasticsearch_dsl import Search, Q, A
+    from elasticsearch.exceptions import ConnectionError, TransportError, NotFoundError, RequestError
+    from elasticsearch.helpers import BulkIndexError, bulk
 
 from DIRAC import gLogger, S_OK, S_ERROR
 from DIRAC.Core.Utilities import Time, DErrno
@@ -53,8 +53,7 @@ def generateDocs(data, withTimeStamp=True):
     :return: doc
     """
     for doc in copy.deepcopy(data):
-        if "_type" not in doc:
-            doc["_type"] = "_doc"
+
         if withTimeStamp:
             if "timestamp" not in doc:
                 sLog.warn("timestamp is not given")
@@ -231,7 +230,7 @@ def update(self, index, query=None, updateByQuery=True, id=None):
             if updateByQuery:
                 esDSLQueryResult = self.client.update_by_query(index=index, body=query)
             else:
-                esDSLQueryResult = self.client.index(index=index, doc_type="_doc", body=query, id=id)
+                esDSLQueryResult = self.client.index(index=index, body=query, id=id)
             return S_OK(esDSLQueryResult)
         except RequestError as re:
             return S_ERROR(re)
@@ -343,11 +342,8 @@ def createIndex(self, indexPrefix, mapping=None, period="day"):
 
         try:
             sLog.info("Create index: ", fullIndex + str(mapping))
-            try:
-                self.client.indices.create(index=fullIndex, body={"mappings": mapping})  # ES7
-            except RequestError as re:
-                if re.error == "mapper_parsing_exception":
-                    self.client.indices.create(index=fullIndex, body={"mappings": {"_doc": mapping}})  # ES6
+            self.client.indices.create(index=fullIndex, body={"mappings": mapping})  # ES7
+
             return S_OK(fullIndex)
         except Exception as e:  # pylint: disable=broad-except
             sLog.error("Can not create the index:", repr(e))
@@ -388,7 +384,7 @@ def index(self, indexName, body=None, docID=None):
             return S_ERROR("Missing index or body")
 
         try:
-            res = self.client.index(index=indexName, doc_type="_doc", body=body, id=docID)
+            res = self.client.index(index=indexName, body=body, id=docID)
         except (RequestError, TransportError) as e:
             sLog.exception()
             return S_ERROR(e)

src/DIRAC/MonitoringSystem/DB/MonitoringDB.py

@@ -138,14 +138,7 @@ def getKeyValues(self, monitoringType):
         docs = retVal["Value"]
         self.log.debug("Doc types", docs)
         monfields = self.documentTypes[monitoringType]["monitoringFields"]
-
-        try:
-            properties = docs[monitoringType]["properties"]  # "old" way, with ES types == Monitoring types
-        except KeyError:
-            try:
-                properties = docs["_doc"]["properties"]  # "ES6" way, with ES types == '_doc'
-            except KeyError:
-                properties = docs["properties"]  # "ES7" way, no types
+        properties = docs["properties"]  # "ES7" way, no types
 
         for i in properties:
             if i not in monfields and not i.startswith("time") and i != "metric":