Feat add dd and otel celery tasks

CHANGELOG.rst

@@ -11,6 +11,13 @@ Change Log
 
 .. There should always be an "Unreleased" section for changes pending release.
 
+
+[5.14.0] - 2024-03-28
+---------------------
+Added
+~~~~~
+* Initial support for sending monitoring data from Celery to  OpenTelemetry collector or Datadog agent.
+
 [5.13.0] - 2024-04-30
 ---------------------
 Added

edx_django_utils/__init__.py

@@ -2,7 +2,7 @@
 EdX utilities for Django Application development..
 """
 
-__version__ = "5.13.0"
+__version__ = "5.14.0"
 
 default_app_config = (
     "edx_django_utils.apps.EdxDjangoUtilsConfig"

edx_django_utils/monitoring/README.rst

@@ -40,6 +40,10 @@ Feature support matrix for built-in telemetry backends:
      - ✅
      - ❌
      - ❌
+   * - Instrument celery tasks directly (``intialize_celery_task``)
+     - ❌
+     - ✅
+     - ✅
 
 Using Custom Attributes
 -----------------------

edx_django_utils/monitoring/__init__.py

@@ -26,6 +26,7 @@
     accumulate,
     background_task,
     increment,
+    initialize_celery_monitoring,
     record_exception,
     set_custom_attribute,
     set_custom_attributes_for_course_key

edx_django_utils/monitoring/internal/backends.py

@@ -50,6 +50,16 @@ def record_exception(self):
         Record the exception that is currently being handled.
         """
 
+    @abstractmethod
+    def initialize_celery_monitoring(self, *args, **kwargs):
+        """
+        Instrument celery to be monitored by the monitoring service.
+
+        Optional kwargs:
+        worker_process_init - required for open telemetry to integrate to the clery signal.
+        Import from from celery.signals.
+        """
+
 
 class NewRelicBackend(TelemetryBackend):
     """
@@ -77,14 +87,17 @@ def record_exception(self):
         # https://docs.newrelic.com/docs/apm/agents/python-agent/python-agent-api/recordexception-python-agent-api/
         newrelic.agent.record_exception()
 
+    def initialize_celery_monitoring(self, *args, **kwargs):
+        pass
+
 
 class OpenTelemetryBackend(TelemetryBackend):
     """
     Send telemetry via OpenTelemetry.
 
     Requirements to use:
 
-    - Install `opentelemetry-api` Python package
+    - Install `opentelemetry-api` and `opentelemetry-instrumentation-celery` Python packages.
     - Configure and initialize OpenTelemetry
 
     API reference: https://opentelemetry-python.readthedocs.io/en/latest/
@@ -93,7 +106,9 @@ class OpenTelemetryBackend(TelemetryBackend):
     def __init__(self):
         # If import fails, the backend won't be used.
         from opentelemetry import trace
+        from opentelemetry.instrumentation.celery import CeleryInstrumentor
         self.otel_trace = trace
+        self.instrumentor = CeleryInstrumentor
 
     def set_attribute(self, key, value):
         # Sets the value on the current span, not necessarily the root
@@ -103,6 +118,17 @@ def set_attribute(self, key, value):
     def record_exception(self):
         self.otel_trace.get_current_span().record_exception(sys.exc_info()[1])
 
+    def initialize_celery_monitoring(self, *args, **kwargs):
+        worker_process_init = kwargs.get('worker_process_init', None)
+        if worker_process_init is not None:
+            @worker_process_init.connect(weak=False)
+            def init_celery_tracing(*args, **kwargs):
+                self.instrumentor().instrument()
+        else:
+            raise Exception(
+                "the worker_process_init celery signal must be provided for OpenTelemetry to monitor celery tasks."
+                )
+
 
 class DatadogBackend(TelemetryBackend):
     """
@@ -118,8 +144,9 @@ class DatadogBackend(TelemetryBackend):
     # pylint: disable=import-outside-toplevel
     def __init__(self):
         # If import fails, the backend won't be used.
-        from ddtrace import tracer
+        from ddtrace import patch, tracer
         self.dd_tracer = tracer
+        self.patch = patch
 
     def set_attribute(self, key, value):
         if root_span := self.dd_tracer.current_root_span():
@@ -129,6 +156,9 @@ def record_exception(self):
         if span := self.dd_tracer.current_span():
             span.set_traceback()
 
+    def initialize_celery_monitoring(self, *args, **kwargs):
+        self.patch(celery=True)
+
 
 # We're using an lru_cache instead of assigning the result to a variable on
 # module load. With the default settings (pointing to a TelemetryBackend

edx_django_utils/monitoring/internal/utils.py

@@ -14,8 +14,6 @@
 We try to keep track of our custom monitoring at:
 https://openedx.atlassian.net/wiki/spaces/PERF/pages/54362736/Custom+Attributes+in+New+Relic
 
-At this time, the custom monitoring will only be reported to New Relic.
-
 """
 from .backends import configured_backends
 from .middleware import CachedCustomMonitoringMiddleware
