geno_lewm.metrics

metrics

GenoLeWM metrics registry and minimal Prometheus textfile exporter.

Defined by RFC-0013 §3.3 / 4 and docs/spec/05-observability.md. Provides:

• :data:METRICS — immutable registry of MetricSpec(name, kind, unit, summary) rows. Renaming a metric name is a MAJOR change.
• :class:Counter, :class:Gauge, :class:Histogram — minimal, thread-safe primitives with no external dependency.
• :func:get_counter / :func:get_gauge / :func:get_histogram — validated accessors. The accessor verifies the metric is registered and has the expected kind, then returns a cached instance.
• :func:export_prometheus_textfile — writes a Prometheus textfile exposition to ${GENO_LEWM_LOG_DIR}/metrics.prom (or a path of the caller's choosing). Honours the standard # HELP / # TYPE lines and the histogram bucket / sum / count expansion.

HistogramSnapshot

Bases: TypedDict

Structured snapshot returned by :meth:Histogram.snapshot.

Buckets are cumulative upper bounds (the last bucket is always +Inf). counts[i] is the cumulative count of observations <= buckets[i] since the last :meth:Histogram.reset.

MetricSpec dataclass

MetricSpec(name: str, kind: MetricKind, unit: str, summary: str)

A single row in the :data:METRICS registry.

Counter

Counter(spec: MetricSpec)

Bases: _Metric

Monotonically non-decreasing total.

Source code in geno_lewm/metrics.py

def __init__(self, spec: MetricSpec) -> None:
    super().__init__(spec)
    self._value: float = 0.0

reset

reset() -> None

Reset the counter. Used in tests; not part of the public production contract.

Source code in geno_lewm/metrics.py

def reset(self) -> None:
    """Reset the counter. Used in tests; not part of the public
    production contract."""
    with self._lock:
        self._value = 0.0

Gauge

Gauge(spec: MetricSpec)

Bases: _Metric

Settable scalar.

Source code in geno_lewm/metrics.py

def __init__(self, spec: MetricSpec) -> None:
    super().__init__(spec)
    self._value: float = 0.0

Histogram

Histogram(spec: MetricSpec, buckets: Iterable[float] | None = None)

Bases: _Metric

Bucketed observation distribution.

Prometheus-compatible: the exposition emits <name>_bucket{le=…} rows, <name>_sum, and <name>_count. Buckets are cumulative by spec — the last bucket is always +Inf.

Source code in geno_lewm/metrics.py

def __init__(
    self,
    spec: MetricSpec,
    buckets: Iterable[float] | None = None,
) -> None:
    super().__init__(spec)
    if buckets is None:
        buckets = (
            DEFAULT_HISTOGRAM_BUCKETS_BYTES
            if spec.unit == "bytes"
            else DEFAULT_HISTOGRAM_BUCKETS_MS
        )
    bs = sorted(float(b) for b in buckets)
    if not bs:
        raise InputError(
            "histogram buckets must be non-empty",
            details={"name": spec.name, "buckets": list(bs)},
        )
    if bs[-1] != float("inf"):
        bs.append(float("inf"))
    self._buckets: tuple[float, ...] = tuple(bs)
    self._counts: list[int] = [0] * len(self._buckets)
    self._sum: float = 0.0
    self._count: int = 0

snapshot_all

snapshot_all() -> dict[str, Mapping[str, object]]

Return a structured snapshot of every live metric.

Source code in geno_lewm/metrics.py

def snapshot_all() -> dict[str, Mapping[str, object]]:
    """Return a structured snapshot of every live metric."""
    out: dict[str, Mapping[str, object]] = {}
    with _INSTANCES_LOCK:
        instances = list(_INSTANCES.values())
    for inst in instances:
        if isinstance(inst, Counter | Gauge):
            out[inst.name] = {"kind": inst.kind, "value": inst.value()}
        else:
            assert isinstance(inst, Histogram)
            out[inst.name] = {"kind": "histogram", **inst.snapshot()}
    return out

metrics_path

metrics_path(log_dir: str | PathLike[str] | None = None) -> Path

Return the default exporter path.

Source code in geno_lewm/metrics.py

def metrics_path(log_dir: str | os.PathLike[str] | None = None) -> Path:
    """Return the default exporter path."""
    if log_dir is not None:
        base = Path(log_dir).expanduser()
    else:
        env = os.environ.get("GENO_LEWM_LOG_DIR")
        base = Path(env).expanduser() if env else Path.home() / ".geno-lewm" / "logs"
    return base / "metrics.prom"

export_prometheus_textfile

export_prometheus_textfile(path: str | PathLike[str] | None = None) -> Path

Write the current metric snapshot to a Prometheus textfile.

The write is atomic: contents go to <path>.tmp and are renamed over <path> once flushed. Scrapers that read the file mid-flush therefore see either the previous or the new value, never a partial record. Returns the destination path.

Source code in geno_lewm/metrics.py

def export_prometheus_textfile(
    path: str | os.PathLike[str] | None = None,
) -> Path:
    """Write the current metric snapshot to a Prometheus textfile.

    The write is atomic: contents go to ``<path>.tmp`` and are renamed
    over ``<path>`` once flushed. Scrapers that read the file mid-flush
    therefore see either the previous or the new value, never a partial
    record. Returns the destination path.
    """
    dest = Path(path) if path is not None else metrics_path()
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".tmp")
    with _INSTANCES_LOCK:
        instances = list(_INSTANCES.values())
    instances.sort(key=lambda i: i.name)
    with tmp.open("w", encoding="utf-8") as f:
        for inst in instances:
            _write_metric_block(f, inst)
            f.write("\n")
    tmp.replace(dest)
    return dest

sync_redaction_counter

sync_redaction_counter() -> None

Pull the in-process redaction counters into the metric.

The redaction filter from #24 keeps its own thread-local counters (so it can run without the metrics package being imported); this function reconciles the two views right before exposition.

Source code in geno_lewm/metrics.py

def sync_redaction_counter() -> None:
    """Pull the in-process redaction counters into the metric.

    The redaction filter from #24 keeps its own thread-local counters
    (so it can run without the metrics package being imported); this
    function reconciles the two views right before exposition.
    """
    try:
        from geno_lewm._redaction import STATS  # local import: optional
    except Exception:  # pragma: no cover - tested via integration
        return
    c = get_counter("geno_lewm.observability.redacted_keys")
    target = STATS.total()
    # The counter is monotonic; just bring it up to the current total.
    current = c.value()
    if target > current:
        c.inc(target - current)