hyperhive/hive-c0re/src/stats/otel_metrics.rs

//! Per-agent container-resource OTEL export. hive-c0re already samples each
//! agent container's cgroup load for the dashboard
//! ([`super::container_stats`]); this rides those same gauges out to the
//! hive-wide OTLP endpoint, reusing the SAME `services.hyperhive.otel` config
//! (endpoint + auth header) that Claude Code's in-container SDK export uses —
//! no new toggle. The auth header is loaded onto hive-c0re's own unit via
//! systemd `LoadCredential` (see `nix/host-modules/hive-c0re`) and read from
//! `$CREDENTIALS_DIRECTORY/otel-headers`.
//!
//! Emits the OTEL **semconv `container.*`** metrics with the standard
//! `container.name` attribute (so off-the-shelf OTEL/Grafana container
//! dashboards work), plus the hive-specific `agent` / `hive` / `swarm` labels
//! for our own dashboards. Uses the OpenTelemetry Rust SDK (same crates as
//! `hive-metric`); the blocking OTLP client is deliberate — the metrics SDK's
//! `PeriodicReader` runs on a background thread with no Tokio reactor.
//!
//! Bridging async→sync: [`container_stats::gather`] is async, but the SDK's
//! observable-instrument callbacks are sync. So an async task refreshes a
//! shared snapshot on an interval, and the (sync) callbacks read it.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;

use anyhow::{Context, Result};
use opentelemetry::KeyValue;
use opentelemetry::metrics::MeterProvider as _;
use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig, WithHttpConfig};
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};

use super::container_stats::{self, ContainerResource};

/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset.
const DEFAULT_INTERVAL: Duration = Duration::from_mins(1);

/// Latest container-stats snapshot: written by the async refresher, read by
/// the sync observable-instrument callbacks.
static SNAPSHOT: OnceLock<Arc<Mutex<Vec<ContainerResource>>>> = OnceLock::new();

/// Keep the provider alive for the process lifetime — the `PeriodicReader`
/// exports only while the provider lives.
static PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();

/// Spawn the container-resource OTEL exporter if OTEL is configured
/// (`HYPERHIVE_OTEL_ENDPOINT` non-empty — the same enable signal
/// [`crate::meta::otel_config`] uses). No-op otherwise. Call once at startup.
pub fn spawn_exporter() {
    let Some(endpoint) = endpoint() else {
        tracing::debug!("otel container-metrics: no endpoint configured, exporter disabled");
        return;
    };
    let snapshot = SNAPSHOT
        .get_or_init(|| Arc::new(Mutex::new(Vec::new())))
        .clone();
    let interval = interval();

    // Async refresher: keep the shared snapshot current for the sync callbacks.
    let refresh = snapshot.clone();
    tokio::spawn(async move {
        loop {
            let fresh = container_stats::gather().await;
            if let Ok(mut g) = refresh.lock() {
                *g = fresh;
            }
            tokio::time::sleep(interval).await;
        }
    });

    match build_provider(&endpoint, interval, snapshot) {
        Ok(provider) => {
            let _ = PROVIDER.set(provider);
            tracing::info!(%endpoint, ?interval, "otel container-metrics: exporter enabled");
        }
        Err(e) => {
            tracing::warn!(error = ?e, "otel container-metrics: exporter init failed");
        }
    }
}

fn build_provider(
    endpoint: &str,
    interval: Duration,
    snapshot: Arc<Mutex<Vec<ContainerResource>>>,
) -> Result<SdkMeterProvider> {
    // http/json — the only OTLP transport this crate enables (matching
    // hive-metric). The Claude SDK path honours `HYPERHIVE_OTEL_PROTOCOL` for
    // its own export; this exporter is always http/json.
    let mut builder = MetricExporter::builder()
        .with_http()
        .with_endpoint(endpoint)
        .with_protocol(Protocol::HttpJson);
    let headers = auth_headers();
    if !headers.is_empty() {
        builder = builder.with_headers(headers);
    }
    let exporter = builder.build().context("build OTLP metric exporter")?;
    // Drive the reader's export at `interval` so `HYPERHIVE_OTEL_METRIC_INTERVAL_MS`
    // is the real export cadence (not just the snapshot-refresh cadence). The
    // refresher runs at the same interval, gather-first, so the snapshot is
    // populated before the first export.
    let reader = PeriodicReader::builder(exporter)
        .with_interval(interval)
        .build();
    let provider = SdkMeterProvider::builder()
        .with_reader(reader)
        .with_resource(resource())
        .build();
    register_instruments(&provider, snapshot);
    Ok(provider)
}

