Record exception in Celery instrumentation

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `opentelemetry-instrumentation-celery` Record exceptions as events on the span.
+  ([#1573](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1573))
 - Add metric instrumentation for urllib
   ([#1553](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1553))
 - `opentelemetry/sdk/extension/aws` Implement [`aws.ecs.*`](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/resource/semantic_conventions/cloud_provider/aws/ecs.md) and [`aws.logs.*`](https://opentelemetry.io/docs/reference/specification/resource/semantic_conventions/cloud_provider/aws/logs/) resource attributes in the `AwsEcsResourceDetector` detector when the ECS Metadata v4 is available
@@ -25,6 +27,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1555](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1555))
 - `opentelemetry-instrumentation-asgi` Fix keys() in class ASGIGetter to correctly fetch values from carrier headers.
   ([#1435](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1435))
+- `opentelemetry/sdk/extension/aws` Implement [`aws.ecs.*`](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/resource/semantic_conventions/cloud_provider/aws/ecs.md) and [`aws.logs.*`](https://opentelemetry.io/docs/reference/specification/resource/semantic_conventions/cloud_provider/aws/logs/) resource attributes in the `AwsEcsResourceDetector` detector when the ECS Metadata v4 is available
+  ([#1212](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1212))
 
 
 ## Version 1.15.0/0.36b0 (2022-12-10)

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

@@ -250,6 +250,7 @@ def _trace_failure(*args, **kwargs):
 
         if ex is not None:
             status_kwargs["description"] = str(ex)
+            span.record_exception(ex)
         span.set_status(Status(**status_kwargs))
 
     @staticmethod

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

@@ -27,3 +27,8 @@ class Config:
 @app.task
 def task_add(num_a, num_b):
     return num_a + num_b
+
+
+@app.task
+def task_error():
+    raise Exception("Test exception")

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

@@ -18,9 +18,9 @@
 from opentelemetry.instrumentation.celery import CeleryInstrumentor
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.test.test_base import TestBase
-from opentelemetry.trace import SpanKind
+from opentelemetry.trace import SpanKind, StatusCode
 
-from .celery_test_tasks import app, task_add
+from .celery_test_tasks import app, task_add, task_error
 
 
 class TestCeleryInstrumentation(TestBase):
@@ -83,6 +83,64 @@ def test_task(self):
         self.assertEqual(consumer.parent.span_id, producer.context.span_id)
         self.assertEqual(consumer.context.trace_id, producer.context.trace_id)
 
+    def test_task_with_error(self):
+        CeleryInstrumentor().instrument()
+
+        result = task_error.delay()
+
+        timeout = time.time() + 60 * 1  # 1 minutes from now
+        while not result.ready():
+            if time.time() > timeout:
+                break
+            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.celery_test_tasks.task_error"
+        )
+        self.assertEqual(consumer.kind, SpanKind.CONSUMER)
+        self.assertSpanHasAttributes(
+            consumer,
+            {
+                "celery.action": "run",
+                "celery.state": "FAILURE",
+                SpanAttributes.MESSAGING_DESTINATION: "celery",
+                "celery.task_name": "tests.celery_test_tasks.task_error",
+            },
+        )
+        self.assertEqual(consumer.status.status_code, StatusCode.ERROR)
+        self.assertEqual(len(consumer.events), 1)
+        event = consumer.events[0]
+        self.assertEqual(event.name, "exception")
+        self.assertEqual(
+            event.attributes[SpanAttributes.EXCEPTION_TYPE], "ExceptionInfo"
+        )
+        self.assertIn(SpanAttributes.EXCEPTION_MESSAGE, event.attributes)
+
+        self.assertEqual(
+            producer.name, "apply_async/tests.celery_test_tasks.task_error"
+        )
+        self.assertEqual(producer.kind, SpanKind.PRODUCER)
+        self.assertSpanHasAttributes(
+            producer,
+            {
+                "celery.action": "apply_async",
+                "celery.task_name": "tests.celery_test_tasks.task_error",
+                SpanAttributes.MESSAGING_DESTINATION_KIND: "queue",
+                SpanAttributes.MESSAGING_DESTINATION: "celery",
+            },
+        )
+        self.assertEqual(producer.status.status_code, StatusCode.UNSET)
+        self.assertEqual(len(producer.events), 0)
+
+        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)
+
     def test_uninstrument(self):
         CeleryInstrumentor().instrument()
         CeleryInstrumentor().uninstrument()