Add conditional elastic_transport import (#1810)

* Add conditional elastic_transport import

* Update changelog

* Add future es8 tests

* Update CHANGELOG.md

Co-authored-by: Diego Hurtado <ocelotl@users.noreply.github.com>

* Add license, rm pylint disable

* Consistent elastic version check

* lint import

* Update CHANGELOG.md

---------

Co-authored-by: Diego Hurtado <ocelotl@users.noreply.github.com>
Co-authored-by: Shalev Roda <65566801+shalevr@users.noreply.github.com>

CHANGELOG.md

@@ -31,6 +31,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1789](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1789))
 - `opentelemetry-instrumentation-grpc` Allow gRPC connections via Unix socket
   ([#1833](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1833))
+- Fix elasticsearch `Transport.perform_request` instrument wrap for elasticsearch >= 8
+  ([#1810](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1810))
 
 ## Version 1.18.0/0.39b0 (2023-05-10)
 

instrumentation/opentelemetry-instrumentation-elasticsearch/src/opentelemetry/instrumentation/elasticsearch/__init__.py

@@ -98,6 +98,12 @@ def response_hook(span, response):
 
 from .utils import sanitize_body
 
+# Split of elasticsearch and elastic_transport in 8.0.0+
+# https://www.elastic.co/guide/en/elasticsearch/client/python-api/master/release-notes.html#rn-8-0-0
+es_transport_split = elasticsearch.VERSION[0] > 7
+if es_transport_split:
+    import elastic_transport
+
 logger = getLogger(__name__)
 
 
@@ -137,16 +143,28 @@ def _instrument(self, **kwargs):
         tracer = get_tracer(__name__, __version__, tracer_provider)
         request_hook = kwargs.get("request_hook")
         response_hook = kwargs.get("response_hook")
-        _wrap(
-            elasticsearch,
-            "Transport.perform_request",
-            _wrap_perform_request(
-                tracer,
-                self._span_name_prefix,
-                request_hook,
-                response_hook,
-            ),
-        )
+        if es_transport_split:
+            _wrap(
+                elastic_transport,
+                "Transport.perform_request",
+                _wrap_perform_request(
+                    tracer,
+                    self._span_name_prefix,
+                    request_hook,
+                    response_hook,
+                ),
+            )
+        else:
+            _wrap(
+                elasticsearch,
+                "Transport.perform_request",
+                _wrap_perform_request(
+                    tracer,
+                    self._span_name_prefix,
+                    request_hook,
+                    response_hook,
+                ),
+            )
 
     def _uninstrument(self, **kwargs):
         unwrap(elasticsearch.Transport, "perform_request")

instrumentation/opentelemetry-instrumentation-elasticsearch/tests/helpers_es8.py

@@ -0,0 +1,41 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from elasticsearch_dsl import Document, Keyword, Text
+
+
+class Article(Document):
+    title = Text(analyzer="snowball", fields={"raw": Keyword()})
+    body = Text(analyzer="snowball")
+
+    class Index:
+        name = "test-index"
+
+
+dsl_create_statement = {
+    "mappings": {
+        "properties": {
+            "title": {
+                "analyzer": "snowball",
+                "fields": {"raw": {"type": "keyword"}},
+                "type": "text",
+            },
+            "body": {"analyzer": "snowball", "type": "text"},
+        }
+    }
+}
+dsl_index_result = (1, {}, '{"result": "created"}')
+dsl_index_span_name = "Elasticsearch/test-index/_doc/2"
+dsl_index_url = "/test-index/_doc/2"
+dsl_search_method = "POST"

instrumentation/opentelemetry-instrumentation-elasticsearch/tests/test_elasticsearch.py

@@ -37,7 +37,9 @@
 
 major_version = elasticsearch.VERSION[0]
 
-if major_version == 7:
+if major_version == 8:
+    from . import helpers_es8 as helpers  # pylint: disable=no-name-in-module
+elif major_version == 7:
     from . import helpers_es7 as helpers  # pylint: disable=no-name-in-module
 elif major_version == 6:
     from . import helpers_es6 as helpers  # pylint: disable=no-name-in-module

tox.ini

@@ -255,6 +255,8 @@ deps =
   ; FIXME: Elasticsearch >=7 causes CI workflow tests to hang, see open-telemetry/opentelemetry-python-contrib#620
   ; elasticsearch7: elasticsearch-dsl>=7.0,<8.0
   ; elasticsearch7: elasticsearch>=7.0,<8.0
+  ; elasticsearch8: elasticsearch-dsl>=8.0,<9.0
+  ; elasticsearch8: elasticsearch>=8.0,<9.0
   falcon1: falcon ==1.4.1
   falcon2: falcon >=2.0.0,<3.0.0
   falcon3: falcon >=3.0.0,<4.0.0