Update celery instrumentation to follow semantic conventions

CHANGELOG.md

@@ -140,6 +140,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3610](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3610))
 - infra(ci): Fix git pull failures in core contrib test
   ([#3357](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3357))
+- `opentelemetry-instrumentation-celery`: Implement new messaging semantic convention opt-in in celery instrumentation and add corresponding tests ([#3712](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3712))
 
 ### Added
 

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

@@ -69,13 +69,22 @@ def add(x, y):
 
 from opentelemetry import context as context_api
 from opentelemetry import trace
+from opentelemetry.instrumentation._semconv import (
+    _get_schema_url,
+    _OpenTelemetrySemanticConventionStability,
+    _OpenTelemetryStabilitySignalType,
+    _report_new,
+    _report_old,
+    _StabilityMode,
+)
 from opentelemetry.instrumentation.celery import utils
 from opentelemetry.instrumentation.celery.package import _instruments
 from opentelemetry.instrumentation.celery.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.metrics import get_meter
 from opentelemetry.propagate import extract, inject
 from opentelemetry.propagators.textmap import Getter
+from opentelemetry.semconv._incubating.attributes import messaging_attributes
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace.status import Status, StatusCode
 
@@ -116,27 +125,32 @@ def keys(self, carrier):
 class CeleryInstrumentor(BaseInstrumentor):
     metrics = None
     task_id_to_start_time = {}
+    _sem_conv_opt_in_mode = _StabilityMode.DEFAULT
 
     def instrumentation_dependencies(self) -> Collection[str]:
         return _instruments
 
     def _instrument(self, **kwargs):
         tracer_provider = kwargs.get("tracer_provider")
 
+        self._sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+            _OpenTelemetryStabilitySignalType.MESSAGING
+        )
+
         # pylint: disable=attribute-defined-outside-init
         self._tracer = trace.get_tracer(
             __name__,
             __version__,
             tracer_provider,
-            schema_url="https://opentelemetry.io/schemas/1.11.0",
+            schema_url=_get_schema_url(self._sem_conv_opt_in_mode),
         )
 
         meter_provider = kwargs.get("meter_provider")
         meter = get_meter(
             __name__,
             __version__,
             meter_provider,
-            schema_url="https://opentelemetry.io/schemas/1.11.0",
+            schema_url=_get_schema_url(self._sem_conv_opt_in_mode),
         )
 
         self.create_celery_metrics(meter)
@@ -204,8 +218,12 @@ def _trace_postrun(self, *args, **kwargs):
         # request context tags
         if span.is_recording():
             span.set_attribute(_TASK_TAG_KEY, _TASK_RUN)
-            utils.set_attributes_from_context(span, kwargs)
-            utils.set_attributes_from_context(span, task.request)
+            utils.set_attributes_from_context(
+                span, kwargs, self._sem_conv_opt_in_mode
+            )
+            utils.set_attributes_from_context(
+                span, task.request, self._sem_conv_opt_in_mode
+            )
             span.set_attribute(_TASK_NAME_KEY, task.name)
 
         activation.__exit__(None, None, None)
@@ -240,9 +258,18 @@ def _trace_before_publish(self, *args, **kwargs):
         # apply some attributes here because most of the data is not available
         if span.is_recording():
             span.set_attribute(_TASK_TAG_KEY, _TASK_APPLY_ASYNC)
-            span.set_attribute(SpanAttributes.MESSAGING_MESSAGE_ID, task_id)
+            if _report_new(self._sem_conv_opt_in_mode):
+                span.set_attribute(
+                    messaging_attributes.MESSAGING_MESSAGE_ID, task_id
+                )  # Not necessary since it has the same name as the old attribute but just in case it changes in the future
+            if _report_old(self._sem_conv_opt_in_mode):
+                span.set_attribute(
+                    SpanAttributes.MESSAGING_MESSAGE_ID, task_id
+                )
             span.set_attribute(_TASK_NAME_KEY, task_name)
-            utils.set_attributes_from_context(span, kwargs)
+            utils.set_attributes_from_context(
+                span, kwargs, self._sem_conv_opt_in_mode
+            )
 
         activation = trace.use_span(span, end_on_exit=True)
         activation.__enter__()  # pylint: disable=E1101

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

@@ -20,6 +20,12 @@
 from celery import registry  # pylint: disable=no-name-in-module
 from celery.app.task import Task
 
+from opentelemetry.instrumentation._semconv import (
+    _report_new,
+    _report_old,
+    _StabilityMode,
+)
+from opentelemetry.semconv._incubating.attributes import messaging_attributes
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace import Span
 
@@ -56,7 +62,11 @@
 
 
 # pylint:disable=too-many-branches
-def set_attributes_from_context(span, context):
+def set_attributes_from_context(
+    span,
+    context,
+    sem_conv_opt_in_mode: _StabilityMode = _StabilityMode.DEFAULT,
+):
     """Helper to extract meta values from a Celery Context"""
     if not span.is_recording():
         return
@@ -80,7 +90,7 @@ def set_attributes_from_context(span, context):
             continue
 
         attribute_name = None
-
+        new_attribute_name = None
         # Celery 4.0 uses `origin` instead of `hostname`; this change preserves
         # the same name for the tag despite Celery version
         if key == "origin":
@@ -91,20 +101,39 @@ def set_attributes_from_context(span, context):
             routing_key = value.get("routing_key")
 
             if routing_key is not None:
-                span.set_attribute(
-                    SpanAttributes.MESSAGING_DESTINATION, routing_key
-                )
+                if _report_new(sem_conv_opt_in_mode):
+                    span.set_attribute(
+                        messaging_attributes.MESSAGING_DESTINATION_NAME,
+                        routing_key,
+                    )
+                if _report_old(sem_conv_opt_in_mode):
+                    span.set_attribute(
+                        SpanAttributes.MESSAGING_DESTINATION, routing_key
+                    )
 
             value = str(value)
 
         elif key == "id":
-            attribute_name = SpanAttributes.MESSAGING_MESSAGE_ID
+            if _report_new(sem_conv_opt_in_mode):
+                new_attribute_name = messaging_attributes.MESSAGING_MESSAGE_ID
+            if _report_old(sem_conv_opt_in_mode):
+                attribute_name = SpanAttributes.MESSAGING_MESSAGE_ID
 
         elif key == "correlation_id":
-            attribute_name = SpanAttributes.MESSAGING_CONVERSATION_ID
+            if _report_new(sem_conv_opt_in_mode):
+                new_attribute_name = (
+                    messaging_attributes.MESSAGING_MESSAGE_CONVERSATION_ID
+                )
+            if _report_old(sem_conv_opt_in_mode):
+                attribute_name = SpanAttributes.MESSAGING_CONVERSATION_ID
 
         elif key == "routing_key":
-            attribute_name = SpanAttributes.MESSAGING_DESTINATION
+            if _report_new(sem_conv_opt_in_mode):
+                new_attribute_name = (
+                    messaging_attributes.MESSAGING_DESTINATION_NAME
+                )
+            if _report_old(sem_conv_opt_in_mode):
+                attribute_name = SpanAttributes.MESSAGING_DESTINATION
 
         # according to https://docs.celeryproject.org/en/stable/userguide/routing.html#exchange-types
         elif key == "declare":
@@ -117,11 +146,13 @@ def set_attributes_from_context(span, context):
                     value = "topic"
                     break
 
+        if attribute_name:
+            span.set_attribute(attribute_name, value)
+        if new_attribute_name:
+            span.set_attribute(new_attribute_name, value)
         # set attribute name if not set specially for a key
-        if attribute_name is None:
-            attribute_name = f"celery.{key}"
-
-        span.set_attribute(attribute_name, value)
+        if attribute_name is None and new_attribute_name is None:
+            span.set_attribute(f"celery.{key}", value)
 
 
 def attach_context(