elasticsearch: test against elasticsearch 7 (#2431)

* Update core repo SHA

* elasticsearch: test against elasticsearch 7

---------

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

.github/workflows/test.yml

@@ -6,7 +6,7 @@ on:
     - 'release/*'
   pull_request:
 env:
-  CORE_REPO_SHA: 955c92e91b5cd4bcfb43c39efcef086b040471d2
+  CORE_REPO_SHA: 47d5ad7aae5aef31238ca66e55dc550b307c7b35
 
 jobs:
   misc:

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

@@ -15,7 +15,7 @@ python-dateutil==2.8.2
 six==1.16.0
 tomli==2.0.1
 typing_extensions==4.10.0
-urllib3==2.2.1
+urllib3==1.26.18
 wrapt==1.16.0
 zipp==3.17.0
 -e opentelemetry-instrumentation

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

@@ -0,0 +1,22 @@
+asgiref==3.7.2
+attrs==23.2.0
+Deprecated==1.2.14
+elasticsearch==7.17.9
+elasticsearch-dsl==7.4.1
+importlib-metadata==6.11.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==1.26.18
+wrapt==1.16.0
+zipp==3.17.0
+-e opentelemetry-instrumentation
+-e instrumentation/opentelemetry-instrumentation-elasticsearch

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

@@ -26,6 +26,6 @@ dsl_create_statement = {
     }
 }
 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"

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

@@ -50,6 +50,23 @@ Article = helpers.Article
 # pylint: disable=too-many-public-methods
 
 
+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}
+    )
+
+
+def get_elasticsearch_client(*args, **kwargs):
+    client = Elasticsearch(*args, **kwargs)
+    if major_version == 7:
+        client.transport._verified_elasticsearch = True
+    return client
+
+
 @mock.patch(
     "elasticsearch.connection.http_urllib3.Urllib3HttpConnection.perform_request"
 )
@@ -79,10 +96,14 @@ class TestElasticsearchIntegration(TestBase):
             ElasticsearchInstrumentor().uninstrument()
 
     def test_instrumentor(self, request_mock):
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
 
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
-        es.index(index="sw", doc_type="_doc", id=1, body={"name": "adam"})
+        es.index(
+            index="sw",
+            id=1,
+            **normalize_arguments(body={"name": "adam"}, doc_type="_doc"),
+        )
 
         spans_list = self.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
@@ -97,20 +118,24 @@ class TestElasticsearchIntegration(TestBase):
         # check that no spans are generated after uninstrument
         ElasticsearchInstrumentor().uninstrument()
 
-        es.index(index="sw", doc_type="_doc", id=1, body={"name": "adam"})
+        es.index(
+            index="sw",
+            id=1,
+            **normalize_arguments(body={"name": "adam"}, doc_type="_doc"),
+        )
 
         spans_list = self.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
 
     def test_span_not_recording(self, request_mock):
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
         mock_tracer = mock.Mock()
         mock_span = mock.Mock()
         mock_span.is_recording.return_value = False
         mock_tracer.start_span.return_value = mock_span
         with mock.patch("opentelemetry.trace.get_tracer") as tracer:
             tracer.return_value = mock_tracer
-            Elasticsearch()
+            get_elasticsearch_client(hosts=["http://localhost:9200"])
             self.assertFalse(mock_span.is_recording())
             self.assertTrue(mock_span.is_recording.called)
             self.assertFalse(mock_span.set_attribute.called)
@@ -122,7 +147,7 @@ class TestElasticsearchIntegration(TestBase):
         prefix = "prefix-from-env"
         ElasticsearchInstrumentor().uninstrument()
         ElasticsearchInstrumentor(span_name_prefix=prefix).instrument()
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
         self._test_prefix(prefix)
 
     def test_prefix_env(self, request_mock):
@@ -131,13 +156,17 @@ class TestElasticsearchIntegration(TestBase):
         os.environ[env_var] = prefix
         ElasticsearchInstrumentor().uninstrument()
         ElasticsearchInstrumentor().instrument()
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
         del os.environ[env_var]
         self._test_prefix(prefix)
 
     def _test_prefix(self, prefix):
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
-        es.index(index="sw", doc_type="_doc", id=1, body={"name": "adam"})
+        es.index(
+            index="sw",
+            id=1,
+            **normalize_arguments(body={"name": "adam"}, doc_type="_doc"),
+        )
 
         spans_list = self.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
@@ -150,8 +179,10 @@ class TestElasticsearchIntegration(TestBase):
             {},
             '{"found": false, "timed_out": true, "took": 7}',
         )
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
-        es.get(index="test-index", doc_type="_doc", id=1)
+        es.get(
+            index="test-index", **normalize_arguments(doc_type="_doc"), id=1
+        )
 
         spans = self.get_finished_spans()
 
