hyperhive/swarm-stt/src/metrics.rs

//! 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);
    }
}