Expose scrapeable OVOS runtime stage histograms

docs/index.md

@@ -15,6 +15,7 @@
 | [converse-fallback.md](converse-fallback.md) | `ConverseService` and `FallbackService` |
 | [skill-installer.md](skill-installer.md) | `SkillsStore`: runtime pip install/uninstall via the bus |
 | [bus-events.md](bus-events.md) | MessageBus events reference |
+| [performance-metrics.md](performance-metrics.md) | Opt-in runtime stage histograms and multi-process aggregation |
 | [prerelease-quirks.md](prerelease-quirks.md) | What changed since the last stable release |
 
 ## Quick Start

docs/performance-metrics.md

@@ -0,0 +1,87 @@
+# Runtime Performance Metrics
+
+`ovos-core` can expose process-local, fixed-cardinality Prometheus histograms
+for the synchronous utterance path. The endpoint is disabled by default and
+binds to loopback unless explicitly configured.
+
+```text
+OVOS_METRICS_ENABLED=true
+OVOS_METRICS_HOST=127.0.0.1
+OVOS_METRICS_PORT=9474
+```
+
+Scrape `GET /metrics`. Do not expose this operational endpoint through the
+public voice API or WebSocket ingress. The endpoint has no authentication. For
+remote scraping, place it behind an authenticating reverse proxy or restrict
+access with a network policy. Do not bind it to `0.0.0.0` on an untrusted
+network.
+
+## Core stages
+
+| Metric | Boundary |
+|---|---|
+| `ovos_utterance_dispatch_seconds` | Complete synchronous handling of one `recognizer_loop:utterance` message by `IntentService` |
+| `ovos_utterance_preprocess_seconds` | Utterance and metadata transforms, language selection, and session validation before matching |
+| `ovos_utterance_transform_seconds` | Utterance and metadata transformer plugin chains |
+| `ovos_language_resolution_seconds` | Resolve the request language against the enabled language set |
+| `ovos_session_validation_seconds` | Fold and validate the message session before matching |
+| `ovos_session_stamp_seconds` | Serialize the validated session back onto the in-process message |
+| `ovos_skill_selection_seconds` | Selection loop across the configured intent pipelines |
+| `ovos_intent_pipeline_build_seconds` | Resolve the session's configured matcher functions before invoking them |
+| `ovos_intent_matching_seconds` | One pipeline matcher invocation; an utterance can produce more than one observation |
+| `ovos_intent_matching_{family}_seconds` | One matcher invocation classified into the fixed `stop`, `converse`, `padatious`, `padacioso`, `adapt`, `common_query`, `ocp`, `m2v`, `fallback`, or `other` family |
+| `ovos_intent_dispatch_seconds` | Post-match transformation, activation, lifecycle emission, and handler scheduling for a matched utterance |
+| `ovos_intent_transform_seconds` | Intent transformer plugin chain after a successful match |
+| `ovos_intent_activation_seconds` | Update active-handler state and emit the selected skill activation event |
+| `ovos_intent_matched_emit_seconds` | Build and emit the public intent-matched notification |
+| `ovos_intent_handler_schedule_seconds` | Register the in-flight lifecycle and emit handler-start plus the selected skill dispatch |
+| `ovos_handler_timeout_arm_seconds` | Register the in-flight dispatch and arm its bounded timeout |
+| `ovos_handler_start_emit_seconds` | Emit the handler-start lifecycle event |
+| `ovos_handler_dispatch_emit_seconds` | Emit the selected skill dispatch message |
+| `ovos_utterance_finalize_seconds` | Session synchronization and per-utterance deactivation cleanup after selection |
+| `ovos_converse_prepare_seconds` | Normalize the language and inspect session response-mode candidates inside the converse matcher |
+| `ovos_converse_poll_seconds` | Prune stale converse owners and collect their bounded capability replies |
+| `ovos_converse_policy_seconds` | Apply blacklist and converse policy checks to the owners that accepted a poll |
+
+The boundaries are nested: utterance dispatch contains preprocessing,
+selection, matched-intent dispatch, and finalization; selection contains
+pipeline construction and one or more matcher observations; matched-intent
+dispatch contains handler scheduling. Plugin-provided handler observations can
+contain further plugin-provided stages. Do not add nested durations as if they
+were disjoint stages.
+
+Installed packages can contribute histograms and cumulative counters through the
+`ovos.performance.metrics` entry-point group. A collector is a zero-argument
+callable returning fixed metric snapshots. Histograms provide `count`,
+`sum_ms`, and cumulative `buckets`; counters provide `type: counter` and an
+numeric `value`, and their names end in `_total`. Collectors are loaded once at
+startup; duplicate or malformed metric names make a scrape fail rather than
+silently publishing misleading data.
+
+## Aggregating partitions
+
+Prometheus must scrape every runtime process. Each process publishes its own
+counters. Aggregate buckets across all scraped processes before calculating a
+percentile:
+
+```promql
+histogram_quantile(
+  0.95,
+  sum by (le) (rate(ovos_utterance_dispatch_seconds_bucket[5m]))
+)
+```
+
+Use the same query shape for dispatch, matching, selection, dialog rendering,
+and weather-service request histograms. Keep process and pod labels for a
+second view when checking shard skew:
+
+```promql
+histogram_quantile(
+  0.95,
+  sum by (pod, le) (rate(ovos_utterance_dispatch_seconds_bucket[5m]))
+)
+```
+
+These histograms are cumulative and reset when the process restarts. They use
+no session, client, utterance, or skill labels, avoiding unbounded cardinality
+and user-content leakage.

