diff --git a/Cargo.lock b/Cargo.lock index a9b68c45..e6fdc375 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1556,6 +1556,9 @@ dependencies = [ "indicatif", "libc", "listenfd", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", "petgraph", "problem_details", "reqwest 0.12.28", diff --git a/docs/observability.md b/docs/observability.md index c95d5d52..97993735 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -127,6 +127,42 @@ Additional labels can be appended via `extraResourceAttributes` (see option reference above); custom per-data-point labels can be passed with `hive-metric --labels` (see below). +## Host-emitted container-resource metrics (hive-c0re) + +When OTEL is enabled, **hive-c0re itself** also exports each agent +container's resource load — the same cgroup gauges shown on the dashboard +LOAD tab — to the configured `endpoint`, reusing the same +`services.hyperhive.otel` config (no separate toggle). These come from the +host, not the in-container Claude SDK, so they cover containers even when +their agent is idle. + +Emitted via the OpenTelemetry Rust SDK, using the +[semconv `container.*`](https://opentelemetry.io/docs/specs/semconv/system/container-metrics/) +metric names + the standard `container.name` attribute where a spec metric +exists, so off-the-shelf OTEL/Grafana container dashboards work. Resource +`service.name = hyperhive-c0re`; each data point is tagged `container.name` +(= the `h-` machine) and the hive `agent` label: + +| Metric | Unit | Kind | Source | +|--------|------|------|--------| +| `container.cpu.time` | `s` | counter | cumulative `cpu.stat` `usage_usec` → seconds | +| `container.memory.usage` | `By` | gauge | `memory.current` | +| `hyperhive.container.memory.limit` | `By` | gauge | `memory.max` (custom — semconv has no `.limit` metric; omitted when unlimited) | +| `hyperhive.container.memory.peak` | `By` | gauge | `memory.peak` (custom — no semconv metric; omitted if unavailable) | +| `hyperhive.container.storage.usage` | `By` | gauge | state dir + writable rootfs (custom — semconv only has `disk.io`; omitted until the slow disk sampler runs) | +| `hyperhive.container.cpu.percent` | `%` | gauge | host-normalised percent (custom — the value the dashboard LOAD tab shows, no `rate()` needed) | + +The `hyperhive.`-prefixed metrics have no semconv equivalent (memory +limit + peak, on-disk footprint, and an instantaneous cpu percent kept +alongside the spec `container.cpu.time` counter for convenience). Hive +labels (`hive`, `swarm`, …) ride on the resource via +`extraResourceAttributes`. + +Cadence follows `metricIntervalMs` (default 60s). Transport is OTLP/HTTP +(JSON) to ``; the auth header is loaded onto hive-c0re's own unit +via systemd `LoadCredential` (from the same `headersCredential` file) and +sent as `Authorization`. + ## Agent-emitted custom metrics (`hive-metric`) Agents can push arbitrary labeled metrics to the same OTEL collector via the diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 91b2c76f..b3e71558 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -17,6 +17,18 @@ url.workspace = true clap.workspace = true clap_complete.workspace = true clap-markdown = "0.1" +# OTEL SDK for the per-agent container-resource metrics exporter +# (stats/otel_metrics.rs). Same versions/features as hive-metric — the +# blocking OTLP client is deliberate: the metrics SDK's PeriodicReader runs +# on a background thread with no Tokio reactor, where the async client panics. +opentelemetry = "0.32" +opentelemetry_sdk = { version = "0.32", features = ["metrics"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = [ + "metrics", + "http-json", + "reqwest-blocking-client", + "reqwest-rustls", +] } indicatif.workspace = true hive-sh4re.workspace = true hive-host-sock.workspace = true diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index 783a48b9..ab6c4724 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -44,7 +44,7 @@ pub mod workers; // Root re-exports: keep every pre-grouping `crate::` / // `hive_c0re::` path compiling without touching consumers. pub use agent_config::{capabilities, limits, tool_groups, topology}; -pub use stats::{container_stats, hive_stats, host_stats, sweep_health, warnings}; +pub use stats::{container_stats, hive_stats, host_stats, otel_metrics, sweep_health, warnings}; pub use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, }; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index f642d111..3997238f 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -428,6 +428,11 @@ async fn cmd_serve( // poll stays cheap cgroup-only reads. Feeds `disk_bytes` on the LOAD // tab. See container_stats::disk_sampler_loop. hive_c0re::container_stats::spawn_disk_sampler(); + // Per-agent container-resource OTEL export: rides the same + // cgroup gauges out to the configured OTLP endpoint, reusing the hive + // `services.hyperhive.otel` config (endpoint + LoadCredential auth). + // No-op when OTEL isn't configured. + hive_c0re::otel_metrics::spawn_exporter(); // build_logs.sqlite vacuum: c0re-side (single db). Failures kept // 30d, successes 24h — see `build_logs::vacuum` for the rule. hive_c0re::build_logs::spawn_vacuum(&coord); diff --git a/hive-c0re/src/stats/container_stats.rs b/hive-c0re/src/stats/container_stats.rs index bf6d71d0..cada8706 100644 --- a/hive-c0re/src/stats/container_stats.rs +++ b/hive-c0re/src/stats/container_stats.rs @@ -43,6 +43,11 @@ pub struct ContainerResource { /// Host-normalised CPU usage over the sample interval, as a /// percentage of total host CPU (0..100 across all cores). pub cpu_pct: f64, + /// Cumulative CPU time (`cpu.stat` `usage_usec`, microseconds, + /// monotonic). `None` if the read failed. Feeds the OTEL semconv + /// `container.cpu.time` counter (which wants absolute cumulative time, + /// converted to seconds), distinct from the sampled `cpu_pct`. + pub cpu_time_usec: Option, /// Current memory usage (`memory.current`), bytes. pub mem_current_bytes: u64, /// High-water memory usage since container start (`memory.peak`), @@ -262,7 +267,11 @@ pub async fn gather() -> Vec { let mut out: Vec = Vec::with_capacity(candidates.len()); for (i, (name, dir)) in candidates.iter().enumerate() { - let cpu_pct = match (t0[i], read_usage_usec(dir)) { + // The second cumulative read — also exposed raw as `cpu_time_usec` + // for the OTEL `container.cpu.time` counter (which wants absolute + // cumulative CPU time, not the sampled percent). + let usage_now = read_usage_usec(dir); + let cpu_pct = match (t0[i], usage_now) { (Some(a), Some(b)) => { #[allow( clippy::cast_precision_loss, @@ -276,6 +285,7 @@ pub async fn gather() -> Vec { out.push(ContainerResource { name: name.clone(), cpu_pct, + cpu_time_usec: usage_now, mem_current_bytes: read_u64(&dir.join("memory.current")).unwrap_or(0), mem_peak_bytes: read_u64(&dir.join("memory.peak")), mem_max_bytes: read_mem_max(&dir.join("memory.max")), diff --git a/hive-c0re/src/stats/mod.rs b/hive-c0re/src/stats/mod.rs index 8b9d5ec8..cfa36917 100644 --- a/hive-c0re/src/stats/mod.rs +++ b/hive-c0re/src/stats/mod.rs @@ -6,5 +6,6 @@ pub mod container_stats; pub mod hive_stats; pub mod host_stats; +pub mod otel_metrics; pub mod sweep_health; pub mod warnings; diff --git a/hive-c0re/src/stats/otel_metrics.rs b/hive-c0re/src/stats/otel_metrics.rs new file mode 100644 index 00000000..15ef341f --- /dev/null +++ b/hive-c0re/src/stats/otel_metrics.rs @@ -0,0 +1,318 @@ +//! 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>>> = OnceLock::new(); + +/// Keep the provider alive for the process lifetime — the `PeriodicReader` +/// exports only while the provider lives. +static PROVIDER: OnceLock = 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>>, +) -> 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. + 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>>) { + 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 { + 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 { + 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::().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 { + 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()) + ] + ); + } +} diff --git a/nix/host-modules/hive-c0re/default.nix b/nix/host-modules/hive-c0re/default.nix index d658dd00..ae2736be 100644 --- a/nix/host-modules/hive-c0re/default.nix +++ b/nix/host-modules/hive-c0re/default.nix @@ -157,6 +157,15 @@ in RuntimeDirectoryPreserve = "yes"; StateDirectory = "hyperhive"; StateDirectoryMode = "0750"; + # OTEL auth-header secret, loaded onto hive-c0re's own unit so its + # per-agent container-resource metrics exporter can read it + # at $CREDENTIALS_DIRECTORY/otel-headers — via systemd, not a world + # path. Same secret the agent containers get (forwarded there via + # nspawn --load-credential); this just also hands it to c0re itself. + # Empty list (no credential) when otel is off or no header is set. + LoadCredential = lib.optional ( + config.services.hyperhive.otel.enable && config.services.hyperhive.otel.headersCredential != null + ) "otel-headers:${config.services.hyperhive.otel.headersCredential}"; # Sandboxing. hive-c0re is unprivileged (runs as hive-core, never # setuid), makes HTTP requests to forge/matrix/Anthropic (keeps INET), # and delegates all privileged ops to hive-priv via a Unix socket.