From b180680427a8297defd6790280641c80a9c63c5f Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 12 Nov 2026 09:46:37 +0100 Subject: [PATCH] swarm-stt: metrics and a Grafana board for the speech-to-text service --- Cargo.lock | 2 + nix/host-modules/swarm-grafana.nix | 1 + .../swarm-grafana/dashboards/stt.json | 441 ++++++++++++++++++ swarm-stt/Cargo.toml | 7 +- swarm-stt/src/main.rs | 56 ++- swarm-stt/src/metrics.rs | 191 ++++++++ 6 files changed, 681 insertions(+), 17 deletions(-) create mode 100644 nix/host-modules/swarm-grafana/dashboards/stt.json create mode 100644 swarm-stt/src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index d20316d18..0edb4a43f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5373,6 +5373,8 @@ dependencies = [ "async-trait", "futures-util", "hive-log", + "opentelemetry", + "opentelemetry_sdk", "serde_json", "sha2 0.11.1", "swarm-queue-client", diff --git a/nix/host-modules/swarm-grafana.nix b/nix/host-modules/swarm-grafana.nix index d872f05c8..6b9611667 100644 --- a/nix/host-modules/swarm-grafana.nix +++ b/nix/host-modules/swarm-grafana.nix @@ -289,6 +289,7 @@ let "notify-router.json" "queue.json" "review-classifier.json" + "stt.json" ] ); diff --git a/nix/host-modules/swarm-grafana/dashboards/stt.json b/nix/host-modules/swarm-grafana/dashboards/stt.json new file mode 100644 index 000000000..fa6ff4168 --- /dev/null +++ b/nix/host-modules/swarm-grafana/dashboards/stt.json @@ -0,0 +1,441 @@ +{ + "title": "hyperhive ยท speech to text", + "uid": "hyperhive-swarm-stt", + "description": "The speech-to-text service: voice memos in, transcripts out, the cache hit rate and the engine's load.", + "tags": ["hyperhive", "swarm", "stt"], + "editable": true, + "refresh": "1m", + "schemaVersion": 39, + "time": { + "from": "now-24h", + "to": "now" + }, + "timezone": "utc", + "version": 1, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Requests in range", + "description": "Voice memos the service received in the selected range.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 0, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.stt.requests\"}[$__range])) or vector(0)", + "instant": true, + "legendFormat": "requests in range" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 2, + "type": "stat", + "title": "Served from cache", + "description": "Share of answered requests that needed no engine run.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 6, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.stt.requests\", outcome=\"cached\"}[$__range])) / sum(increase({__name__=\"hyperhive.stt.requests\", outcome=~\"cached|transcribed\"}[$__range]))", + "instant": true, + "legendFormat": "served from cache" + } + ], + "fieldConfig": { + "defaults": { + "unit": "percentunit", + "decimals": 2, + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 3, + "type": "stat", + "title": "Failed or turned away", + "description": "Requests that failed in the engine or hit a limit, in the range.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 12, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.stt.requests\", outcome=~\"failed|busy\"}[$__range])) or vector(0)", + "instant": true, + "legendFormat": "failed or turned away" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + }, + { + "color": "red", + "value": 1 + } + ] + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 4, + "type": "stat", + "title": "Mean time per run", + "description": "Mean time from taking a request to answering it, in the range.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 4, + "w": 6, + "x": 18, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum(increase({__name__=\"hyperhive.stt.run.duration_sum\"}[$__range])) / sum(increase({__name__=\"hyperhive.stt.run.duration_count\"}[$__range]))", + "instant": true, + "legendFormat": "mean time per run" + } + ], + "fieldConfig": { + "defaults": { + "unit": "s", + "decimals": 2, + "color": { + "mode": "thresholds" + }, + "mappings": [], + "thresholds": { + "mode": "absolute", + "steps": [ + { + "color": "green", + "value": null + } + ] + } + }, + "overrides": [] + }, + "options": { + "graphMode": "none", + "colorMode": "value", + "textMode": "value", + "justifyMode": "auto", + "reduceOptions": { + "calcs": ["lastNotNull"], + "fields": "", + "values": false + } + } + }, + { + "id": 5, + "type": "timeseries", + "title": "Requests by outcome", + "description": "Requests per hour, by what the service did with them.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 4 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "sum by (outcome) (increase({__name__=\"hyperhive.stt.requests\"}[1h]))", + "legendFormat": "{{outcome}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 10, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0, + "stacking": { + "mode": "normal", + "group": "A" + } + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, + { + "id": 6, + "type": "timeseries", + "title": "Time per request", + "description": "Median and 95th percentile seconds from taking a request to answering it.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 4 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "histogram_quantile(0.5, sum by (le) (increase({__name__=\"hyperhive.stt.run.duration_bucket\"}[5m])))", + "legendFormat": "median" + }, + { + "refId": "B", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "histogram_quantile(0.95, sum by (le) (increase({__name__=\"hyperhive.stt.run.duration_bucket\"}[5m])))", + "legendFormat": "p95" + } + ], + "fieldConfig": { + "defaults": { + "unit": "s", + "decimals": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 10, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0 + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, + { + "id": 7, + "type": "timeseries", + "title": "Requests held", + "description": "Requests held at once, running or waiting for the engine's one slot. Pinned at the limit means senders are being turned away.", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 12 + }, + "targets": [ + { + "refId": "A", + "datasource": { + "type": "prometheus", + "uid": "@datasourceUid@" + }, + "expr": "max({__name__=\"hyperhive.stt.held\"})", + "legendFormat": "held" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 10, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0 + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + } + ] +} diff --git a/swarm-stt/Cargo.toml b/swarm-stt/Cargo.toml index 5dc411632..4a8adba21 100644 --- a/swarm-stt/Cargo.toml +++ b/swarm-stt/Cargo.toml @@ -15,13 +15,17 @@ async-trait.workspace = true # Draining the subscription. futures-util.workspace = true hive-log.workspace = true +# The metrics the service exports (`metrics`), pushed through the OTLP exporter +# `swarm-secret-client`'s `meter_provider` builds. +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true serde_json.workspace = true # The SHA-256 of the audio, which keys the cache and the in-flight map. sha2.workspace = true # The queue connect with its OIDC client credentials. swarm-queue-client.workspace = true # The client secret, read with the service's own store leaf. -swarm-secret-client.workspace = true +swarm-secret-client = { workspace = true, features = ["otel"] } # The subject, request and response of the wire. swarm-wire.workspace = true thiserror.workspace = true @@ -30,6 +34,7 @@ tracing.workspace = true [dev-dependencies] # A scratch cache directory per test. +opentelemetry_sdk = { workspace = true, features = ["testing"] } tempfile = "3" # `start_paused` for the test that overlaps two requests. tokio = { workspace = true, features = ["test-util"] } diff --git a/swarm-stt/src/main.rs b/swarm-stt/src/main.rs index 50d6542b3..23921c291 100644 --- a/swarm-stt/src/main.rs +++ b/swarm-stt/src/main.rs @@ -10,10 +10,11 @@ use std::collections::HashSet; use std::path::PathBuf; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use anyhow::{Context as _, Result, bail}; use futures_util::StreamExt as _; +use metrics::{Metrics, Outcome}; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::stored_secret::StoredSecret; use swarm_stt::engine::Whisper; @@ -21,6 +22,8 @@ use swarm_stt::{Config, Error, Service, cache::Cache}; use swarm_wire::stt::{Response, subject_wildcard}; use tokio::sync::Semaphore; +mod metrics; + /// The cert-auth role this service logs in to the store as. `swarm-bao.nix` /// writes it, matching the leaf's subject. const CERT_ROLE: &str = "swarm-stt"; @@ -133,6 +136,7 @@ async fn main() -> Result<()> { } = settings(&var, &number)?; let per_agent: Arc>> = Arc::default(); let queue = connect_queue(&var).await?; + let metrics = Arc::new(Metrics::from_env()); tokio::spawn(async move { loop { @@ -154,6 +158,7 @@ async fn main() -> Result<()> { ); while let Some(request) = requests.next().await { let (queue, service) = (queue.clone(), Arc::clone(&service)); + let metrics = Arc::clone(&metrics); // Held at once overall, and one per asking agent, so one agent // cannot fill every slot. The agent is the subject's last token. let agent = request @@ -178,13 +183,19 @@ async fn main() -> Result<()> { }; let response = match turned_away { Ok(held) => { - let response = answer(&service, &request.payload).await; - drop(held); + let count = metrics.hold(); + let started = Instant::now(); + let (response, outcome) = answer(&service, &request.payload).await; + metrics.request(outcome, Some(started.elapsed())); + drop((held, count)); response } - Err(message) => Response::Failed { - message: message.to_owned(), - }, + Err(message) => { + metrics.request(Outcome::Busy, None); + Response::Failed { + message: message.to_owned(), + } + } }; match serde_json::to_vec(&response) { Ok(body) => { @@ -226,20 +237,33 @@ fn take_slot(agents: &Arc>>, agent: &str) -> Option }) } -/// The wire answer to one request's audio. -async fn answer(service: &Service, audio: &[u8]) -> Response { +/// The wire answer to one request's audio, and what came of it. +async fn answer(service: &Service, audio: &[u8]) -> (Response, Outcome) { match service.transcribe(audio).await { - Ok(reply) => Response::Ok { - text: reply.text, - cached: reply.cached, - }, + Ok(reply) => { + let outcome = if reply.cached { + Outcome::Cached + } else { + Outcome::Transcribed + }; + ( + Response::Ok { + text: reply.text, + cached: reply.cached, + }, + outcome, + ) + } Err(Error::Engine(message)) => { tracing::warn!(%message, "a transcription failed"); - Response::Failed { message } + (Response::Failed { message }, Outcome::Failed) } - Err(e) => Response::Failed { - message: e.to_string(), - }, + Err(e) => ( + Response::Failed { + message: e.to_string(), + }, + Outcome::Refused, + ), } } diff --git a/swarm-stt/src/metrics.rs b/swarm-stt/src/metrics.rs new file mode 100644 index 000000000..03a201286 --- /dev/null +++ b/swarm-stt/src/metrics.rs @@ -0,0 +1,191 @@ +//! What the service exports for its dashboard, through the swarm collector. +//! +//! - `hyperhive.stt.requests`: requests answered, by `outcome`: `transcribed` +//! (an engine run for this request), `cached` (no run), `refused` (empty, +//! too large, or not audio), `failed` (the engine failed), `busy` (turned +//! away at a limit). +//! - `hyperhive.stt.run.duration`: seconds from taking a request to answering +//! it, by `outcome`, for the requests that reached the engine or its cache. +//! - `hyperhive.stt.held`: requests held right now, running or waiting for the +//! engine's one slot. Close to `MAX_WAITING` for long means senders are +//! being turned away. +//! +//! The provider is built from the standard `OTEL_*` environment, so without a +//! collector the instruments do nothing. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider as _}; +use opentelemetry_sdk::metrics::SdkMeterProvider; + +const METER: &str = "hyperhive.stt"; + +/// Seconds. A cache hit takes milliseconds; a run takes seconds to minutes. +const DURATION_BOUNDARIES: [f64; 12] = [ + 0.01, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0, 120.0, 240.0, +]; + +/// What came of one request. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Outcome { + /// An engine run for this request produced the text. + Transcribed, + /// The text was already there. + Cached, + /// The audio was empty, too large, or not an audio container. + Refused, + /// The engine failed. + Failed, + /// Turned away at a limit. + Busy, +} + +impl Outcome { + const fn name(self) -> &'static str { + match self { + Self::Transcribed => "transcribed", + Self::Cached => "cached", + Self::Refused => "refused", + Self::Failed => "failed", + Self::Busy => "busy", + } + } +} + +/// The service's instruments, and the provider that exports them. +pub struct Metrics { + /// Kept alive: it exports only while it lives. + _provider: Option, + requests: Counter, + duration: Histogram, + held: Arc, +} + +/// One request held; dropping it releases the count. +pub struct Held(Arc); + +impl Drop for Held { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::Relaxed); + } +} + +impl Metrics { + /// The instruments on the provider the environment names, or no-op ones. + #[must_use] + pub fn from_env() -> Self { + let provider = swarm_secret_client::meter_provider::from_env(METER, "stt"); + let meter = provider + .as_ref() + .map_or_else(|| opentelemetry::global::meter(METER), |p| p.meter(METER)); + Self::on(&meter, provider) + } + + fn on(meter: &Meter, provider: Option) -> Self { + let held = Arc::new(AtomicU64::new(0)); + let observed = Arc::clone(&held); + meter + .u64_observable_gauge(format!("{METER}.held")) + .with_callback(move |obs| obs.observe(observed.load(Ordering::Relaxed), &[])) + .build(); + Self { + _provider: provider, + requests: meter.u64_counter(format!("{METER}.requests")).build(), + duration: meter + .f64_histogram(format!("{METER}.run.duration")) + .with_unit("s") + .with_boundaries(DURATION_BOUNDARIES.to_vec()) + .build(), + held, + } + } + + /// One request answered, after `took` (`None` for a request turned away + /// before any work). + pub fn request(&self, outcome: Outcome, took: Option) { + let outcome = KeyValue::new("outcome", outcome.name()); + self.requests.add(1, std::slice::from_ref(&outcome)); + if let Some(took) = took { + self.duration.record(took.as_secs_f64(), &[outcome]); + } + } + + /// A request is held until the returned value drops. + #[must_use] + pub fn hold(&self) -> Held { + self.held.fetch_add(1, Ordering::Relaxed); + Held(Arc::clone(&self.held)) + } +} + +#[cfg(test)] +mod tests { + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader}; + + use super::*; + + fn recorded() -> (Metrics, SdkMeterProvider, InMemoryMetricExporter) { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let metrics = Metrics::on(&provider.meter("test"), None); + (metrics, provider, exporter) + } + + /// The `outcome` label and value of every point of the counter `name`. + fn counts(exporter: &InMemoryMetricExporter, name: &str) -> Vec<(String, u64)> { + let mut out = Vec::new(); + for resource in exporter.get_finished_metrics().unwrap() { + for scope in resource.scope_metrics() { + for metric in scope.metrics().filter(|m| m.name() == name) { + if let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() { + for point in sum.data_points() { + let outcome = point + .attributes() + .find(|kv| kv.key.as_str() == "outcome") + .map(|kv| kv.value.to_string()) + .unwrap_or_default(); + out.push((outcome, point.value())); + } + } + } + } + } + out.sort(); + out + } + + #[test] + fn requests_are_counted_by_outcome() { + let (metrics, provider, exporter) = recorded(); + metrics.request(Outcome::Transcribed, Some(Duration::from_secs(3))); + metrics.request(Outcome::Cached, Some(Duration::from_millis(2))); + metrics.request(Outcome::Cached, Some(Duration::from_millis(3))); + metrics.request(Outcome::Busy, None); + provider.force_flush().unwrap(); + assert_eq!( + counts(&exporter, "hyperhive.stt.requests"), + [ + ("busy".to_owned(), 1), + ("cached".to_owned(), 2), + ("transcribed".to_owned(), 1) + ] + ); + } + + #[test] + fn a_held_request_counts_until_it_drops() { + let (metrics, _provider, _exporter) = recorded(); + let first = metrics.hold(); + let second = metrics.hold(); + assert_eq!(metrics.held.load(Ordering::Relaxed), 2); + drop(first); + assert_eq!(metrics.held.load(Ordering::Relaxed), 1); + drop(second); + assert_eq!(metrics.held.load(Ordering::Relaxed), 0); + } +}