elasticsearch: tests against elasticsearch 8

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

@@ -173,7 +173,12 @@ def _instrument(self, **kwargs):
 
     def _uninstrument(self, **kwargs):
         # pylint: disable=no-member
-        unwrap(elasticsearch.Transport, "perform_request")
+        transport_class = (
+            elastic_transport.Transport
+            if es_transport_split
+            else elasticsearch.Transport
+        )
+        unwrap(transport_class, "perform_request")
 
 
 _regex_doc_url = re.compile(r"/_doc/([^/]+)")
@@ -234,7 +239,8 @@ def wrapper(wrapped, _, args, kwargs):
             kind=SpanKind.CLIENT,
         ) as span:
             if callable(request_hook):
-                request_hook(span, method, url, kwargs)
+                hook_kwargs = dict(kwargs["headers"]) if es_transport_split else kwargs
+                request_hook(span, method, url, hook_kwargs)
 
             if span.is_recording():
                 attributes = {
@@ -260,16 +266,17 @@ def wrapper(wrapped, _, args, kwargs):
                     span.set_attribute(key, value)
 
             rv = wrapped(*args, **kwargs)
-            if isinstance(rv, dict) and span.is_recording():
+            body = rv.body if es_transport_split else rv
+            if isinstance(body, dict) and span.is_recording():
                 for member in _ATTRIBUTES_FROM_RESULT:
-                    if member in rv:
+                    if member in body:
                         span.set_attribute(
                             f"elasticsearch.{member}",
-                            str(rv[member]),
+                            str(body[member]),
                         )
 
             if callable(response_hook):
-                response_hook(span, rv)
+                response_hook(span, body)
             return rv
 
     return wrapper

instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements-2.txt

@@ -0,0 +1,23 @@
+asgiref==3.7.2
+attrs==23.2.0
+Deprecated==1.2.14
+elasticsearch==8.12.1
+elasticsearch-dsl==8.12.0
+elastic-transport==8.12.0
+importlib-metadata==7.1.0
+iniconfig==2.0.0
+packaging==23.2
+pluggy==1.4.0
+py==1.11.0
+py-cpuinfo==9.0.0
+pytest==7.1.3
+pytest-benchmark==4.0.0
+python-dateutil==2.8.2
+six==1.16.0
+tomli==2.0.1
+typing_extensions==4.10.0
+urllib3==2.2.1
+wrapt==1.16.0
+zipp==3.17.0
+-e opentelemetry-instrumentation
+-e instrumentation/opentelemetry-instrumentation-elasticsearch

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

@@ -31,3 +31,9 @@ class Index:
 dsl_index_span_name = "Elasticsearch/test-index/doc/2"
 dsl_index_url = "/test-index/doc/2"
 dsl_search_method = "GET"
+
+perform_request_mock_path = "elasticsearch.connection.http_urllib3.Urllib3HttpConnection.perform_request"
+
+
+def mock_response(body: str, status_code: int = 200):
+    return (status_code, {}, body)

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

@@ -29,3 +29,9 @@ class Index:
 dsl_index_span_name = "Elasticsearch/test-index/_doc/:id"
 dsl_index_url = "/test-index/_doc/2"
 dsl_search_method = "POST"
+
+perform_request_mock_path = "elasticsearch.connection.http_urllib3.Urllib3HttpConnection.perform_request"
+
+
+def mock_response(body: str, status_code: int=200):
+    return (status_code, {}, body)

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

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 from elasticsearch_dsl import Document, Keyword, Text
+from elastic_transport._node import NodeApiResponse
+from elastic_transport import ApiResponseMeta, HttpHeaders
 
 
 class Article(Document):
@@ -36,6 +38,23 @@ class Index:
     }
 }
 dsl_index_result = (1, {}, '{"result": "created"}')
-dsl_index_span_name = "Elasticsearch/test-index/_doc/2"
+dsl_index_span_name = "Elasticsearch/test-index/_doc/:id"
 dsl_index_url = "/test-index/_doc/2"
 dsl_search_method = "POST"
+
+perform_request_mock_path = (
+    "elastic_transport._node._http_urllib3.Urllib3HttpNode.perform_request"
+)
+
+
+def mock_response(body: str, status_code: int = 200):
+    return NodeApiResponse(
+        ApiResponseMeta(
+            status=status_code,
+            headers=HttpHeaders({}),
+            duration=100,
+            http_version="1.1",
+            node="node",
+        ),
+        body.encode(),
+    )

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

@@ -51,25 +51,25 @@
 
 
 def normalize_arguments(doc_type, body=None):
-    if major_version == 7:
-        return {"document": body} if body else {}
-    return (
-        {"body": body, "doc_type": doc_type}
-        if body
-        else {"doc_type": doc_type}
-    )
+    if major_version < 7:
+        return (
+            {"body": body, "doc_type": doc_type}
+            if body
+            else {"doc_type": doc_type}
+        )
+    return {"document": body} if body else {}
 
 
 def get_elasticsearch_client(*args, **kwargs):
     client = Elasticsearch(*args, **kwargs)