/// Register the observable instruments. Each callback reads the shared
/// snapshot and reports one data point per agent. Instruments are held by the
/// meter/provider (kept alive in `PROVIDER`).
fn register_instruments(provider: &SdkMeterProvider, snapshot: Arc<Mutex<Vec<ContainerResource>>>) {
    let meter = provider.meter("hyperhive.container_stats");

    // semconv `container.cpu.time` — cumulative CPU seconds (monotonic counter).
    let snap = snapshot.clone();
    meter
        .f64_observable_counter("container.cpu.time")
        .with_unit("s")
        .with_callback(move |obs| {
            if let Ok(g) = snap.lock() {
                for c in g.iter() {
                    if let Some(usec) = c.cpu_time_usec {
                        #[allow(clippy::cast_precision_loss)]
                        obs.observe(usec as f64 / 1_000_000.0, &attrs(c));
                    }
                }
            }
        })
        .build();

    // semconv `container.memory.usage` — current usage bytes.
    let snap = snapshot.clone();
    meter
        .u64_observable_gauge("container.memory.usage")
        .with_unit("By")
        .with_callback(move |obs| {
            if let Ok(g) = snap.lock() {
                for c in g.iter() {
                    obs.observe(c.mem_current_bytes, &attrs(c));
                }
            }
        })
        .build();

    // The cgroup memory ceiling (omit when unlimited). Kept `hyperhive.`
    // custom — semconv defines `container.memory.usage` but not a matching
    // `.limit` metric, so don't claim a spec name that isn't in the spec.
    let snap = snapshot.clone();
    meter
        .u64_observable_gauge("hyperhive.container.memory.limit")
        .with_unit("By")
        .with_callback(move |obs| {
            if let Ok(g) = snap.lock() {
                for c in g.iter() {
                    if let Some(limit) = c.mem_max_bytes {
                        obs.observe(limit, &attrs(c));
                    }
                }
            }
        })
        .build();

    // Custom (`hyperhive.`) metrics semconv doesn't standardise: memory peak,
    // on-disk footprint, and the instantaneous cpu percent the dashboard shows.
    let snap = snapshot.clone();
    meter
        .u64_observable_gauge("hyperhive.container.memory.peak")
        .with_unit("By")
        .with_callback(move |obs| {
            if let Ok(g) = snap.lock() {
                for c in g.iter() {
                    if let Some(peak) = c.mem_peak_bytes {
                        obs.observe(peak, &attrs(c));
                    }
                }
            }
        })
        .build();

    let snap = snapshot.clone();
    meter
        .u64_observable_gauge("hyperhive.container.storage.usage")
        .with_unit("By")
        .with_callback(move |obs| {
            if let Ok(g) = snap.lock() {
                for c in g.iter() {
                    if let Some(disk) = c.disk_bytes {
                        obs.observe(disk, &attrs(c));
                    }
                }
            }
        })
        .build();

    let snap = snapshot;
    meter
        .f64_observable_gauge("hyperhive.container.cpu.percent")
        .with_unit("%")
        .with_callback(move |obs| {
            if let Ok(g) = snap.lock() {
                for c in g.iter() {
                    obs.observe(c.cpu_pct, &attrs(c));
                }
            }
        })
        .build();
}

/// Per-data-point attributes: the spec `container.name` plus the hive `agent`
/// label. (`hive` / `swarm` are constant across this c0re, so they live on the
/// resource.)
fn attrs(c: &ContainerResource) -> Vec<KeyValue> {
    vec![
        KeyValue::new("container.name", format!("h-{}", c.name)),
        KeyValue::new("agent", c.name.clone()),
    ]
}

/// Resource: `service.name = hyperhive-c0re` plus whatever the operator set in
/// `extraResourceAttributes` (where the hive `hive` / `swarm` labels live, same
/// channel the agent-side export uses).
fn resource() -> Resource {
    let mut builder = Resource::builder().with_service_name("hyperhive-c0re");
    for (k, v) in resource_attributes() {
        builder = builder.with_attribute(KeyValue::new(k, v));
    }
    builder.build()
}

/// `HYPERHIVE_OTEL_ENDPOINT`, non-empty. The enable signal.
fn endpoint() -> Option<String> {
    std::env::var("HYPERHIVE_OTEL_ENDPOINT")
        .ok()
        .map(|s| s.trim().to_owned())
        .filter(|s| !s.is_empty())
}

/// Export cadence from `HYPERHIVE_OTEL_METRIC_INTERVAL_MS`, else the default.
fn interval() -> Duration {
    std::env::var("HYPERHIVE_OTEL_METRIC_INTERVAL_MS")
        .ok()
        .and_then(|s| s.trim().parse::<u64>().ok())
        .filter(|ms| *ms > 0)
        .map_or(DEFAULT_INTERVAL, Duration::from_millis)
}

/// Extra resource attributes from `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES`
/// (`key=value,key=value`), same env the agent-side export reads.
fn resource_attributes() -> Vec<(String, String)> {
    std::env::var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES")
        .ok()
        .map(|s| parse_kv(&s))
        .unwrap_or_default()
}

/// OTLP auth headers from the systemd credential at
/// `$CREDENTIALS_DIRECTORY/otel-headers` (an `OTEL_EXPORTER_OTLP_HEADERS`-style
/// `Key=Value` line). Empty when the credential is absent — the collector is
/// then assumed unauthenticated.
fn auth_headers() -> HashMap<String, String> {
    let Some(dir) = std::env::var("CREDENTIALS_DIRECTORY").ok() else {
        return HashMap::new();
    };
    let path = std::path::Path::new(&dir).join("otel-headers");
    let Ok(raw) = std::fs::read_to_string(&path) else {
        tracing::warn!(
            "otel container-metrics: no auth header at $CREDENTIALS_DIRECTORY/otel-headers; \
             exporting without Authorization"
        );
        return HashMap::new();
    };
    parse_kv(&raw).into_iter().collect()
}

/// Parse `key=value` pairs separated by commas and/or newlines. The value
/// keeps any `=` after the first (so `Authorization=Bearer x=y` → `Bearer x=y`).
fn parse_kv(s: &str) -> Vec<(String, String)> {
    s.split([',', '\n'])
        .filter_map(|pair| {
            let pair = pair.trim();
            if pair.is_empty() {
                return None;
            }
            let (k, v) = pair.split_once('=')?;
            let k = k.trim();
            if k.is_empty() {
                None
            } else {
                Some((k.to_owned(), v.trim().to_owned()))
            }
        })
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parse_kv_keeps_bearer_value() {
        assert_eq!(
            parse_kv("Authorization=Bearer abc=def"),
            vec![("Authorization".to_owned(), "Bearer abc=def".to_owned())]
        );
        assert_eq!(
            parse_kv("a=1,\n b=2 ,=bad,"),
            vec![
                ("a".to_owned(), "1".to_owned()),
                ("b".to_owned(), "2".to_owned())
            ]
        );
    }
}