OpenTelemetry Instrumentor for tracing Django management command calls

django_extras_instrumentor.py

from functools import wraps

from django.core.management.base import BaseCommand
from opentelemetry import trace
from opentelemetry.instrumentation.django_extras.package import _instruments
from opentelemetry.instrumentation.instrumentor import BaseInstrumentor

_COMMAND_TRACE_PREFIX = "command/"
_DJANGO_COMMAND_KEY = "django.command"


class DjangoExtrasInstrumentor(BaseInstrumentor):
    def instrumentation_dependencies(self):
        return _instruments

    def _instrument(self):
        tracer = trace.get_tracer(__name__)

        wrapped_execute = BaseCommand.execute

        @wraps(wrapped_execute)
        def instrumented_execute(self, *args, **kwargs):
            command_name = self.__module__.split(".")[-1]
            span_name = f"{_COMMAND_TRACE_PREFIX}{command_name}"
            span_attributes = {_DJANGO_COMMAND_KEY: command_name}
            with tracer.start_as_current_span(
                span_name, attributes=span_attributes
            ) as span:
                try:
                    return wrapped_execute(self, *args, **kwargs)
                except Exception as e:
                    span.set_attribute("error", True)
                    span.record_exception(e)
                    raise

        instrumented_execute.opentelemetry_instrumentation_django_extras_applied = True
        BaseCommand.execute = instrumented_execute

    def _uninstrument(self):
        instr_func = BaseCommand.execute
        if not getattr(
            instr_func,
            "opentelemetry_instrumentation_django_extras_applied",
            False,
        ):
            return
        original = instr_func.__wrapped__
        BaseCommand.execute = original

Very interesting.

How do you use it to automatically monitor all django commands? Do you have an example?

thanks!