fix: Add management command monitoring pipeline

edx_filters_pipelines/management/__init__.py

@@ -0,0 +1 @@
+"""Management command filter pipeline integrations."""

edx_filters_pipelines/management/pipelines/__init__.py

@@ -0,0 +1 @@
+"""Pipeline steps for management command integrations."""

edx_filters_pipelines/management/pipelines/monitoring.py

@@ -0,0 +1,72 @@
+"""Pipeline steps for management command observability."""
+
+import logging
+import time
+from contextlib import contextmanager, nullcontext
+
+import django
+from edx_django_utils.monitoring import function_trace, set_custom_attribute, set_monitoring_transaction_name
+from openedx_filters import PipelineStep
+
+from edx_filters_pipelines.waffle import ENABLE_MANAGEMENT_COMMAND_MONITORING
+
+log = logging.getLogger(__name__)
+
+DEFAULT_TRACE_NAME = 'django.management.command'
+
+
+@contextmanager
+def monitor_management_command(command_name, service_variant, trace_name=DEFAULT_TRACE_NAME):
+    """Wrap a management command execution with Datadog monitoring metadata."""
+    transaction_name = f'{service_variant}.management.{command_name}'
+
+    set_monitoring_transaction_name(transaction_name)
+    set_custom_attribute('management_command.name', command_name)
+    set_custom_attribute('management_command.service_variant', service_variant)
+    set_custom_attribute('management_command.transaction_name', transaction_name)
+
+    start_time = time.monotonic()
+    status = 'failure'
+
+    try:
+        with function_trace(trace_name):
+            yield
+        status = 'success'
+    except BaseException as exc:
+        set_custom_attribute('management_command.exception_class', exc.__class__.__name__)
+        if isinstance(exc, SystemExit):
+            set_custom_attribute('management_command.exit_code', exc.code)
+        raise
+    finally:
+        set_custom_attribute('management_command.status', status)
+        set_custom_attribute('management_command.duration_seconds', time.monotonic() - start_time)
+
+
+class ManagementCommandMonitoringPipelineStep(PipelineStep):
+    """Add Datadog monitoring around Django management command execution."""
+
+    def run_filter(self, command_name, service_variant, command_runner):  # pylint: disable=arguments-differ
+        """Return a wrapped command runner that applies monitoring when enabled."""
+        trace_name = self.extra_config.get('trace_name', DEFAULT_TRACE_NAME)
+
+        def wrapped_runner():
+            monitor_context = nullcontext()
+
+            try:
+                django.setup()
+                if ENABLE_MANAGEMENT_COMMAND_MONITORING.is_enabled():
+                    monitor_context = monitor_management_command(command_name, service_variant, trace_name)
+            except Exception:  # pylint: disable=broad-except
+                log.exception(
+                    'Failed to initialize management command monitoring for %s; continuing without monitoring.',
+                    command_name,
+                )
+
+            with monitor_context:
+                return command_runner()
+
+        return {
+            'command_name': command_name,
+            'service_variant': service_variant,
+            'command_runner': wrapped_runner,
+        }

edx_filters_pipelines/waffle.py

@@ -14,3 +14,15 @@
 # .. toggle_target_removal_date: None because this is a long-term feature
 # .. toggle_warning: When the flag is ON, recaptcha validation is enabled on registration.
 ENABLE_RECAPTCHA_VALIDATION = WaffleFlag(f'{WAFFLE_NAMESPACE}.enable_registration_recaptcha_validation', __name__)
+
+# .. toggle_name: filters_pipelines.enable_management_command_monitoring
+# .. toggle_implementation: WaffleFlag
+# .. toggle_default: False
+# .. toggle_description: Waffle flag to enable Datadog monitoring for Django management commands.
+# .. toggle_use_cases: opt_in
+# .. toggle_creation_date: 2026-03-26
+# .. toggle_target_removal_date: None because this is a long-term feature
+# .. toggle_warning: When the flag is ON, management command execution is wrapped with Datadog monitoring.
+ENABLE_MANAGEMENT_COMMAND_MONITORING = WaffleFlag(
+	f'{WAFFLE_NAMESPACE}.enable_management_command_monitoring', __name__
+)

requirements/test.in

@@ -5,6 +5,7 @@
 
 pytest
 pytest-cov                # pytest extension for code coverage statistics
+pytest-mock               # pytest plugin providing mocker fixture
 edx-lint
 pylint-celery
 setuptools

requirements/test.txt

