Merge branch 'main' into metrics-instrumentation-celery

Author: shalevr

CHANGELOG.md

@@ -10,6 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `opentelemetry-instrumentation-system-metrics` Add `process.` prefix to `runtime.memory`, `runtime.cpu.time`, and `runtime.gc_count`. Change `runtime.memory` from count to UpDownCounter. ([#1735](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1735))
 - Add request and response hooks for GRPC instrumentation (client only)
   ([#1706](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1706))
+- `opentelemetry-instrumentation-pymemcache` Update instrumentation to support pymemcache >4
+  ([#1764](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1764))
 
 ### Added
 
@@ -19,9 +21,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1733](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1733))
 - Make Django request span attributes available for `start_span`. 
   ([#1730](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1730))
+- Make ASGI request span attributes available for `start_span`. 
+  ([#1762](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1762))
+- `opentelemetry-instrumentation-celery` Add support for anonymous tasks.
+  ([#1407](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1407)
+
 
 ### Fixed
 
+- Fix elasticsearch db.statement attribute to be sanitized by default
+  ([#1758](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1758))
 - Fix `AttributeError` when AWS Lambda handler receives a list event
   ([#1738](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1738))
 - Fix `None does not implement middleware` error when there are no middlewares registered

instrumentation/README.md

@@ -26,7 +26,7 @@
 | [opentelemetry-instrumentation-mysql](./opentelemetry-instrumentation-mysql) | mysql-connector-python ~= 8.0 | No
 | [opentelemetry-instrumentation-pika](./opentelemetry-instrumentation-pika) | pika >= 0.12.0 | No
 | [opentelemetry-instrumentation-psycopg2](./opentelemetry-instrumentation-psycopg2) | psycopg2 >= 2.7.3.1 | No
-| [opentelemetry-instrumentation-pymemcache](./opentelemetry-instrumentation-pymemcache) | pymemcache >= 1.3.5, < 4 | No
+| [opentelemetry-instrumentation-pymemcache](./opentelemetry-instrumentation-pymemcache) | pymemcache >= 1.3.5, < 5 | No
 | [opentelemetry-instrumentation-pymongo](./opentelemetry-instrumentation-pymongo) | pymongo >= 3.1, < 5.0 | No
 | [opentelemetry-instrumentation-pymysql](./opentelemetry-instrumentation-pymysql) | PyMySQL < 2 | No
 | [opentelemetry-instrumentation-pyramid](./opentelemetry-instrumentation-pyramid) | pyramid >= 1.7 | Yes

instrumentation/opentelemetry-instrumentation-aiohttp-client/tests/test_aiohttp_client_integration.py

@@ -122,7 +122,7 @@ def test_status_codes(self):
                             (span_status, None),
                             {
                                 SpanAttributes.HTTP_METHOD: "GET",
-                                SpanAttributes.HTTP_URL: f"http://{host}:{port}/test-path?query=param#foobar",
+                                SpanAttributes.HTTP_URL: f"http://{host}:{port}/test-path#foobar",
                                 SpanAttributes.HTTP_STATUS_CODE: int(
                                     status_code
                                 ),

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

@@ -531,15 +531,16 @@ async def __call__(self, scope, receive, send):
 
         span_name, additional_attributes = self.default_span_details(scope)
 
+        attributes = collect_request_attributes(scope)
+        attributes.update(additional_attributes)
         span, token = _start_internal_or_server_span(
             tracer=self.tracer,
             span_name=span_name,
             start_time=None,
             context_carrier=scope,
             context_getter=asgi_getter,
+            attributes=attributes,
         )
-        attributes = collect_request_attributes(scope)
-        attributes.update(additional_attributes)
         active_requests_count_attrs = _parse_active_request_count_attrs(
             attributes
         )

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

@@ -198,10 +198,17 @@ def _trace_before_publish(self, *args, **kwargs):
         task = utils.retrieve_task_from_sender(kwargs)
         task_id = utils.retrieve_task_id_from_message(kwargs)
 
-        if task is None or task_id is None:
+        if task_id is None:
             return
 
-        operation_name = f"{_TASK_APPLY_ASYNC}/{task.name}"
+        if task is None:
+            # task is an anonymous task send using send_task or using canvas workflow
+            # Signatures() to send to a task not in the current processes dependency
+            # tree
+            task_name = kwargs.get("sender", "unknown")
+        else:
+            task_name = task.name
+        operation_name = f"{_TASK_APPLY_ASYNC}/{task_name}"
         span = self._tracer.start_span(
             operation_name, kind=trace.SpanKind.PRODUCER
         )
@@ -210,7 +217,7 @@ def _trace_before_publish(self, *args, **kwargs):
         if span.is_recording():
             span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)
             span.set_attribute(SpanAttributes.MESSAGING_MESSAGE_ID, task_id)
-            span.set_attribute(_TASK_NAME_KEY, task.name)
+            span.set_attribute(_TASK_NAME_KEY, task_name)
             utils.set_attributes_from_context(span, kwargs)
 
         activation = trace.use_span(span, end_on_exit=True)

instrumentation/opentelemetry-instrumentation-celery/src/opentelemetry/instrumentation/celery/utils.py

@@ -132,6 +132,8 @@ def attach_span(task, task_id, span, is_publish=False):
     NOTE: We cannot test for this well yet, because we do not run a celery worker,
     and cannot run `task.apply_async()`
     """
+    if task is None:
+        return
     span_dict = getattr(task, CTX_KEY, None)
     if span_dict is None:
         span_dict = {}

instrumentation/opentelemetry-instrumentation-celery/tests/test_tasks.py

@@ -36,6 +36,7 @@ def tearDown(self):
         CeleryInstrumentor().uninstrument()
         self._worker.stop()
         self._thread.join()
+        CeleryInstrumentor().uninstrument()
 
     def test_task(self):
         CeleryInstrumentor().instrument()
@@ -97,3 +98,52 @@ def test_uninstrument(self):
 
         spans = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans), 0)
+
+
+class TestCelerySignatureTask(TestBase):
+    def setUp(self):
+        super().setUp()
+
+        def start_app(*args, **kwargs):
+            # Add an additional task that will not be registered with parent thread
+            @app.task
+            def hidden_task(num_a):
+                return num_a * 2
+
+            self._worker = app.Worker(app=app, pool="solo", concurrency=1)
+            return self._worker.start(*args, **kwargs)
+
+        self._thread = threading.Thread(target=start_app)
+        self._worker = app.Worker(app=app, pool="solo", concurrency=1)
+        self._thread.daemon = True
+        self._thread.start()
+
+    def tearDown(self):
+        super().tearDown()
+        self._worker.stop()
+        self._thread.join()
+        CeleryInstrumentor().uninstrument()
+
+    def test_hidden_task(self):
+        # no-op since already instrumented
+        CeleryInstrumentor().instrument()
+
+        res = app.signature("tests.test_tasks.hidden_task", (2,)).apply_async()
+        while not res.ready():
+            time.sleep(0.05)
+        spans = self.sorted_spans(self.memory_exporter.get_finished_spans())
+        self.assertEqual(len(spans), 2)
+
+        consumer, producer = spans
+
+        self.assertEqual(consumer.name, "run/tests.test_tasks.hidden_task")
+        self.assertEqual(consumer.kind, SpanKind.CONSUMER)
+
+        self.assertEqual(
+            producer.name, "apply_async/tests.test_tasks.hidden_task"
+        )
+        self.assertEqual(producer.kind, SpanKind.PRODUCER)
+
+        self.assertNotEqual(consumer.parent, producer.context)
+        self.assertEqual(consumer.parent.span_id, producer.context.span_id)
+        self.assertEqual(consumer.context.trace_id, producer.context.trace_id)

instrumentation/opentelemetry-instrumentation-celery/tests/test_utils.py

@@ -185,6 +185,13 @@ def fn_task():
         utils.detach_span(fn_task, task_id)
         self.assertEqual(utils.retrieve_span(fn_task, task_id), (None, None))
 
+    def test_optional_task_span_attach(self):
+        task_id = "7c6731af-9533-40c3-83a9-25b58f0d837f"
+        span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
+
+        # assert this is is a no-aop
+        self.assertIsNone(utils.attach_span(None, task_id, span))
+
     def test_span_delete_empty(self):
         # ensure detach_span doesn't raise an exception if span is not present
         @self.app.task

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

@@ -44,7 +44,6 @@
 
 The instrument() method accepts the following keyword args:
 tracer_provider (TracerProvider) - an optional tracer provider
-sanitize_query (bool) - an optional query sanitization flag
 request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request
 this function signature is:
 def request_hook(span: Span, method: str, url: str, kwargs)
@@ -138,13 +137,11 @@ def _instrument(self, **kwargs):
         tracer = get_tracer(__name__, __version__, tracer_provider)
         request_hook = kwargs.get("request_hook")
         response_hook = kwargs.get("response_hook")
-        sanitize_query = kwargs.get("sanitize_query", False)
         _wrap(
             elasticsearch,
             "Transport.perform_request",
             _wrap_perform_request(
                 tracer,
-                sanitize_query,
                 self._span_name_prefix,
                 request_hook,
                 response_hook,
@@ -163,7 +160,6 @@ def _uninstrument(self, **kwargs):
 
 def _wrap_perform_request(
     tracer,
-    sanitize_query,
     span_name_prefix,
     request_hook=None,
     response_hook=None,
@@ -225,10 +221,9 @@ def wrapper(wrapped, _, args, kwargs):
                 if method:
                     attributes["elasticsearch.method"] = method
                 if body:
-                    statement = str(body)
-                    if sanitize_query:
-                        statement = sanitize_body(body)
-                    attributes[SpanAttributes.DB_STATEMENT] = statement
+                    attributes[SpanAttributes.DB_STATEMENT] = sanitize_body(
+                        body
+                    )
                 if params:
                     attributes["elasticsearch.params"] = str(params)
                 if doc_id:

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

@@ -29,7 +29,8 @@ def _flatten_dict(d, parent_key=""):
     items = []
     for k, v in d.items():
         new_key = parent_key + "." + k if parent_key else k
-        if isinstance(v, dict):
+        # recursive call _flatten_dict for a non-empty dict value
+        if isinstance(v, dict) and v:
             items.extend(_flatten_dict(v, new_key).items())
         else:
             items.append((new_key, v))

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

@@ -58,9 +58,7 @@ class TestElasticsearchIntegration(TestBase):
         "elasticsearch.url": "/test-index/_search",
         "elasticsearch.method": helpers.dsl_search_method,
         "elasticsearch.target": "test-index",
-        SpanAttributes.DB_STATEMENT: str(
-            {"query": {"bool": {"filter": [{"term": {"author": "testing"}}]}}}
-        ),
+        SpanAttributes.DB_STATEMENT: str({"query": {"bool": {"filter": "?"}}}),
     }
 
     create_attributes = {
@@ -264,18 +262,6 @@ def test_dsl_search(self, request_mock):
         )
 
     def test_dsl_search_sanitized(self, request_mock):
-        # Reset instrumentation to use sanitized query (default)
-        ElasticsearchInstrumentor().uninstrument()
-        ElasticsearchInstrumentor().instrument(sanitize_query=True)
-
-        # update expected attributes to match sanitized query
-        sanitized_search_attributes = self.search_attributes.copy()
-        sanitized_search_attributes.update(
-            {
-                SpanAttributes.DB_STATEMENT: "{'query': {'bool': {'filter': '?'}}}"
-            }
-        )
-
         request_mock.return_value = (1, {}, '{"hits": {"hits": []}}')
         client = Elasticsearch()
         search = Search(using=client, index="test-index").filter(
@@ -289,7 +275,7 @@ def test_dsl_search_sanitized(self, request_mock):
         self.assertIsNotNone(span.end_time)
         self.assertEqual(
             span.attributes,
-            sanitized_search_attributes,
+            self.search_attributes,
         )
 
     def test_dsl_create(self, request_mock):
@@ -320,9 +306,6 @@ def test_dsl_create(self, request_mock):
         )
 
     def test_dsl_create_sanitized(self, request_mock):
-        # Reset instrumentation to explicitly use sanitized query
-        ElasticsearchInstrumentor().uninstrument()
-        ElasticsearchInstrumentor().instrument(sanitize_query=True)
         request_mock.return_value = (1, {}, {})
         client = Elasticsearch()
         Article.init(using=client)

instrumentation/opentelemetry-instrumentation-pymemcache/pyproject.toml

@@ -33,7 +33,7 @@ dependencies = [
 
 [project.optional-dependencies]
 instruments = [
-  "pymemcache >= 1.3.5, < 4",
+  "pymemcache >= 1.3.5, < 5",
 ]
 test = [
   "opentelemetry-instrumentation-pymemcache[instruments]",

instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/package.py

@@ -13,4 +13,4 @@
 # limitations under the License.
 
 
-_instruments = ("pymemcache >= 1.3.5, < 4",)
+_instruments = ("pymemcache >= 1.3.5, < 5",)

opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py

@@ -105,7 +105,7 @@
         "instrumentation": "opentelemetry-instrumentation-psycopg2==0.39b0.dev",
     },
     "pymemcache": {
-        "library": "pymemcache >= 1.3.5, < 4",
+        "library": "pymemcache >= 1.3.5, < 5",
         "instrumentation": "opentelemetry-instrumentation-pymemcache==0.39b0.dev",
     },
     "pymongo": {

tox.ini

@@ -122,8 +122,8 @@ envlist =
     ; ext-psycopg2 intentionally excluded from pypy3
 
     ; opentelemetry-instrumentation-pymemcache
-    py3{7,8,9,10,11}-test-instrumentation-pymemcache{135,200,300,342}
-    pypy3-test-instrumentation-pymemcache{135,200,300,342}
+    py3{7,8,9,10,11}-test-instrumentation-pymemcache{135,200,300,342,400}
+    pypy3-test-instrumentation-pymemcache{135,200,300,342,400}
 
     ; opentelemetry-instrumentation-pymongo
     py3{7,8,9,10,11}-test-instrumentation-pymongo
@@ -267,6 +267,7 @@ deps =
   pymemcache200: pymemcache >2.0.0,<3.0.0
   pymemcache300: pymemcache >3.0.0,<3.4.2
   pymemcache342: pymemcache ==3.4.2
+  pymemcache400: pymemcache ==4.0.0
   httpx18: httpx>=0.18.0,<0.19.0
   httpx18: respx~=0.17.0
   httpx21: httpx>=0.19.0
@@ -311,7 +312,7 @@ changedir =
   test-instrumentation-pika{0,1}: instrumentation/opentelemetry-instrumentation-pika/tests
   test-instrumentation-aio-pika{7,8,9}: instrumentation/opentelemetry-instrumentation-aio-pika/tests
   test-instrumentation-psycopg2: instrumentation/opentelemetry-instrumentation-psycopg2/tests
-  test-instrumentation-pymemcache{135,200,300,342}: instrumentation/opentelemetry-instrumentation-pymemcache/tests
+  test-instrumentation-pymemcache{135,200,300,342,400}: instrumentation/opentelemetry-instrumentation-pymemcache/tests
   test-instrumentation-pymongo: instrumentation/opentelemetry-instrumentation-pymongo/tests
   test-instrumentation-pymysql: instrumentation/opentelemetry-instrumentation-pymysql/tests
   test-instrumentation-pyramid: instrumentation/opentelemetry-instrumentation-pyramid/tests
@@ -390,7 +391,7 @@ commands_pre =
 
   mysql: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-dbapi {toxinidir}/instrumentation/opentelemetry-instrumentation-mysql[test]
 
-  pymemcache{135,200,300,342}: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-pymemcache[test]
+  pymemcache{135,200,300,342,400}: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-pymemcache[test]
 
   pymongo: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-pymongo[test]