@@ -171,14 +202,18 @@ class TestElasticsearchIntegration(TestBase):
     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.return_value = (1, {}, "{}")
         request_mock.side_effect = exc
         self._test_trace_error(StatusCode.ERROR, exc)
 
     def _test_trace_error(self, code, exc):
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         try:
-            es.get(index="test-index", doc_type="_doc", id=1)
+            es.get(
+                index="test-index",
+                **normalize_arguments(doc_type="_doc"),
+                id=1,
+            )
         except Exception:  # pylint: disable=broad-except
             pass
 
@@ -192,10 +227,14 @@ class TestElasticsearchIntegration(TestBase):
         )
 
     def test_parent(self, request_mock):
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         with self.tracer.start_as_current_span("parent"):
-            es.index(index="sw", doc_type="_doc", id=1, body={"name": "adam"})
+            es.index(
+                index="sw",
+                **normalize_arguments(doc_type="_doc", body={"name": "adam"}),
+                id=1,
+            )
 
         spans = self.get_finished_spans()
         self.assertEqual(len(spans), 2)
@@ -206,8 +245,8 @@ class TestElasticsearchIntegration(TestBase):
         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 = (1, {}, "{}")
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         ev = threading.Event()
 
         # 1. Start tracing from thread-1; make thread-2 wait
@@ -215,13 +254,21 @@ class TestElasticsearchIntegration(TestBase):
         # 3. Check the spans got different parents, and are in the expected order.
         def target1(parent_span):
             with trace.use_span(parent_span):
-                es.get(index="test-index", doc_type="_doc", id=1)
+                es.get(
+                    index="test-index",
+                    **normalize_arguments(doc_type="_doc"),
+                    id=1,
+                )
                 ev.set()
                 ev.wait()
 
         def target2():
             ev.wait()
-            es.get(index="test-index", doc_type="_doc", id=2)
+            es.get(
+                index="test-index",
+                **normalize_arguments(doc_type="_doc"),
+                id=2,
+            )
             ev.set()
 
         with self.tracer.start_as_current_span("parent") as span:
@@ -247,7 +294,7 @@ class TestElasticsearchIntegration(TestBase):
     def test_dsl_search(self, request_mock):
         request_mock.return_value = (1, {}, '{"hits": {"hits": []}}')
 
-        client = Elasticsearch()
+        client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         search = Search(using=client, index="test-index").filter(
             "term", author="testing"
         )
@@ -264,7 +311,7 @@ class TestElasticsearchIntegration(TestBase):
 
     def test_dsl_search_sanitized(self, request_mock):
         request_mock.return_value = (1, {}, '{"hits": {"hits": []}}')
-        client = Elasticsearch()
+        client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         search = Search(using=client, index="test-index").filter(
             "term", author="testing"
         )
@@ -280,8 +327,8 @@ class TestElasticsearchIntegration(TestBase):
         )
 
     def test_dsl_create(self, request_mock):
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
-        client = Elasticsearch()
+        client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         Article.init(using=client)
 
         spans = self.get_finished_spans()
@@ -307,8 +354,8 @@ class TestElasticsearchIntegration(TestBase):
         )
 
     def test_dsl_create_sanitized(self, request_mock):
-        request_mock.return_value = (1, {}, {})
+        request_mock.return_value = (1, {}, "{}")
-        client = Elasticsearch()
+        client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         Article.init(using=client)
 
         spans = self.get_finished_spans()
@@ -323,9 +370,9 @@ class TestElasticsearchIntegration(TestBase):
         )
 
     def test_dsl_index(self, request_mock):
-        request_mock.return_value = helpers.dsl_index_result
+        request_mock.return_value = (1, {}, helpers.dsl_index_result[2])
 
-        client = Elasticsearch()
+        client = get_elasticsearch_client(hosts=["http://localhost:9200"])
         article = Article(
             meta={"id": 2},
             title="About searching",
@@ -374,11 +421,16 @@ class TestElasticsearchIntegration(TestBase):
             {},
             '{"found": false, "timed_out": true, "took": 7}',
         )
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         index = "test-index"
         doc_id = 1
-        kwargs = {"params": {"test": True}}
+        kwargs = {"params": {"refresh": True, "realtime": True}}
-        es.get(index=index, doc_type="_doc", id=doc_id, **kwargs)
+        es.get(
+            index=index,
+            id=doc_id,
+            **normalize_arguments(doc_type="_doc"),
+            **kwargs,
+        )
 
         spans = self.get_finished_spans()
 