@@ -196,6 +196,8 @@ pytest-cov==6.2.1
     # via
     #   -r requirements/base.txt
     #   -r requirements/test.in
+pytest-mock
+    # via -r requirements/test.in
 python-slugify==8.0.4
     # via
     #   -r requirements/base.txt

tests/test_edx-filters-pipelines.py

@@ -2,9 +2,13 @@
 Tests for edx-filters-pipelines.py.
 """
 
+from contextlib import nullcontext
+
 import pytest
 from openedx_filters.learning.filters import StudentRegistrationRequested
+
 from edx_filters_pipelines.auth.pipelines.registration import PreventForbiddenUsernameRegistration
+from edx_filters_pipelines.management.pipelines.monitoring import ManagementCommandMonitoringPipelineStep
 
 
 def test_username_blocked():
@@ -20,3 +24,82 @@ def test_username_blocked():
         step.run_filter(form_data=form_data)
 
     assert "Usernames can't include words that could be mistaken for course roles." in str(exc_info.value)
+
+
+def test_management_command_monitoring_step_disabled(mocker):
+    step = ManagementCommandMonitoringPipelineStep(
+        'org.openedx.platform.management.command.execute.requested.v1',
+        'edx_filters_pipelines.management.pipelines.monitoring.ManagementCommandMonitoringPipelineStep',
+    )
+    command_runner = mocker.Mock(return_value='ok')
+    django_setup = mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.django.setup')
+    toggle = mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.ENABLE_MANAGEMENT_COMMAND_MONITORING.is_enabled',
+        return_value=False,
+    )
+    mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.function_trace')
+
+    result = step.run_filter('migrate', 'lms', command_runner)
+
+    assert result['command_name'] == 'migrate'
+    assert result['service_variant'] == 'lms'
+    assert result['command_runner']() == 'ok'
+    django_setup.assert_called_once()
+    toggle.assert_called_once()
+    command_runner.assert_called_once()
+
+
+def test_management_command_monitoring_step_enabled(mocker):
+    step = ManagementCommandMonitoringPipelineStep(
+        'org.openedx.platform.management.command.execute.requested.v1',
+        'edx_filters_pipelines.management.pipelines.monitoring.ManagementCommandMonitoringPipelineStep',
+    )
+    command_runner = mocker.Mock(return_value='ok')
+    mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.django.setup')
+    mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.ENABLE_MANAGEMENT_COMMAND_MONITORING.is_enabled',
+        return_value=True,
+    )
+    function_trace = mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.function_trace',
+        return_value=nullcontext(),
+    )
+    set_transaction_name = mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.set_monitoring_transaction_name'
+    )
+    set_custom_attribute = mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.set_custom_attribute'
+    )
+
+    wrapped = step.run_filter('migrate', 'lms', command_runner)['command_runner']
+
+    assert wrapped() == 'ok'
+    function_trace.assert_called_once_with('django.management.command')
+    set_transaction_name.assert_called_once_with('lms.management.migrate')
+    set_custom_attribute.assert_any_call('management_command.name', 'migrate')
+    set_custom_attribute.assert_any_call('management_command.service_variant', 'lms')
+    set_custom_attribute.assert_any_call('management_command.status', 'success')
+
+
+def test_management_command_monitoring_step_uses_configured_trace_name(mocker):
+    step = ManagementCommandMonitoringPipelineStep(
+        'org.openedx.platform.management.command.execute.requested.v1',
+        'edx_filters_pipelines.management.pipelines.monitoring.ManagementCommandMonitoringPipelineStep',
+        trace_name='custom.management.trace',
+    )
+    command_runner = mocker.Mock(return_value=None)
+    mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.django.setup')
+    mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.ENABLE_MANAGEMENT_COMMAND_MONITORING.is_enabled',
+        return_value=True,
+    )
+    function_trace = mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.function_trace',
+        return_value=nullcontext(),
+    )
+    mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.set_monitoring_transaction_name')
+    mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.set_custom_attribute')
+
+    step.run_filter('collectstatic', 'cms', command_runner)['command_runner']()
+
+    function_trace.assert_called_once_with('custom.management.trace')

tox.ini

@@ -31,7 +31,7 @@ ignore = D101,D200,D203,D212,D215,D404,D405,D406,D407,D408,D409,D410,D411,D412,D
 
 
 [pytest]
-addopts = --cov edx-filters-pipelines --cov-report term-missing --cov-report xml
+addopts = --cov edx_filters_pipelines --cov-report term-missing --cov-report xml
 norecursedirs = .* docs requirements site-packages
 
 [testenv]