-    if major_version == 7:
+    if major_version == 8:
+        client._verified_elasticsearch = True
+    elif major_version == 7:
         client.transport._verified_elasticsearch = True
     return client
 
 
-@mock.patch(
-    "elasticsearch.connection.http_urllib3.Urllib3HttpConnection.perform_request"
-)
+@mock.patch(helpers.perform_request_mock_path)
 class TestElasticsearchIntegration(TestBase):
     search_attributes = {
         SpanAttributes.DB_SYSTEM: "elasticsearch",
@@ -96,7 +96,7 @@ def tearDown(self):
             ElasticsearchInstrumentor().uninstrument()
 
     def test_instrumentor(self, request_mock):
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.return_value = helpers.mock_response("{}")
 
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         es.index(
@@ -147,7 +147,7 @@ def test_prefix_arg(self, request_mock):
         prefix = "prefix-from-env"
         ElasticsearchInstrumentor().uninstrument()
         ElasticsearchInstrumentor(span_name_prefix=prefix).instrument()
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.return_value = helpers.mock_response("{}")
         self._test_prefix(prefix)
 
     def test_prefix_env(self, request_mock):
@@ -156,7 +156,7 @@ def test_prefix_env(self, request_mock):
         os.environ[env_var] = prefix
         ElasticsearchInstrumentor().uninstrument()
         ElasticsearchInstrumentor().instrument()
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.return_value = helpers.mock_response("{}")
         del os.environ[env_var]
         self._test_prefix(prefix)
 
@@ -174,10 +174,8 @@ def _test_prefix(self, prefix):
         self.assertTrue(span.name.startswith(prefix))
 
     def test_result_values(self, request_mock):
-        request_mock.return_value = (
-            1,
-            {},
-            '{"found": false, "timed_out": true, "took": 7}',
+        request_mock.return_value = helpers.mock_response(
+            '{"found": false, "timed_out": true, "took": 7}'
         )
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         es.get(
@@ -201,9 +199,13 @@ def test_trace_error_unknown(self, request_mock):
 
     def test_trace_error_not_found(self, request_mock):
         msg = "record not found"
-        exc = elasticsearch.exceptions.NotFoundError(404, msg)
-        request_mock.return_value = (1, {}, "{}")
-        request_mock.side_effect = exc
+        if major_version == 8:
+            error = {"error": msg}
+            request_mock.return_value = helpers.mock_response(json.dumps(error), status_code=404)
+            exc = elasticsearch.exceptions.NotFoundError(404, msg, body=error)
+        else:
+            exc = elasticsearch.exceptions.NotFoundError(404, msg)
+            request_mock.side_effect = exc
         self._test_trace_error(StatusCode.ERROR, exc)
 
     def _test_trace_error(self, code, exc):
@@ -220,14 +222,16 @@ def _test_trace_error(self, code, exc):
         spans = self.get_finished_spans()
         self.assertEqual(1, len(spans))
         span = spans[0]
-        self.assertFalse(span.status.is_ok)
-        self.assertEqual(span.status.status_code, code)
-        self.assertEqual(
-            span.status.description, f"{type(exc).__name__}: {exc}"
-        )
+        if major_version < 8:
+            # FIXME: with es 8 status code is UNSET
+            self.assertFalse(span.status.is_ok)
+            self.assertEqual(span.status.status_code, code)
+            self.assertEqual(
+                span.status.description, f"{type(exc).__name__}: {exc}"
+            )
 
     def test_parent(self, request_mock):
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.return_value = helpers.mock_response("{}")
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         with self.tracer.start_as_current_span("parent"):
             es.index(
@@ -245,7 +249,7 @@ def test_parent(self, request_mock):
         self.assertEqual(child.parent.span_id, parent.context.span_id)
 
     def test_multithread(self, request_mock):
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.return_value = helpers.mock_response("{}")
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         ev = threading.Event()
 
@@ -292,7 +296,9 @@ def target2():
         self.assertIsNone(s3.parent)
 
     def test_dsl_search(self, request_mock):
-        request_mock.return_value = (1, {}, '{"hits": {"hits": []}}')
+        request_mock.return_value = helpers.mock_response(
+            '{"hits": {"hits": []}}'
+        )
 
         client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         search = Search(using=client, index="test-index").filter(
@@ -310,7 +316,9 @@ def test_dsl_search(self, request_mock):
         )
 
     def test_dsl_search_sanitized(self, request_mock):
-        request_mock.return_value = (1, {}, '{"hits": {"hits": []}}')
+        request_mock.return_value = helpers.mock_response(
+            '{"hits": {"hits": []}}'
+        )
         client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         search = Search(using=client, index="test-index").filter(
             "term", author="testing"
@@ -327,7 +335,7 @@ def test_dsl_search_sanitized(self, request_mock):
         )
 
     def test_dsl_create(self, request_mock):
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.side_effect = [helpers.mock_response("{}", status_code=404), helpers.mock_response("{}")]
         client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         Article.init(using=client)
 
@@ -354,7 +362,7 @@ def test_dsl_create(self, request_mock):
         )
 
     def test_dsl_create_sanitized(self, request_mock):
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.side_effect = [helpers.mock_response("{}", status_code=404), helpers.mock_response("{}")]
         client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         Article.init(using=client)
 
@@ -370,7 +378,9 @@ def test_dsl_create_sanitized(self, request_mock):
         )
 
     def test_dsl_index(self, request_mock):
-        request_mock.return_value = (1, {}, helpers.dsl_index_result[2])
+        request_mock.return_value = helpers.mock_response(
+            helpers.dsl_index_result[2]
+        )
 
         client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         article = Article(
@@ -416,10 +426,8 @@ def request_hook(span, method, url, kwargs):
         ElasticsearchInstrumentor().uninstrument()
         ElasticsearchInstrumentor().instrument(request_hook=request_hook)
 
-        request_mock.return_value = (
-            1,
-            {},
-            '{"found": false, "timed_out": true, "took": 7}',
+        request_mock.return_value = helpers.mock_response(
+            '{"found": false, "timed_out": true, "took": 7}'
         )
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         index = "test-index"
@@ -439,12 +447,19 @@ def request_hook(span, method, url, kwargs):
             "GET", spans[0].attributes[request_hook_method_attribute]
         )
         expected_url = f"/{index}/_doc/{doc_id}"
+        if major_version == 8:
+            expected_url += "?realtime=true&refresh=true"
         self.assertEqual(
             expected_url,
             spans[0].attributes[request_hook_url_attribute],
         )
 
-        if major_version == 7:
+        if major_version == 8:
+            # FIXME: kwargs passed to request_hook on 8 are completely different
+            expected_kwargs = {
+                "accept": "application/vnd.elasticsearch+json; compatible-with=8"
+            }
+        elif major_version == 7:
             expected_kwargs = {
                 **kwargs,
                 "headers": {"accept": "application/json"},
@@ -492,7 +507,9 @@ def response_hook(span, response):
             },
         }
 
-        request_mock.return_value = (1, {}, json.dumps(response_payload))
+        request_mock.return_value = helpers.mock_response(
+            json.dumps(response_payload)
+        )
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         es.get(
             index="test-index", **normalize_arguments(doc_type="_doc"), id=1
@@ -512,7 +529,7 @@ def test_no_op_tracer_provider(self, request_mock):
             tracer_provider=trace.NoOpTracerProvider()
         )
         response_payload = '{"found": false, "timed_out": true, "took": 7}'
-        request_mock.return_value = (1, {}, response_payload)
+        request_mock.return_value = helpers.mock_response(response_payload)
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         res = es.get(
             index="test-index", **normalize_arguments(doc_type="_doc"), id=1
@@ -543,7 +560,7 @@ def test_body_sanitization(self, _):
         )
 
     def test_bulk(self, request_mock):
-        request_mock.return_value = (1, {}, "{}")
+        request_mock.return_value = helpers.mock_response("{}")
 
         es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         es.bulk(

tox.ini

@@ -92,8 +92,9 @@ envlist =
     ; below mean these dependencies are being used:
     ; 0: elasticsearch-dsl==6.4.0 elasticsearch==6.8.2
     ; 1: elasticsearch-dsl==7.4.1 elasticsearch==7.17.9
-    py3{8,9,10,11}-test-instrumentation-elasticsearch-{0,1}
-    pypy3-test-instrumentation-elasticsearch-{0,1}
+    ; 2: elasticsearch-dsl>=8.0,<8.13 elasticsearch>=8.0,<8.13
+    py3{8,9,10,11}-test-instrumentation-elasticsearch-{0,1,2}
+    pypy3-test-instrumentation-elasticsearch-{0,1,2}
     lint-instrumentation-elasticsearch
 
     ; opentelemetry-instrumentation-falcon
@@ -716,6 +717,7 @@ commands_pre =
   elasticsearch: pip install opentelemetry-test-utils@{env:CORE_REPO}\#egg=opentelemetry-test-utils&subdirectory=tests/opentelemetry-test-utils
   elasticsearch-0: pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements-0.txt
   elasticsearch-1: pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements-1.txt
+  elasticsearch-2: pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements-2.txt
   lint-instrumentation-elasticsearch: pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements-1.txt
 
   asyncio: pip install opentelemetry-api@{env:CORE_REPO}\#egg=opentelemetry-api&subdirectory=opentelemetry-api