orchestrata_ultimate.refactor.codemod.observability.metrics

Reference › orchestrata_ultimate › refactor › codemod › observability › metrics

MetricsObserver turns codemod events into Prometheus metrics. Its counters are codemod_passes_run_total (by pass name and type), codemod_files_mutated_total (by pass name) and codemod_findings_total (by severity and rule), and its histograms are codemod_pass_duration_seconds and codemod_commit_duration_seconds. _build_metrics creates the instruments in the registry passed in; when the Prometheus client is not installed it returns no-op counters and histograms, so the observer still works. The workspace uses it among its default observers.

At a glance

Class diagram

Dependency diagram

API

Classes

_NoopCounter

Line 87

Fallback when prometheus_client is not installed.

_NoopHistogram

Line 97

Fallback when prometheus_client is not installed.

MetricsObserver

Decorators: traces(spec='110', story='US-110-005', reqs=['REQ-110-026']) · Line 124

Records Prometheus-compatible counters and histograms for codemod events.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/observability/metrics.py

"""MetricsObserver — Prometheus counters and histograms for codemod operations.

SPEC-110 §15.4: Records counters (passes run, files mutated, findings) and
histograms (pass duration, commit duration, workspace load) using the
``prometheus_client`` library.
"""

from __future__ import annotations

from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.refactor.codemod.observability.events import (
    CodemodEvent,
    CodemodEventType,
)

if TYPE_CHECKING:
    from prometheus_client import CollectorRegistry, Counter, Histogram
else:
    try:
        from prometheus_client import CollectorRegistry, Counter, Histogram
    except ImportError:  # pragma: no cover - prometheus-client is a hard dependency
        CollectorRegistry = None
        Counter = None
        Histogram = None

_PASS_DURATION_BUCKETS = (0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0)
_COMMIT_DURATION_BUCKETS = (0.1, 0.5, 1.0, 5.0, 10.0, 30.0, 60.0)
_LOAD_DURATION_BUCKETS = (1.0, 5.0, 10.0, 30.0, 60.0, 120.0)


def _build_metrics(
    registry: CollectorRegistry | None,
) -> tuple[
    CounterInstrument,
    CounterInstrument,
    CounterInstrument,
    HistogramInstrument,
    HistogramInstrument,
]:
    """Build Prometheus metrics instruments, returning no-op fallbacks if unavailable."""
    if Counter is None or Histogram is None:
        return (
            _NoopCounter(),
            _NoopCounter(),
            _NoopCounter(),
            _NoopHistogram(),
            _NoopHistogram(),
        )

    reg = registry
    passes = Counter(
        "codemod_passes_run_total",
        "Total codemod transform/review passes executed",
        ["pass_name", "pass_type"],
        registry=reg,
    )
    files = Counter(
        "codemod_files_mutated_total",
        "Total files mutated by codemod passes",
        ["pass_name"],
        registry=reg,
    )
    findings = Counter(
        "codemod_findings_total",
        "Total review findings emitted",
        ["severity", "rule"],
        registry=reg,
    )
    pass_dur = Histogram(
        "codemod_pass_duration_seconds",
        "Duration of individual codemod passes",
        ["pass_name"],
        buckets=_PASS_DURATION_BUCKETS,
        registry=reg,
    )
    commit_dur = Histogram(
        "codemod_commit_duration_seconds",
        "Duration of codemod commit (write-back) phase",
        buckets=_COMMIT_DURATION_BUCKETS,
        registry=reg,
    )
    return passes, files, findings, pass_dur, commit_dur


class _NoopCounter:
    """Fallback when prometheus_client is not installed."""

    def labels(self, **_kwargs: str) -> _NoopCounter:
        return self

    def inc(self, amount: float = 1) -> None:
        pass


class _NoopHistogram:
    """Fallback when prometheus_client is not installed."""

    def labels(self, **_kwargs: str) -> _NoopHistogram:
        return self

    def observe(self, value: float) -> None:
        pass


if TYPE_CHECKING:
    # The instrument slots hold EITHER a real prometheus instrument or the
    # no-op fallback below, so they are typed as that union rather than as
    # `object`. `object` type-checks the assignment and then loses every
    # method: `self._passes.labels(...)` became "object has no attribute
    # labels" -- 16 errors across this file and debug_panel.py.
    #
    # A hand-written Protocol standing in for Counter/Histogram does NOT work
    # here and was tried: structural matching compares PARAMETER NAMES, and
    # prometheus's real signature is `observe(amount, exemplar=None)` while
    # the natural stand-in writes `observe(value)`. The union needs no such
    # guess -- both members are concrete types that already exist.
    CounterInstrument = Counter | _NoopCounter
    HistogramInstrument = Histogram | _NoopHistogram


@traces(spec="110", story="US-110-005", reqs=["REQ-110-026"])
class MetricsObserver:
    """Records Prometheus-compatible counters and histograms for codemod events.

    Counters:
        - ``codemod_passes_run_total{pass_name, pass_type}``
        - ``codemod_files_mutated_total{pass_name}``
        - ``codemod_findings_total{severity, rule}``

    Histograms:
        - ``codemod_pass_duration_seconds{pass_name}``
        - ``codemod_commit_duration_seconds``
    """

    def __init__(self, registry: CollectorRegistry | None = None) -> None:
        self._passes: CounterInstrument
        self._files: CounterInstrument
        self._findings: CounterInstrument
        self._pass_dur: HistogramInstrument
        self._commit_dur: HistogramInstrument
        (
            self._passes,
            self._files,
            self._findings,
            self._pass_dur,
            self._commit_dur,
        ) = _build_metrics(registry)

    def on_event(self, event: CodemodEvent) -> None:
        """Record metrics for *event*."""
        et = event.type
        data = event.data

        if et == CodemodEventType.PASS_COMPLETED:
            pass_name = data.get("pass_name", "unknown")
            pass_type = data.get("pass_type", "transform")
            self._passes.labels(pass_name=pass_name, pass_type=pass_type).inc()
            file_count = data.get("file_count", 0)
            if file_count:
                self._files.labels(pass_name=pass_name).inc(file_count)
            dur = data.get("duration_seconds", 0)
            if dur:
                self._pass_dur.labels(pass_name=pass_name).observe(dur)

        elif et == CodemodEventType.REVIEW_COMPLETED:
            finding_count = data.get("finding_count", 0)
            error_count = data.get("error_count", 0)
            warning_count = data.get("warning_count", 0)
            if error_count:
                self._findings.labels(severity="error", rule="review").inc(error_count)
            if warning_count:
                self._findings.labels(severity="warning", rule="review").inc(
                    warning_count
                )
            if finding_count and not error_count and not warning_count:
                self._findings.labels(severity="info", rule="review").inc(finding_count)

        elif et == CodemodEventType.COMMIT_COMPLETED:
            dur = data.get("duration_seconds", 0)
            if dur:
                self._commit_dur.observe(dur)

        elif et == CodemodEventType.WORKSPACE_LOADED:
            pass

        elif et == CodemodEventType.ERROR:
            self._findings.labels(severity="error", rule="runtime").inc()