Metrics

GenoLeWM ships a registered metrics registry with a Prometheus textfile exporter, defined by RFC-0013 §4. The registry is the single source of truth: new metrics MUST be added to geno_lewm.metrics.METRICS and the AST linter (check_event_names.py) prevents call-site drift.

Registry

Primitives

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.

MetricKind module-attribute

MetricKind = Literal['counter', 'gauge', 'histogram']

DEFAULT_HISTOGRAM_BUCKETS_MS module-attribute

DEFAULT_HISTOGRAM_BUCKETS_MS: tuple[float, ...] = (0.5, 1, 2, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, float('inf'))

DEFAULT_HISTOGRAM_BUCKETS_BYTES module-attribute

DEFAULT_HISTOGRAM_BUCKETS_BYTES: tuple[float, ...] = (1000000.0, 10000000.0, 100000000.0, 500000000.0, 1000000000.0, 5000000000.0, 10000000000.0, 50000000000.0, 100000000000.0, float('inf'))

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

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.

Accessors

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.

get_counter

get_counter(name: str) -> Counter

Source code in geno_lewm/metrics.py

def get_counter(name: str) -> Counter:
    spec = _require_spec(name, "counter")
    with _INSTANCES_LOCK:
        instance = _INSTANCES.get(name)
        if instance is None:
            instance = Counter(spec)
            _INSTANCES[name] = instance
        if not isinstance(instance, Counter):  # pragma: no cover - kind already checked
            raise InvariantViolation(
                "registered metric is not a Counter",
                details={"name": name, "actual_kind": type(instance).__name__},
            )
        return instance

get_gauge

get_gauge(name: str) -> Gauge

Source code in geno_lewm/metrics.py

def get_gauge(name: str) -> Gauge:
    spec = _require_spec(name, "gauge")
    with _INSTANCES_LOCK:
        instance = _INSTANCES.get(name)
        if instance is None:
            instance = Gauge(spec)
            _INSTANCES[name] = instance
        if not isinstance(instance, Gauge):  # pragma: no cover
            raise InvariantViolation(
                "registered metric is not a Gauge",
                details={"name": name, "actual_kind": type(instance).__name__},
            )
        return instance

get_histogram

get_histogram(name: str, *, buckets: Iterable[float] | None = None) -> Histogram

Source code in geno_lewm/metrics.py

def get_histogram(name: str, *, buckets: Iterable[float] | None = None) -> Histogram:
    spec = _require_spec(name, "histogram")
    with _INSTANCES_LOCK:
        instance = _INSTANCES.get(name)
        if instance is None:
            instance = Histogram(spec, buckets=buckets)
            _INSTANCES[name] = instance
        if not isinstance(instance, Histogram):  # pragma: no cover
            raise InvariantViolation(
                "registered metric is not a Histogram",
                details={"name": name, "actual_kind": type(instance).__name__},
            )
        return instance

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

Prometheus exporter

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

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"

Atomicity guarantee: the exporter writes to a *.tmp file and renames on top of the destination, so scrapers always observe either the previous or the new value, never a torn record.