swarm-stt: metrics and a Grafana board for the speech-to-text service

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<Mutex<HashSet<String>>> = 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<Mutex<HashSet<String>>>, agent: &str) -> Option<Slot>
     })
 }
 
-/// The wire answer to one request's audio.
-async fn answer(service: &Service<Whisper>, audio: &[u8]) -> Response {
+/// The wire answer to one request's audio, and what came of it.
+async fn answer(service: &Service<Whisper>, 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,
+        ),
     }
 }
 

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<SdkMeterProvider>,
+    requests: Counter<u64>,
+    duration: Histogram<f64>,
+    held: Arc<AtomicU64>,
+}
+
+/// One request held; dropping it releases the count.
+pub struct Held(Arc<AtomicU64>);
+
+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<SdkMeterProvider>) -> 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<Duration>) {
+        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);
+    }
+}