ovos_core/__main__.py

@@ -23,6 +23,7 @@
 from ovos_utils import wait_for_exit_signal
 from ovos_utils.log import LOG, init_service_logger
 
+from ovos_core._prometheus import start_metrics_server, stop_metrics_server
 from ovos_core.skill_manager import SkillManager, on_error, on_stopping, on_ready, on_alive, on_started
 
 
@@ -42,6 +43,10 @@ def main(alive_hook=on_alive, started_hook=on_started, ready_hook=on_ready,
 
     setup_locale()
 
+    # Opt-in scrape endpoint for the runtime stage histograms; a no-op
+    # unless OVOS_METRICS_ENABLED is set (see ovos_core._prometheus).
+    metrics_server = start_metrics_server()
+
     # Connect this process to the OpenVoiceOS message bus
     bus = MessageBusClient()
     bus_thread = bus.run_in_thread()
@@ -65,6 +70,8 @@ def main(alive_hook=on_alive, started_hook=on_started, ready_hook=on_ready,
 
     skill_manager.shutdown()
 
+    stop_metrics_server(metrics_server)
+
     # Stop the messagebus websocket thread and its event dispatcher before
     # the interpreter starts tearing down. `bus.run_in_thread()` spawns a
     # daemon thread that keeps receiving messages and dispatching them onto

ovos_core/_metrics.py

@@ -0,0 +1,222 @@
+"""Low-overhead, fixed-cardinality runtime latency histograms."""
+
+from __future__ import annotations
+
+import math
+import time
+from collections.abc import Callable, Iterable, Iterator, Mapping
+from contextlib import contextmanager
+from functools import wraps
+from threading import Lock
+from typing import Any, ParamSpec, TypeVar
+
+DEFAULT_BUCKETS_MS = (
+    1.0,
+    2.5,
+    5.0,
+    10.0,
+    25.0,
+    50.0,
+    100.0,
+    250.0,
+    500.0,
+    1_000.0,
+    2_500.0,
+    5_000.0,
+    10_000.0,
+    30_000.0,
+)
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+class _LatencyMeasurement:
+    """A pausable, single-observation histogram measurement."""
+
+    def __init__(self, histogram: "LatencyHistogram") -> None:
+        self._histogram = histogram
+        self._started = time.monotonic()
+        self._elapsed_ms = 0.0
+        self._running = True
+        self._finished = False
+
+    def pause(self) -> None:
+        """Exclude subsequent time until :meth:`resume` is called."""
+        if self._running and not self._finished:
+            self._elapsed_ms += (time.monotonic() - self._started) * 1_000
+            self._running = False
+
+    def resume(self) -> None:
+        """Resume measuring after a pause."""
+        if not self._running and not self._finished:
+            self._started = time.monotonic()
+            self._running = True
+
+    def finish(self) -> None:
+        """Observe accumulated active time exactly once."""
+        if self._finished:
+            return
+        self.pause()
+        self._finished = True
+        self._histogram.observe_ms(self._elapsed_ms)
+
+
+class LatencyHistogram:
+    """Thread-safe cumulative latency histogram with fixed buckets."""
+
+    def __init__(self, name: str, *,
+                 buckets_ms: Iterable[float] = DEFAULT_BUCKETS_MS) -> None:
+        self.name = name
+        self._bounds = tuple(sorted(float(value) for value in buckets_ms))
+        self._buckets = [0] * len(self._bounds)
+        self._count = 0
+        self._sum_ms = 0.0
+        self._lock = Lock()
+
+    def observe_ms(self, elapsed_ms: float) -> None:
+        """Record one finite, non-negative duration in milliseconds."""
+        value = float(elapsed_ms)
+        if not math.isfinite(value):
+            raise ValueError("elapsed_ms must be finite")
+        value = max(0.0, value)
+        with self._lock:
+            self._count += 1
+            self._sum_ms += value
+            for index, bound in enumerate(self._bounds):
+                if value <= bound:
+                    self._buckets[index] += 1
+
+    @contextmanager
+    def measure(self) -> Iterator[_LatencyMeasurement]:
+        """Observe active enclosed time, including exceptional exits."""
+        measurement = _LatencyMeasurement(self)
+        try:
+            yield measurement
+        finally:
+            measurement.finish()
+
+    def timed(self, function: Callable[P, R]) -> Callable[P, R]:
+        """Decorate a synchronous function with this histogram."""
+        @wraps(function)
+        def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
+            with self.measure():
+                return function(*args, **kwargs)
+
+        return wrapped
+
+    def snapshot(self) -> Mapping[str, Any]:
+        """Return an immutable, JSON-friendly cumulative snapshot."""
+        with self._lock:
+            buckets = {
+                f"le_{bound:g}": count
+                for bound, count in zip(
+                    self._bounds, self._buckets, strict=True
+                )
+            }
+            buckets["inf"] = self._count
+            return {
+                "name": self.name,
+                "count": self._count,
+                "sum_ms": self._sum_ms,
+                "buckets": buckets,
+            }
+
+
+UTTERANCE_DISPATCH = LatencyHistogram("ovos_utterance_dispatch_ms")
+UTTERANCE_PREPROCESS = LatencyHistogram("ovos_utterance_preprocess_ms")
+UTTERANCE_TRANSFORM = LatencyHistogram("ovos_utterance_transform_ms")
+LANGUAGE_RESOLUTION = LatencyHistogram("ovos_language_resolution_ms")
+SESSION_VALIDATION = LatencyHistogram("ovos_session_validation_ms")
+SESSION_STAMP = LatencyHistogram("ovos_session_stamp_ms")
+INTENT_MATCHING = LatencyHistogram("ovos_intent_matching_ms")
+INTENT_PIPELINE_BUILD = LatencyHistogram("ovos_intent_pipeline_build_ms")
+SKILL_SELECTION = LatencyHistogram("ovos_skill_selection_ms")
+INTENT_DISPATCH = LatencyHistogram("ovos_intent_dispatch_ms")
+INTENT_TRANSFORM = LatencyHistogram("ovos_intent_transform_ms")
+INTENT_ACTIVATION = LatencyHistogram("ovos_intent_activation_ms")
+INTENT_MATCHED_EMIT = LatencyHistogram("ovos_intent_matched_emit_ms")
+INTENT_HANDLER_SCHEDULE = LatencyHistogram(
+    "ovos_intent_handler_schedule_ms"
+)
+HANDLER_TIMEOUT_ARM = LatencyHistogram("ovos_handler_timeout_arm_ms")
+HANDLER_START_EMIT = LatencyHistogram("ovos_handler_start_emit_ms")
+HANDLER_DISPATCH_EMIT = LatencyHistogram("ovos_handler_dispatch_emit_ms")
+UTTERANCE_FINALIZE = LatencyHistogram("ovos_utterance_finalize_ms")
+CONVERSE_PREPARE = LatencyHistogram("ovos_converse_prepare_ms")
+CONVERSE_POLL = LatencyHistogram("ovos_converse_poll_ms")
+CONVERSE_POLICY = LatencyHistogram("ovos_converse_policy_ms")
+
+# Pipeline identifiers are session-selectable, so they must never become raw
+# metric names or labels.  These families cover the built-in matchers while an
+# explicit ``other`` bucket keeps third-party plugins observable without
+# unbounded cardinality.
+_PIPELINE_FAMILIES = (
+    "stop",
+    "converse",
+    "padatious",
+    "padacioso",
+    "adapt",
+    "common_query",
+    "ocp",
+    "m2v",
+    "fallback",
+    "other",
+)
+PIPELINE_MATCHING = {
+    family: LatencyHistogram(f"ovos_intent_matching_{family}_ms")
+    for family in _PIPELINE_FAMILIES
+}
+_PIPELINE_PREFIXES = (
+    ("ovos-stop-pipeline", "stop"),
+    ("ovos-converse-pipeline", "converse"),
+    ("ovos-padatious-pipeline", "padatious"),
+    ("ovos-padacioso-pipeline", "padacioso"),
+    ("ovos-adapt-pipeline", "adapt"),
+    ("ovos-common-query-pipeline", "common_query"),
+    ("ovos-ocp-pipeline", "ocp"),
+    ("ovos-m2v-pipeline", "m2v"),
+    ("ovos-fallback-pipeline", "fallback"),
+)
+
+
+def pipeline_matching_histogram(pipeline_id: str) -> LatencyHistogram:
+    """Return the fixed-cardinality histogram for ``pipeline_id``."""
+    normalized = str(pipeline_id).lower().replace("_", "-")
+    family = next(
+        (family for prefix, family in _PIPELINE_PREFIXES
+         if normalized.startswith(prefix)),
+        "other",
+    )
+    return PIPELINE_MATCHING[family]
+
+
+def performance_histograms() -> Mapping[str, Mapping[str, Any]]:
+    """Return the process-local Core runtime histograms."""
+    return {
+        histogram.name: histogram.snapshot()
+        for histogram in (
+            UTTERANCE_DISPATCH,
+            UTTERANCE_PREPROCESS,
+            UTTERANCE_TRANSFORM,
+            LANGUAGE_RESOLUTION,
+            SESSION_VALIDATION,
+            SESSION_STAMP,
+            INTENT_MATCHING,
+            INTENT_PIPELINE_BUILD,
+            SKILL_SELECTION,
+            INTENT_DISPATCH,
+            INTENT_TRANSFORM,
+            INTENT_ACTIVATION,
+            INTENT_MATCHED_EMIT,
+            INTENT_HANDLER_SCHEDULE,
+            HANDLER_TIMEOUT_ARM,
+            HANDLER_START_EMIT,
+            HANDLER_DISPATCH_EMIT,
+            UTTERANCE_FINALIZE,
+            CONVERSE_PREPARE,
+            CONVERSE_POLL,
+            CONVERSE_POLICY,
+            *PIPELINE_MATCHING.values(),
+        )
+    }