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,82 @@
+"""Pipeline steps for management command observability."""
+
+import logging
+import time
+from contextlib import contextmanager, nullcontext
+
+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 SystemExit as exc:
+        set_custom_attribute('management_command.exception_class', exc.__class__.__name__)
+        set_custom_attribute('management_command.exit_code', exc.code)
+        if exc.code in (0, None):
+            status = 'success'
+        raise
+    except Exception as exc:
+        set_custom_attribute('management_command.exception_class', exc.__class__.__name__)
+        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_contextmanager, command_name, service_variant):  # pylint: disable=arguments-differ
+        """
+        Return a wrapped context manager that applies monitoring when enabled.
+        """
+        trace_name = self.extra_config.get('trace_name', DEFAULT_TRACE_NAME)
+
+        @contextmanager
+        def wrapped_contextmanager():
+            monitor_context = nullcontext()
+
+            try:
+                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 command_contextmanager:
+                with monitor_context:
+                    yield
+
+        return {
+            'command_contextmanager': wrapped_contextmanager(),
+            'command_name': command_name,
+            'service_variant': service_variant,
+        }

edx_filters_pipelines/waffle.py

@@ -1,7 +1,7 @@
 """
-waffle flags used in the filters_pipelines app.
+Feature toggles used in the filters_pipelines app.
 """
-from edx_toggles.toggles import WaffleFlag
+from edx_toggles.toggles import SettingToggle, WaffleFlag
 
 WAFFLE_NAMESPACE = 'filters_pipelines'
 
@@ -14,3 +14,16 @@
 # .. 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: SettingToggle
+# .. toggle_default: False
+# .. toggle_description: Settings toggle 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 setting is True, management command execution is wrapped with Datadog monitoring.
+ENABLE_MANAGEMENT_COMMAND_MONITORING = SettingToggle(
+    'FILTERS_PIPELINES_ENABLE_MANAGEMENT_COMMAND_MONITORING',
+    default=False,
+)

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,16 @@
 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,
+    monitor_management_command,
+)
 
 
 def test_username_blocked():
@@ -20,3 +27,123 @@ 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.contextmanager.requested.v1',
+        'edx_filters_pipelines.management.pipelines.monitoring.ManagementCommandMonitoringPipelineStep',
+    )
+    command_execution = mocker.Mock()
+    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(nullcontext(), 'migrate', 'lms')
+
+    assert result['command_name'] == 'migrate'
+    assert result['service_variant'] == 'lms'
+    with result['command_contextmanager']:
+        command_execution()
+    toggle.assert_called_once()
+    command_execution.assert_called_once()
+
+
+def test_management_command_monitoring_step_enabled(mocker):
+    step = ManagementCommandMonitoringPipelineStep(
+        'org.openedx.platform.management.command.contextmanager.requested.v1',
+        'edx_filters_pipelines.management.pipelines.monitoring.ManagementCommandMonitoringPipelineStep',
+    )
+    command_execution = mocker.Mock()
+    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(nullcontext(), 'migrate', 'lms')['command_contextmanager']
+
+    with wrapped:
+        command_execution()
+
+    command_execution.assert_called_once()
+    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.contextmanager.requested.v1',
+        'edx_filters_pipelines.management.pipelines.monitoring.ManagementCommandMonitoringPipelineStep',
+        trace_name='custom.management.trace',
+    )
+    command_execution = mocker.Mock()
+    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')
+
+    with step.run_filter(nullcontext(), 'collectstatic', 'cms')['command_contextmanager']:
+        command_execution()
+
+    command_execution.assert_called_once()
+    function_trace.assert_called_once_with('custom.management.trace')
+
+
+@pytest.mark.parametrize(
+    'raised_exception, expected_status, expected_exit_code, exception_class_recorded',
+    [
+        pytest.param(SystemExit(0), 'success', 0, True, id='system-exit-zero'),
+        pytest.param(KeyboardInterrupt(), 'failure', None, False, id='keyboard-interrupt'),
+    ],
+)
+def test_monitor_management_command_exception_handling(
+    mocker,
+    raised_exception,
+    expected_status,
+    expected_exit_code,
+    exception_class_recorded,
+):
+    mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.function_trace',
+        return_value=nullcontext(),
+    )
+    set_custom_attribute = mocker.patch(
+        'edx_filters_pipelines.management.pipelines.monitoring.set_custom_attribute'
+    )
+    mocker.patch('edx_filters_pipelines.management.pipelines.monitoring.set_monitoring_transaction_name')
+
+    with pytest.raises(type(raised_exception)) as exc_info:
+        with monitor_management_command('migrate', 'lms'):
+            raise raised_exception
+
+    if expected_exit_code is not None:
+        assert exc_info.value.code == expected_exit_code
+        set_custom_attribute.assert_any_call('management_command.exit_code', expected_exit_code)
+
+    if not exception_class_recorded:
+        assert ('management_command.exception_class', raised_exception.__class__.__name__) not in [
+            call.args for call in set_custom_attribute.call_args_list
+        ]
+
+    set_custom_attribute.assert_any_call('management_command.status', expected_status)

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]