@@ -110,3 +108,12 @@ def noop_decorator(func):
         return newrelic.agent.background_task(*args, **kwargs)
     else:
         return noop_decorator
+
+
+def initialize_celery_monitoring(*args, **kwargs):
+    """
+    Set monitoring custom attribute.
+    This is not cached.
+    """
+    for backend in configured_backends():
+        backend.initialize_celery_monitoring(*args, **kwargs)

edx_django_utils/monitoring/tests/test_backends.py

@@ -1,13 +1,13 @@
 """
 Tests for TelemetryBackend and implementations.
 """
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import ddt
 import pytest
 from django.test import TestCase, override_settings
 
-from edx_django_utils.monitoring import record_exception, set_custom_attribute
+from edx_django_utils.monitoring import initialize_celery_monitoring, record_exception, set_custom_attribute
 from edx_django_utils.monitoring.internal.backends import configured_backends
 
 
@@ -151,3 +151,19 @@ def test_record_exception(
         mock_nr_record_exception.assert_called_once()
         mock_otel_record_exception.assert_called_once()
         mock_dd_span.assert_called_once()
+
+    # Record exception on current span, not root span.
+    @patch('ddtrace.patch')
+    def test_initialize_celery_monitoring(
+            self, mock_dd_patch
+    ):
+        celery_signal_decorator_mock = Mock()
+        with override_settings(OPENEDX_TELEMETRY=[
+                'edx_django_utils.monitoring.NewRelicBackend',
+                'edx_django_utils.monitoring.OpenTelemetryBackend',
+                'edx_django_utils.monitoring.DatadogBackend',
+        ]):
+            initialize_celery_monitoring(worker_process_init=celery_signal_decorator_mock)
+        mock_dd_patch.assert_called_once()
+        # celery hook should be defined, but not called.
+        self.assertFalse(celery_signal_decorator_mock.called)

requirements/base.txt

@@ -24,19 +24,19 @@ django-crum==0.7.9
     # via -r requirements/base.in
 django-waffle==4.1.0
     # via -r requirements/base.in
-newrelic==9.8.0
+newrelic==9.9.0
     # via -r requirements/base.in
 pbr==6.0.0
     # via stevedore
 psutil==5.9.8
     # via -r requirements/base.in
-pycparser==2.21
+pycparser==2.22
     # via cffi
 pynacl==1.5.0
     # via -r requirements/base.in
-sqlparse==0.4.4
+sqlparse==0.5.0
     # via django
 stevedore==5.2.0
     # via -r requirements/base.in
-typing-extensions==4.10.0
+typing-extensions==4.11.0
     # via asgiref

requirements/ci.txt

@@ -12,27 +12,27 @@ colorama==0.4.6
     # via tox
 distlib==0.3.8
     # via virtualenv
-filelock==3.13.3
+filelock==3.14.0
     # via
     #   tox
     #   virtualenv
 packaging==24.0
     # via
     #   pyproject-api
     #   tox
-platformdirs==4.2.0
+platformdirs==4.2.1
     # via
     #   tox
     #   virtualenv
-pluggy==1.4.0
+pluggy==1.5.0
     # via tox
 pyproject-api==1.6.1
     # via tox
 tomli==2.0.1
     # via
     #   pyproject-api
     #   tox
-tox==4.14.2
+tox==4.15.0
     # via -r requirements/ci.in
-virtualenv==20.25.1
+virtualenv==20.26.1
     # via tox