diff --git a/Cargo.lock b/Cargo.lock index 2e0bf188..a769c873 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1593,6 +1593,9 @@ dependencies = [ "hyper", "hyper-util", "libc", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", "reqwest", "rmcp", "rusqlite", diff --git a/hive-agent/Cargo.toml b/hive-agent/Cargo.toml index 3b6ffff4..7e4c90d2 100644 --- a/hive-agent/Cargo.toml +++ b/hive-agent/Cargo.toml @@ -23,6 +23,9 @@ hive-core-agent-sock.workspace = true hive-sh4re.workspace = true hive-sock-client.workspace = true libc.workspace = true +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true +opentelemetry-otlp.workspace = true rmcp.workspace = true rusqlite.workspace = true schemars.workspace = true diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 9f659a65..edacee1a 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -17,6 +17,7 @@ mod identity; mod login; mod login_session; mod mcp_config; +mod otel_turn_metrics; mod paths; mod plugins; mod prompt; @@ -1006,30 +1007,42 @@ async fn handle_turn( let tool_calls = bus.take_tool_calls(); let todo_wake_checked = (from == "todo").then(|| tool_calls.contains_key("mcp__hyperhive__get_loose_ends")); + // Read-and-clear once per turn unconditionally (same rationale as + // `take_tool_calls` above — the harness gets exactly one chance to + // observe this flag) rather than only when the sqlite sink happens to be + // configured: the OTEL turn-metrics exporter's session-boundary counter + // needs it too, independent of `stats`. + let fresh_session = bus.take_fresh_session(); if let Some(stats) = stats { // Fresh session this turn → mint a `sessions` row and set its id on // the bus so this turn (and subsequent ones until the next fresh - // start) stamp `turn_stats.session_id`. Takes the one-shot flag - // `run_claude` set when it suppressed `--continue`. - if bus.take_fresh_session() { + // start) stamp `turn_stats.session_id`. + if fresh_session { let sid = stats.start_session(started_at, &model_at_start); bus.set_session_id(sid); } - let ended_at = chrono::Utc::now().timestamp(); - let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); - let (open_threads, open_reminders) = S::post_turn_counts(socket).await; - let row = serve_common::build_row(serve_common::TurnRowArgs { - started_at, - ended_at, - duration_ms, - model: model_at_start, - wake_from: from.clone(), - outcome: &outcome, - bus, - tool_calls, - open_threads_count: open_threads, - open_reminders_count: open_reminders, - }); + } + let ended_at = chrono::Utc::now().timestamp(); + let duration_ms = i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX); + let (open_threads, open_reminders) = S::post_turn_counts(socket).await; + let row = serve_common::build_row(serve_common::TurnRowArgs { + started_at, + ended_at, + duration_ms, + model: model_at_start, + wake_from: from.clone(), + outcome: &outcome, + bus, + tool_calls, + open_threads_count: open_threads, + open_reminders_count: open_reminders, + }); + // Harness-only OTEL metrics (duration/wake_from/result_kind/loose-ends/ + // session boundaries) — independent of the sqlite sink below, and a + // cheap no-op when OTEL isn't configured. See `otel_turn_metrics`'s + // module doc for why token/cost/tool-count are deliberately not here. + otel_turn_metrics::record(&row, fresh_session); + if let Some(stats) = stats { stats.record(&row); } let pending = S::inbox_unread(socket).await; diff --git a/hive-agent/src/otel_turn_metrics.rs b/hive-agent/src/otel_turn_metrics.rs new file mode 100644 index 00000000..be13a517 --- /dev/null +++ b/hive-agent/src/otel_turn_metrics.rs @@ -0,0 +1,159 @@ +//! Per-turn OTEL metric export for the fields Claude Code's own built-in +//! OTEL integration cannot know about. Claude's own export already covers +//! token usage, cost, and tool-call counts (see `docs/observability.md`) — +//! duplicating those here under different metric names would just give a +//! collector two disagreeing series for the same number. This module emits +//! only the harness-only concepts: wall-clock turn duration as *this harness* +//! measures it (not Claude's own per-request latency), what woke the turn, +//! the harness's own outcome classification, the loose-ends backlog at turn +//! end, and session boundaries (fresh vs. `--continue`'d). Scoped this way +//! per the tracker discussion on the "emit agent stats as OTEL metrics" issue. +//! +//! Synchronous instruments, not observable ones: unlike hive-c0re's +//! container-resource gauges (`hive-c0re/src/stats/otel_metrics.rs`), which +//! poll a periodically-refreshed snapshot because the underlying cgroup value +//! is continuously live, every value here only exists at one instant — the +//! moment a turn ends — so it's recorded directly (`counter.add`, +//! `histogram.record`, `gauge.record`) from [`record`], called once per turn +//! from `handle_turn`. The `PeriodicReader` still batches + exports on its +//! own interval; only the recording is event-driven, not the export. +//! +//! Resource attributes (`service.name`, `agent`, `hive`, `swarm`) are picked +//! up automatically by the SDK from the container-wide `OTEL_RESOURCE_ATTRIBUTES` +//! env var (same mechanism `hive-metric` relies on) — nothing to set here. + +use std::sync::OnceLock; +use std::time::Duration; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, Gauge, Histogram, MeterProvider as _}; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; + +use crate::turn_stats::TurnStatRow; + +/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset. +/// Matches `hive-c0re`'s container-metrics default so the two exporters +/// batch on the same rhythm by default. +const DEFAULT_INTERVAL: Duration = Duration::from_mins(1); + +struct Instruments { + /// Kept alive for the process lifetime — the `PeriodicReader` only + /// exports while its provider lives. + _provider: SdkMeterProvider, + turn_duration_ms: Histogram, + turn_count: Counter, + session_count: Counter, + open_threads: Gauge, + open_reminders: Gauge, +} + +/// Lazily built on the first call to [`record`]. `None` when OTEL isn't +/// configured (`OTEL_EXPORTER_OTLP_ENDPOINT` unset/empty) or the exporter +/// failed to build — either way `record` becomes a cheap no-op. +static INSTRUMENTS: OnceLock> = OnceLock::new(); + +/// Record one turn's harness-only metrics. No-op (and cheap — one +/// `OnceLock` read after the first call) when OTEL isn't configured. +/// +/// `fresh_session` is the harness's own "did this turn start a brand-new +/// claude session" flag (`Bus::take_fresh_session`) — pass it through rather +/// than re-deriving it here, since the harness only gets to observe it once +/// per turn (the same flag gates minting a `turn_stats` `sessions` row). +pub fn record(row: &TurnStatRow, fresh_session: bool) { + let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else { + return; + }; + let attrs = [ + KeyValue::new("wake_from", row.wake_from.clone()), + KeyValue::new("result_kind", row.result_kind), + KeyValue::new("model", row.model.clone()), + ]; + // `try_from` rather than `as` — `duration_ms` is never negative in + // practice (it's `Instant::elapsed().as_millis()`), but a fallback of 0 + // beats a lossy cast or a panic if that ever stops holding. + inst.turn_duration_ms + .record(u64::try_from(row.duration_ms).unwrap_or(0), &attrs); + inst.turn_count.add(1, &attrs); + if let Some(threads) = row.open_threads_count { + inst.open_threads.record(threads, &[]); + } + if let Some(reminders) = row.open_reminders_count { + inst.open_reminders.record(reminders, &[]); + } + if fresh_session { + inst.session_count + .add(1, &[KeyValue::new("model", row.model.clone())]); + } +} + +fn build() -> Option { + if !enabled() { + tracing::debug!("otel turn-metrics: no endpoint configured, exporter disabled"); + return None; + } + let interval = interval(); + match build_provider(interval) { + Ok(provider) => { + tracing::info!(?interval, "otel turn-metrics: exporter enabled"); + let meter = provider.meter("hyperhive.turn_stats"); + Some(Instruments { + turn_duration_ms: meter + .u64_histogram("hyperhive.agent.turn.duration") + .with_unit("ms") + .build(), + turn_count: meter.u64_counter("hyperhive.agent.turn.count").build(), + session_count: meter.u64_counter("hyperhive.agent.session.count").build(), + open_threads: meter + .u64_gauge("hyperhive.agent.loose_ends.threads") + .build(), + open_reminders: meter + .u64_gauge("hyperhive.agent.loose_ends.reminders") + .build(), + _provider: provider, + }) + } + Err(e) => { + tracing::warn!(error = ?e, "otel turn-metrics: exporter init failed"); + None + } + } +} + +fn build_provider(interval: Duration) -> anyhow::Result { + // http/json, no explicit endpoint/resource: same rationale as + // `hive-c0re/src/stats/otel_metrics.rs` for the transport (the blocking + // OTLP client is required — `PeriodicReader` drives export from a + // background thread with no Tokio reactor), and same rationale as + // `hive-metric` for the resource (the SDK reads `OTEL_RESOURCE_ATTRIBUTES` + // — set container-wide, see `docs/observability.md::Built-in resource + // labels` — on its own; passing an explicit `Resource` here would need to + // duplicate `agent`/`hive`/`swarm` this crate has no other reason to know). + let exporter = MetricExporter::builder() + .with_http() + .with_protocol(Protocol::HttpJson) + .build()?; + let reader = PeriodicReader::builder(exporter) + .with_interval(interval) + .build(); + Ok(SdkMeterProvider::builder().with_reader(reader).build()) +} + +/// `OTEL_EXPORTER_OTLP_ENDPOINT` non-empty — the same enable signal every +/// other exporter in this workspace uses (see `hive-c0re/src/stats/ +/// otel_metrics.rs::endpoint` for why this exact variable and not a +/// hyperhive-specific one: it's also the variable the SDK itself reads to +/// build the exporter's URL, so "configured" and "where it goes" cannot +/// disagree). +fn enabled() -> bool { + std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok_and(|s| !s.trim().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) +}