@@ -386,12 +438,21 @@ class TestElasticsearchIntegration(TestBase):
         self.assertEqual(
             "GET", spans[0].attributes[request_hook_method_attribute]
         )
+        expected_url = f"/{index}/_doc/{doc_id}"
         self.assertEqual(
-            f"/{index}/_doc/{doc_id}",
+            expected_url,
             spans[0].attributes[request_hook_url_attribute],
         )
+
+        if major_version == 7:
+            expected_kwargs = {
+                **kwargs,
+                "headers": {"accept": "application/json"},
+            }
+        else:
+            expected_kwargs = {**kwargs}
         self.assertEqual(
-            json.dumps(kwargs),
+            json.dumps(expected_kwargs),
             spans[0].attributes[request_hook_kwargs_attribute],
         )
 
@@ -431,13 +492,11 @@ class TestElasticsearchIntegration(TestBase):
             },
         }
 
-        request_mock.return_value = (
+        request_mock.return_value = (1, {}, json.dumps(response_payload))
-            1,
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
-            {},
+        es.get(
-            json.dumps(response_payload),
+            index="test-index", **normalize_arguments(doc_type="_doc"), id=1
         )
-        es = Elasticsearch()
-        es.get(index="test-index", doc_type="_doc", id=1)
 
         spans = self.get_finished_spans()
 
@@ -453,13 +512,11 @@ class TestElasticsearchIntegration(TestBase):
             tracer_provider=trace.NoOpTracerProvider()
         )
         response_payload = '{"found": false, "timed_out": true, "took": 7}'
-        request_mock.return_value = (
+        request_mock.return_value = (1, {}, response_payload)
-            1,
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
-            {},
+        res = es.get(
-            response_payload,
+            index="test-index", **normalize_arguments(doc_type="_doc"), id=1
         )
-        es = Elasticsearch()
-        res = es.get(index="test-index", doc_type="_doc", id=1)
         self.assertEqual(
             res.get("found"), json.loads(response_payload).get("found")
         )
@@ -486,11 +543,11 @@ class TestElasticsearchIntegration(TestBase):
         )
 
     def test_bulk(self, request_mock):
-        request_mock.return_value = (1, {}, "")
+        request_mock.return_value = (1, {}, "{}")
 
-        es = Elasticsearch()
+        es = get_elasticsearch_client(hosts=["http://localhost:9200"])
         es.bulk(
-            [
+            body=[
                 {
                     "_op_type": "index",
                     "_index": "sw",

tox.ini

@@ -75,9 +75,12 @@ envlist =
     ; pypy3-test-instrumentation-boto
 
     ; opentelemetry-instrumentation-elasticsearch
-    ; FIXME: Elasticsearch >=7 causes CI workflow tests to hang, see open-telemetry/opentelemetry-python-contrib#620
+    ; The numbers at the end of the environment names
-    py3{8,9,10,11}-test-instrumentation-elasticsearch
+    ; below mean these dependencies are being used:
-    pypy3-test-instrumentation-elasticsearch
+    ; 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}
 
     ; opentelemetry-instrumentation-falcon
     ; py310 does not work with falcon 1
@@ -640,7 +643,8 @@ commands_pre =
   elasticsearch: pip install opentelemetry-semantic-conventions@{env:CORE_REPO}\#egg=opentelemetry-semantic-conventions&subdirectory=opentelemetry-semantic-conventions
   elasticsearch: pip install opentelemetry-sdk@{env:CORE_REPO}\#egg=opentelemetry-sdk&subdirectory=opentelemetry-sdk
   elasticsearch: pip install opentelemetry-test-utils@{env:CORE_REPO}\#egg=opentelemetry-test-utils&subdirectory=tests/opentelemetry-test-utils
-  elasticsearch: pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements.txt
+  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
 
   asyncio: pip install opentelemetry-api@{env:CORE_REPO}\#egg=opentelemetry-api&subdirectory=opentelemetry-api
   asyncio: pip install opentelemetry-semantic-conventions@{env:CORE_REPO}\#egg=opentelemetry-semantic-conventions&subdirectory=opentelemetry-semantic-conventions
@@ -830,7 +834,7 @@ commands_pre =
   # prerequisite: follow the instructions here https://github.com/PyMySQL/mysqlclient#install
   # for your OS to install the required dependencies
   pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-mysqlclient/test-requirements.txt
-  pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements.txt
+  pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-elasticsearch/test-requirements-1.txt
   pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-tornado/test-requirements.txt
   pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-asyncpg/test-requirements.txt
   pip install -r {toxinidir}/instrumentation/opentelemetry-instrumentation-pymongo/test-requirements.txt