tracing

azure_bootstrap.tracing

Tier 1 tracing primitives: @traced decorator, latency histograms, slow thresholds.

Modules:

Name	Description
decorators	@traced decorator: auto-async, latency recording, slow alerts, error alerts.
latency	Per-operation latency histograms.
log_exception_context	Rich exception logger for except-blocks that swallow the exception.
slow_thresholds	Per-operation slow-budget defaults.
timed_operation	Block-scoped diagnostic timer.

Functions:

Name	Description
traced	Trace sync or async functions.
traced_async	Alias for @traced. The decorator auto-detects async.

traced

traced(*, operation: str | None = None, alert_on_error: str | None = None, sensitive_args: tuple[str, ...] = (), log_result: bool = False, slow_threshold_seconds: float | None = None) -> Callable[[F], F]

Trace sync or async functions.

Records latency on every call (success + exception). Logs entry/exit at DEBUG only. When DEBUG is off, inspect.signature is skipped so the hot path stays cheap.

Source code in azure_bootstrap/tracing/decorators.py

def traced(
    *,
    operation: str | None = None,
    alert_on_error: str | None = None,
    sensitive_args: tuple[str, ...] = (),
    log_result: bool = False,
    slow_threshold_seconds: float | None = None,
) -> Callable[[F], F]:
    """Trace sync or async functions.

    Records latency on every call (success + exception). Logs entry/exit at
    DEBUG only. When DEBUG is off, ``inspect.signature`` is skipped so the
    hot path stays cheap.
    """

    def decorator(func: F) -> F:
        op = operation or f"{func.__module__}.{func.__qualname__}"
        is_async = asyncio.iscoroutinefunction(func)
        logger = logging.getLogger(func.__module__)

        if is_async:

            @functools.wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                trace_id = uuid.uuid4().hex[:12]
                debug_on = logger.isEnabledFor(logging.DEBUG)
                masked_args: dict[str, str] | None = None
                if debug_on:
                    masked_args = _mask_args_for_log(func, args, kwargs, sensitive_args)
                    logger.debug("→ %s", op, extra=_build_entry_extra(op, trace_id, masked_args))
                start = time.monotonic()
                try:
                    result = await func(*args, **kwargs)
                except BaseException as exc:
                    elapsed = round(time.monotonic() - start, 3)
                    _record_latency(op, elapsed, error=True, slow=False)
                    logger.exception(
                        "✗ %s raised %s after %ss",
                        op,
                        type(exc).__name__,
                        elapsed,
                        extra={
                            "operation": op,
                            "trace_id": trace_id,
                            "elapsed_seconds": elapsed,
                            "exception_type": type(exc).__name__,
                        },
                    )
                    _maybe_alert_error(op, exc, alert_on_error, logger)
                    raise
                elapsed = round(time.monotonic() - start, 3)
                threshold = _resolve_threshold(op, slow_threshold_seconds)
                is_slow = threshold is not None and elapsed > threshold
                _record_latency(op, elapsed, error=False, slow=is_slow)
                if debug_on:
                    exit_extra: dict[str, Any] = {
                        "operation": op,
                        "trace_id": trace_id,
                        "elapsed_seconds": elapsed,
                    }
                    if log_result:
                        exit_extra["result"] = _safe_repr(result)
                    logger.debug("✓ %s ok in %.3fs", op, elapsed, extra=exit_extra)
                if is_slow and threshold is not None:
                    _slow_alert(op, elapsed, threshold)
                return result

            return cast(F, async_wrapper)

        @functools.wraps(func)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            trace_id = uuid.uuid4().hex[:12]
            debug_on = logger.isEnabledFor(logging.DEBUG)
            masked_args: dict[str, str] | None = None
            if debug_on:
                masked_args = _mask_args_for_log(func, args, kwargs, sensitive_args)
                logger.debug("→ %s", op, extra=_build_entry_extra(op, trace_id, masked_args))
            start = time.monotonic()
            try:
                result = func(*args, **kwargs)
            except BaseException as exc:
                elapsed = round(time.monotonic() - start, 3)
                _record_latency(op, elapsed, error=True, slow=False)
                logger.exception(
                    "✗ %s raised %s after %ss",
                    op,
                    type(exc).__name__,
                    elapsed,
                    extra={
                        "operation": op,
                        "trace_id": trace_id,
                        "elapsed_seconds": elapsed,
                        "exception_type": type(exc).__name__,
                    },
                )
                _maybe_alert_error(op, exc, alert_on_error, logger)
                raise
            elapsed = round(time.monotonic() - start, 3)
            threshold = _resolve_threshold(op, slow_threshold_seconds)
            is_slow = threshold is not None and elapsed > threshold
            _record_latency(op, elapsed, error=False, slow=is_slow)
            if debug_on:
                exit_extra = {
                    "operation": op,
                    "trace_id": trace_id,
                    "elapsed_seconds": elapsed,
                }
                if log_result:
                    exit_extra["result"] = _safe_repr(result)
                logger.debug("✓ %s ok in %.3fs", op, elapsed, extra=exit_extra)
            if is_slow and threshold is not None:
                _slow_alert(op, elapsed, threshold)
            return result

        return cast(F, sync_wrapper)

    return decorator

traced_async

traced_async(**kwargs: Any) -> Callable[[F], F]

Alias for @traced. The decorator auto-detects async.

Source code in azure_bootstrap/tracing/decorators.py

def traced_async(**kwargs: Any) -> Callable[[F], F]:
    """Alias for ``@traced``. The decorator auto-detects async."""
    return traced(**kwargs)