//! 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's own collector — the same first hop the agents use, named by the //! standard `OTEL_EXPORTER_OTLP_ENDPOINT` (see `nix/host-modules/hive-c0re`, //! which sets it from the same binding it hands agents). No toggle of //! its own, and no credential: a hive's collector takes unauthenticated OTLP //! on the bridge, and the only hop that presents anything is the swarm tier's, //! which is the one that leaves the swarm. //! //! 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. The resource additionally carries semconv //! `host.arch` and `service.version` (the running hyperhive flake rev), so a //! sample can be attributed to a machine and a deploy. //! 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. //! //! Also emits instruments whose subject is the hive itself rather than a //! container in it — see [`register_hive_instruments`]. They carry **no //! `agent` attribute**, and that absence is the point: it is what makes them //! selectable as `{hive!="", agent=""}`. use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, Instant}; use anyhow::{Context, Result}; use opentelemetry::KeyValue; use opentelemetry::metrics::MeterProvider as _; use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; use super::container_stats::{self, ContainerResource}; use super::warnings; /// 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>>> = OnceLock::new(); /// Keep the provider alive for the process lifetime — the `PeriodicReader` /// exports only while the provider lives. static PROVIDER: OnceLock = OnceLock::new(); /// When this exporter started, i.e. the origin `hyperhive.hive.uptime` counts /// from. Set once in [`spawn_exporter`], which runs during hive-c0re startup — /// so it is the process's age to within the startup sequence, and the metric's /// doc says exporter-start rather than claiming a precision it does not have. static STARTED: OnceLock = OnceLock::new(); /// Spawn the container-resource OTEL exporter if OTEL is configured /// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty). No-op otherwise. Call once at /// startup. pub fn spawn_exporter(hyperhive_flake: &str) { let Some(endpoint) = endpoint() else { tracing::debug!("otel container-metrics: no endpoint configured, exporter disabled"); return; }; let _ = STARTED.set(Instant::now()); 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(interval, snapshot, hyperhive_flake) { 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( interval: Duration, snapshot: Arc>>, hyperhive_flake: &str, ) -> Result { // 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. // // The address is deliberately NOT passed here. The SDK reads // `OTEL_EXPORTER_OTLP_ENDPOINT` itself and appends the signal path // (`/v1/metrics`); `with_endpoint` is taken **verbatim**, so handing it a // collector's base address POSTs to `/` and 404s on every export — with // nothing logged, because OTLP export failures go to an error handler no // binary here installs. That is not hypothetical: it was this module's // behaviour for its whole existence, and no sample ever reached the store. // An endpoint names a BASE throughout hyperhive and the // layer that knows the signal appends to it (the one exception is the // VictoriaMetrics exporter, whose far end is not a standard OTLP path). // Construction is identical to `hive-metric`'s on purpose: two producers, // one collector, one way to resolve the address. // // No auth headers: the destination is this hive's own collector, which // takes unauthenticated OTLP on the bridge. Adding one here would put the // upstream credential on a hop that never uses it. let exporter = MetricExporter::builder() .with_http() .with_protocol(Protocol::HttpJson) .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(hyperhive_flake)) .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>>) { 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(); register_hive_instruments(&meter); } /// Register the instruments whose subject is **this hive**, not a container in /// it. Each observes with an empty attribute set (`hive` / `swarm` ride the /// resource), so no data point here carries `agent` — see the module doc for /// why that absence is the point. /// /// Each is a projection of a value the process already computes; none is a new /// source of truth. In particular the health pair comes from /// [`warnings::readiness`], which is the one place that decides what counts as /// unhealthy — deriving `degraded` from `warnings{level="crit"}` in a /// dashboard query instead would put that rule in two places, and they would /// disagree silently the first time a second degrading condition is added. /// /// The two health instruments each read the registry separately, because the /// multi-instrument batch observer (one callback feeding several instruments) /// does not exist in this SDK: the API crate has no `register_callback`, and /// the SDK's is `pub(crate)`, reachable only via these per-instrument /// builders. Two consecutive reads inside one collection pass can therefore /// straddle a registry change. Both readings are individually valid and the /// next export agrees again; the shared-snapshot trick the container /// instruments use would move that window rather than close it, and exists /// only because `gather()` is async — which `readiness()` is not. fn register_hive_instruments(meter: &opentelemetry::metrics::Meter) { // Seconds since the exporter started. A hive that is up reports this every // interval whether or not it hosts a single agent, which is what makes // "no agents" distinguishable from "not reporting"; a restart reads as a // drop back to ~0. // // The semconv name, not a `hyperhive.` one: the spec defines // `process.uptime` as a double gauge in seconds, which is exactly this // instrument, and inventing a name for a quantity the spec already names // is the same mistake as claiming a spec name that does not exist (see // `host_arch` below). Same split the container family already uses — spec // names where a spec metric exists, `hyperhive.` for the rest. meter .f64_observable_gauge("process.uptime") .with_unit("s") .with_callback(|obs| { if let Some(started) = STARTED.get() { obs.observe(started.elapsed().as_secs_f64(), HIVE_ATTRS); } }) .build(); // The hive's own readiness verdict, 1 when degraded. Emitted as a series // rather than left for a query to compute — see the note above. meter .u64_observable_gauge("hyperhive.hive.degraded") .with_unit("1") .with_callback(|obs| { obs.observe(u64::from(warnings::readiness().is_degraded()), HIVE_ATTRS); }) .build(); // How many warnings are currently raised, split by level. Every level is // reported every cycle, zero included: an instrument that goes silent when // there is nothing to report makes "healthy" and "not reporting" the same // observation, which is the confusion this whole family exists to remove. meter .u64_observable_gauge("hyperhive.hive.warnings") .with_unit("1") .with_callback(|obs| { for (level, n) in warning_counts(&warnings::snapshot()) { obs.observe(n, &[KeyValue::new("level", level)]); } }) .build(); } /// Attributes for the hive-scoped family: **none**. /// /// Named rather than written as a bare `&[]` at each call site because the /// emptiness is the load-bearing part — it is what makes these series /// selectable as `{hive!="", agent=""}`. A bare empty slice reads like nothing /// was decided; this one has a test asserting no `agent` key ever appears in /// it, so "helpfully" labelling the hive family per-agent goes red. const HIVE_ATTRS: &[KeyValue] = &[]; /// One count per warning level, **including the levels at zero**. /// /// The zeros are the point. An instrument that reports only the levels /// currently raised goes silent on a healthy hive, and a silent series is /// indistinguishable from a hive that has stopped reporting — the exact /// confusion the `hyperhive.hive.*` family exists to remove. fn warning_counts(raised: &[crate::host_stats::ServerWarning]) -> Vec<(&'static str, u64)> { warnings::LEVELS .into_iter() .map(|level| { // Summed rather than `count()`ed to stay cast-free: `count()` // yields a `usize` and the instrument takes a `u64`. let n: u64 = raised.iter().filter(|w| w.level == level).map(|_| 1).sum(); (level, n) }) .collect() } /// 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 { vec![ KeyValue::new("container.name", format!("h-{}", c.name)), KeyValue::new("agent", c.name.clone()), ] } /// Resource: `service.name = hyperhive-c0re`, the host architecture, the /// running hyperhive revision, plus whatever the operator set in /// `extraResourceAttributes` (where the hive `hive` / `swarm` labels live, same /// channel the agent-side export uses). /// /// These belong on the resource rather than on each data point for the same /// reason `hive` / `swarm` do: they are constant across one `hive-c0re`. Only /// facts that vary per container (see [`attrs`]) are per-data-point. /// /// `service.version` is the **flake revision**, not the crate version — the /// crate version is a workspace constant that never moves between deploys, so /// it could not answer "which build produced this sample?". `None` when the /// flake ref carries no rev (a local path ref), in which case the attribute is /// omitted rather than reported as a guess. fn resource(hyperhive_flake: &str) -> Resource { let mut builder = Resource::builder() .with_service_name("hyperhive-c0re") .with_attribute(KeyValue::new("host.arch", host_arch())); if let Some(rev) = crate::auto_update::current_flake_rev(hyperhive_flake) { builder = builder.with_attribute(KeyValue::new("service.version", rev)); } for (k, v) in resource_attributes() { builder = builder.with_attribute(KeyValue::new(k, v)); } builder.build() } /// The host architecture as OTEL semconv spells it. /// /// Rust's [`std::env::consts::ARCH`] and semconv `host.arch` disagree on the /// two architectures this actually runs on: rust says `x86_64` / `aarch64`, /// semconv says `amd64` / `arm64`. Forwarding rust's spelling would emit a /// label that off-the-shelf dashboards silently fail to match — the same class /// of problem as guessing a metric name. Anything else is passed through: a /// non-standard value is more useful than a wrong standard one. fn host_arch() -> &'static str { match std::env::consts::ARCH { "x86_64" => "amd64", "aarch64" => "arm64", other => other, } } /// `OTEL_EXPORTER_OTLP_ENDPOINT`, non-empty. The enable signal — and the very /// variable the SDK reads to build the exporter's URL, so "configured" and /// "where it goes" cannot disagree. Read here only to decide whether to start /// at all; the value is never handed to the builder (see `build_provider`). fn endpoint() -> Option { std::env::var("OTEL_EXPORTER_OTLP_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::().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() } /// 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::*; use crate::host_stats::ServerWarning; fn warning(level: &'static str) -> ServerWarning { ServerWarning { kind: "t", level, message: String::new(), } } /// The hive family must not be labelled per-agent: `{hive!="", agent=""}` /// is the whole selector that distinguishes "a measurement of the hive" /// from "a measurement of an agent that happens to live in one", and a /// single `agent` attribute here collapses the two. /// /// Paired with a control on [`attrs`], because this assertion is an /// *absence*: without proof that the same check finds an `agent` key when /// one is genuinely present, an empty result would equally mean the test /// is looking for the wrong thing. #[test] fn hive_attrs_carry_no_agent_but_container_attrs_do() { let has_agent = |kvs: &[KeyValue]| kvs.iter().any(|kv| kv.key.as_str() == "agent"); assert!( !has_agent(HIVE_ATTRS), "a hive-scoped series must not carry an `agent` attribute" ); // Control: the same predicate, on the per-container attributes, which // are supposed to carry exactly that key. let sample = ContainerResource { name: "atlas".to_owned(), cpu_pct: 0.0, cpu_time_usec: None, mem_current_bytes: 0, mem_peak_bytes: None, mem_max_bytes: None, disk_bytes: None, }; assert!( has_agent(&attrs(&sample)), "control failed: the predicate cannot find an `agent` key that IS \ present, so the assertion above proves nothing" ); } /// A healthy hive still reports every level, at zero. Reporting only the /// levels currently raised would make a healthy hive and a hive that has /// stopped exporting produce the same thing — no series. #[test] fn warning_counts_reports_every_level_even_when_none_are_raised() { let counts = warning_counts(&[]); assert_eq!(counts.len(), warnings::LEVELS.len()); for (level, n) in counts { assert_eq!(n, 0, "{level} should be reported as zero, not omitted"); } } #[test] fn warning_counts_splits_by_level() { let raised = [ warning(warnings::LEVEL_WARN), warning(warnings::LEVEL_CRIT), warning(warnings::LEVEL_WARN), ]; let counts = warning_counts(&raised); assert_eq!( counts, vec![(warnings::LEVEL_WARN, 2), (warnings::LEVEL_CRIT, 1)] ); } #[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()) ] ); } /// `host.arch` must be the semconv spelling, not rust's. The two disagree /// on exactly the architectures hyperhive runs on, and the failure is /// silent — a dashboard filtering `host.arch="amd64"` simply matches /// nothing if we emit `x86_64`. #[test] fn host_arch_is_semconv_not_rust_spelling() { assert!( !matches!(host_arch(), "x86_64" | "aarch64"), "host_arch() forwarded rust's ARCH spelling ({}) — semconv wants amd64 / arm64", host_arch() ); // On the architectures we actually ship, pin the exact expected value // rather than only asserting the negative above. #[cfg(target_arch = "x86_64")] assert_eq!(host_arch(), "amd64"); #[cfg(target_arch = "aarch64")] assert_eq!(host_arch(), "arm64"); } /// The exporter's destination must come from the standard OTLP variable, /// because that is the only spelling the SDK appends the signal path to. /// /// Asserted rather than left to review because the fix here is an /// *absence* — no `with_endpoint` call — and an absence is exactly what /// a later "the endpoint is right there, just pass it" edit restores. If /// this exporter is ever gated on a variable the SDK does not itself read, /// it resumes posting to a base URL that 404s in silence. #[test] fn endpoint_is_the_standard_otlp_var() { // Serialised against every other env-mutating test in the crate. Both // variables below are also read by `meta::tests` in this same test // binary, so "no other test in this module" would be the wrong // boundary — the module is not the unit that shares the environment, // the process is. let _env = crate::test_env::lock(); // SAFETY: serialised by the guard above; both names are restored (to // absent) before it drops at the end of this test. unsafe { std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT"); std::env::set_var("HYPERHIVE_OTEL_ENDPOINT", "http://hyperhive.invalid:4318"); } assert_eq!( endpoint(), None, "the hive-wide agent-config variable must NOT enable this exporter — \ the SDK does not read it, so its address would never reach the builder" ); unsafe { std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", " http://10.42.0.1:4318 "); } assert_eq!( endpoint(), Some("http://10.42.0.1:4318".to_owned()), "the standard variable enables the exporter, trimmed" ); unsafe { std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", " "); } assert_eq!( endpoint(), None, "whitespace-only is not a configured endpoint" ); unsafe { std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT"); std::env::remove_var("HYPERHIVE_OTEL_ENDPOINT"); } } }