From ef6a86872d95fdb95ac9c7ce14b14e8907fefb4a Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 24 Jun 2026 20:48:38 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(#1974):=20hivectl=20migrate-stats=20?= =?UTF-8?q?=E2=80=94=20backfill=20per-agent=20token=20history=20to=20otel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hive-c0re/src/bin/hivectl.rs | 25 ++++ hive-c0re/src/lib.rs | 1 + hive-c0re/src/otel_migrate.rs | 207 ++++++++++++++++++++++++++++++++++ 3 files changed, 233 insertions(+) create mode 100644 hive-c0re/src/otel_migrate.rs diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index 4379f1bf..474907da 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -204,6 +204,26 @@ enum Cmd { #[command(subcommand)] cmd: SubvolCmd, }, + /// One-shot backfill of historical per-agent turn stats into the OTEL + /// collector as cumulative `claude_code.token.usage` metrics, so + /// pre-OTEL token history shows up in grafana alongside the live + /// stream. Reads each agent's `hyperhive-turn-stats.sqlite` and POSTs + /// OTLP/HTTP JSON to the collector. Every datapoint is tagged + /// `hyperhive-migration="true"`. Safe to re-run (cumulative samples + /// at fixed historical timestamps are idempotent on the backend). + MigrateStats { + /// OTLP collector base URL (e.g. `http://10.42.0.1:4318`). + /// Defaults to `$OTEL_EXPORTER_OTLP_ENDPOINT`, then + /// `$HYPERHIVE_OTEL_ENDPOINT`. + #[arg(long)] + endpoint: Option, + /// Limit to a single agent (default: every agent with a state dir). + #[arg(long)] + agent: Option, + /// Report datapoint counts without sending anything. + #[arg(long)] + dry_run: bool, + }, /// Emit the full CLI reference as `CommonMark` to stdout. /// /// Hidden tooling command (not part of day-to-day operator admin): @@ -608,6 +628,11 @@ async fn main() -> Result<()> { SubvolCmd::Upgrade { name, yes } => subvol_upgrade(&socket, &name, yes).await, }, Cmd::Choom { name, fresh } => choom(&name, fresh), + Cmd::MigrateStats { + endpoint, + agent, + dry_run, + } => hive_c0re::otel_migrate::run(endpoint, agent, dry_run).await, Cmd::Quota { cmd } => match cmd { QuotaCmd::Enable => quota_enable().await, QuotaCmd::Show { name } => quota_show(name.as_deref()).await, diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index 53f66241..257995f7 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -42,6 +42,7 @@ pub mod matrix; pub mod meta; pub mod migrate; pub mod operator_questions; +pub mod otel_migrate; pub mod paths; pub mod priv_client; pub mod questions; diff --git a/hive-c0re/src/otel_migrate.rs b/hive-c0re/src/otel_migrate.rs new file mode 100644 index 00000000..e67d4887 --- /dev/null +++ b/hive-c0re/src/otel_migrate.rs @@ -0,0 +1,207 @@ +//! One-shot backfill of historical per-agent turn stats into the OTEL +//! collector as `claude_code.token.usage` metrics, so pre-OTEL history +//! shows up alongside the live telemetry stream. Driven by +//! `hivectl migrate-stats`. +//! +//! Reads each agent's `hyperhive-turn-stats.sqlite` (host path via +//! `Coordinator::agent_harness_dir`) and turns the four per-turn token +//! columns into **cumulative** `claude_code.token.usage` counter samples: +//! one series per `(type, model)`, value = the running total at each +//! turn's end time. Cumulative — not delta — because Prometheus/Mimir +//! family backends silently drop delta sums (the same reason the live +//! export forces cumulative temporality in harness-base.nix). The samples +//! are sent as OTLP/HTTP JSON to `/v1/metrics`. Every +//! datapoint carries `hyperhive-migration="true"` so the backfill is +//! distinguishable from the live stream. +//! +//! Timestamps: `turn_stats.{started_at,ended_at}` are epoch *seconds* +//! (see `hive-ag3nt` turn.rs), scaled to nanoseconds for OTLP. + +use anyhow::{Context, Result}; +use rusqlite::Connection; +use serde_json::{Value, json}; +use std::collections::HashMap; +use std::path::Path; + +use crate::coordinator::Coordinator; + +const METRIC: &str = "claude_code.token.usage"; + +/// Cap datapoints per POST so a long-lived agent's history is chunked +/// into reasonably-sized OTLP requests rather than one huge body. +const MAX_DATAPOINTS_PER_POST: usize = 1000; + +/// `claude_code.token.usage` `type` attribute value paired with the +/// `turn_stats` column it maps to (column index in the SELECT below). +const TOKEN_TYPES: &[(&str, usize)] = &[ + ("input", 3), + ("output", 4), + ("cacheRead", 5), + ("cacheCreation", 6), +]; + +/// Run the migration. `endpoint` falls back to `OTEL_EXPORTER_OTLP_ENDPOINT` +/// then `HYPERHIVE_OTEL_ENDPOINT`. `only_agent` limits to one agent. +/// `dry_run` reports counts without sending. +/// +/// # Errors +/// Returns an error when no endpoint is resolvable, a stats db can't be +/// read, or the collector rejects a POST. +pub async fn run( + endpoint: Option, + only_agent: Option, + dry_run: bool, +) -> Result<()> { + let endpoint = endpoint + .or_else(|| std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok()) + .or_else(|| std::env::var("HYPERHIVE_OTEL_ENDPOINT").ok()) + .filter(|s| !s.is_empty()) + .context("no OTLP endpoint: pass --endpoint or set OTEL_EXPORTER_OTLP_ENDPOINT")?; + let url = format!("{}/v1/metrics", endpoint.trim_end_matches('/')); + let hive = std::env::var("HYPERHIVE_HIVE_NAME").unwrap_or_else(|_| "unknown".to_owned()); + let swarm = std::env::var("HYPERHIVE_SWARM_NAME").unwrap_or_else(|_| "unknown".to_owned()); + + let agents = match only_agent { + Some(a) => vec![a], + None => Coordinator::kept_state_names(), + }; + + let client = reqwest::Client::new(); + let mut total_points = 0usize; + let mut hit_agents = 0usize; + for agent in &agents { + let db = Coordinator::agent_harness_dir(agent).join("hyperhive-turn-stats.sqlite"); + if !db.exists() { + continue; + } + let points = + collect_agent_points(&db).with_context(|| format!("read turn_stats for {agent}"))?; + if points.is_empty() { + println!("{agent}: no token rows, skipping"); + continue; + } + hit_agents += 1; + total_points += points.len(); + println!("{agent}: {} datapoints", points.len()); + if dry_run { + continue; + } + for chunk in points.chunks(MAX_DATAPOINTS_PER_POST) { + let payload = build_payload(agent, &hive, &swarm, chunk); + post(&client, &url, &payload) + .await + .with_context(|| format!("POST metrics for {agent}"))?; + } + } + println!( + "migrate-stats: {hit_agents} agent(s) with stats, {total_points} datapoints {}", + if dry_run { + "(dry-run, nothing sent)" + } else { + "sent" + } + ); + Ok(()) +} + +/// Build cumulative `claude_code.token.usage` datapoints for one agent's +/// stats db. One running-total series per `(type, model)`; a datapoint is +/// emitted whenever that series' value changes (token delta > 0). +fn collect_agent_points(db: &Path) -> Result> { + let conn = Connection::open_with_flags(db, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) + .context("open stats db read-only")?; + let mut stmt = conn.prepare( + "SELECT started_at, ended_at, model, input_tokens, output_tokens, \ + cache_read_input_tokens, cache_creation_input_tokens \ + FROM turn_stats ORDER BY ended_at ASC", + )?; + let rows = stmt.query_map([], |r| { + Ok(( + r.get::<_, i64>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, String>(2)?, + r.get::<_, i64>(3)?, + r.get::<_, i64>(4)?, + r.get::<_, i64>(5)?, + r.get::<_, i64>(6)?, + )) + })?; + + // (type, model) -> (running total, series start time in nanos string) + let mut series: HashMap<(&'static str, String), (u64, String)> = HashMap::new(); + let mut points: Vec = Vec::new(); + for row in rows { + let (started, ended, model, inp, out, cache_read, cache_create) = row?; + let cols = [inp, out, cache_read, cache_create]; + let start_nanos = u64::try_from(started).unwrap_or(0) * 1_000_000_000; + let time_nanos = (u64::try_from(ended).unwrap_or(0) * 1_000_000_000).to_string(); + for &(type_, idx) in TOKEN_TYPES { + let val = cols[idx - 3]; + if val <= 0 { + continue; + } + let entry = series + .entry((type_, model.clone())) + .or_insert_with(|| (0u64, start_nanos.to_string())); + entry.0 += u64::try_from(val).unwrap_or(0); + points.push(json!({ + "asInt": entry.0.to_string(), + "startTimeUnixNano": entry.1.clone(), + "timeUnixNano": time_nanos, + "attributes": [ + {"key": "type", "value": {"stringValue": type_}}, + {"key": "model", "value": {"stringValue": model.clone()}}, + {"key": "hyperhive-migration", "value": {"stringValue": "true"}}, + ], + })); + } + } + Ok(points) +} + +/// Wrap a chunk of datapoints in an OTLP/HTTP JSON `ExportMetricsServiceRequest`. +/// Resource attributes mirror the live export (`hive-serve-otel`): a fixed +/// `service.name` plus this agent's name and the hive/swarm names. +fn build_payload(agent: &str, hive: &str, swarm: &str, points: &[Value]) -> Value { + json!({ + "resourceMetrics": [{ + "resource": {"attributes": [ + {"key": "service.name", "value": {"stringValue": "hyperhive-agent"}}, + {"key": "agent", "value": {"stringValue": agent}}, + {"key": "hive", "value": {"stringValue": hive}}, + {"key": "swarm", "value": {"stringValue": swarm}}, + ]}, + "scopeMetrics": [{ + "scope": {"name": "hyperhive-migration"}, + "metrics": [{ + "name": METRIC, + "unit": "tokens", + "sum": { + // 2 = AGGREGATION_TEMPORALITY_CUMULATIVE + "aggregationTemporality": 2, + "isMonotonic": true, + "dataPoints": points, + }, + }], + }], + }], + }) +} + +/// POST one OTLP/HTTP JSON payload to `/v1/metrics`. +async fn post(client: &reqwest::Client, url: &str, payload: &Value) -> Result<()> { + let body = serde_json::to_vec(payload)?; + let resp = client + .post(url) + .header("content-type", "application/json") + .body(body) + .send() + .await + .context("send to collector")?; + let status = resp.status(); + if !status.is_success() { + let txt = resp.text().await.unwrap_or_default(); + anyhow::bail!("collector returned {status}: {txt}"); + } + Ok(()) +} From d9d5495f1f52669429a667a073c9be0f6382b425 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 24 Jun 2026 21:05:37 +0200 Subject: [PATCH 2/4] docs: regenerate hivectl-cli.md for the migrate-stats verb --- docs/tools/hivectl-cli.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/docs/tools/hivectl-cli.md b/docs/tools/hivectl-cli.md index 49b83751..6349c71b 100644 --- a/docs/tools/hivectl-cli.md +++ b/docs/tools/hivectl-cli.md @@ -35,6 +35,7 @@ This document contains the help content for the `hivectl` command-line program. * [`hivectl quota limit`↴](#hivectl-quota-limit) * [`hivectl subvol`↴](#hivectl-subvol) * [`hivectl subvol upgrade`↴](#hivectl-subvol-upgrade) +* [`hivectl migrate-stats`↴](#hivectl-migrate-stats) * [`hivectl completions`↴](#hivectl-completions) ## `hivectl` @@ -57,6 +58,7 @@ Sibling to the `hive-c0re` daemon binary. Covers host-side admin operations that * `restart` — Restart containers hive-wide — `stop` then `start` over the same scope. Bare `hivectl restart` restarts **everything** (all sub-agents plus the ci/forge/gateway/matrix infra containers); the same scope flags as `stop`/`start` narrow it (`--agents`, `--ci`, `--forge`, `--gateway`, `--matrix`, `--agent `). If the stop phase reports a failure the start phase is skipped so the operator can investigate. Requires the hive-c0re daemon * `quota` — Per-agent disk accounting + optional quotas via btrfs qgroups * `subvol` — btrfs subvolume management for agent state dirs +* `migrate-stats` — One-shot backfill of historical per-agent turn stats into the OTEL collector as cumulative `claude_code.token.usage` metrics, so pre-OTEL token history shows up in grafana alongside the live stream. Reads each agent's `hyperhive-turn-stats.sqlite` and POSTs OTLP/HTTP JSON to the collector. Every datapoint is tagged `hyperhive-migration="true"`. Safe to re-run (cumulative samples at fixed historical timestamps are idempotent on the backend) * `completions` — Generate a shell completion script for `hivectl` and print it to stdout ###### **Options:** @@ -513,6 +515,20 @@ Convert an existing plain-dir agent state root into a btrfs subvolume in place. +## `hivectl migrate-stats` + +One-shot backfill of historical per-agent turn stats into the OTEL collector as cumulative `claude_code.token.usage` metrics, so pre-OTEL token history shows up in grafana alongside the live stream. Reads each agent's `hyperhive-turn-stats.sqlite` and POSTs OTLP/HTTP JSON to the collector. Every datapoint is tagged `hyperhive-migration="true"`. Safe to re-run (cumulative samples at fixed historical timestamps are idempotent on the backend) + +**Usage:** `hivectl migrate-stats [OPTIONS]` + +###### **Options:** + +* `--endpoint ` — OTLP collector base URL (e.g. `http://10.42.0.1:4318`). Defaults to `$OTEL_EXPORTER_OTLP_ENDPOINT`, then `$HYPERHIVE_OTEL_ENDPOINT` +* `--agent ` — Limit to a single agent (default: every agent with a state dir) +* `--dry-run` — Report datapoint counts without sending anything + + + ## `hivectl completions` Generate a shell completion script for `hivectl` and print it to stdout. From 5393f0e7d44b1afd43094c83252c11474368ec89 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 24 Jun 2026 21:15:30 +0200 Subject: [PATCH 3/4] =?UTF-8?q?feat(#1974):=20migrate-stats=20=E2=80=94=20?= =?UTF-8?q?add=20session.id/start=5Ftype/terminal.type=20+=20session.count?= =?UTF-8?q?=20(review)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hive-c0re/src/otel_migrate.rs | 158 +++++++++++++++++++++++++--------- 1 file changed, 117 insertions(+), 41 deletions(-) diff --git a/hive-c0re/src/otel_migrate.rs b/hive-c0re/src/otel_migrate.rs index e67d4887..e7d2342a 100644 --- a/hive-c0re/src/otel_migrate.rs +++ b/hive-c0re/src/otel_migrate.rs @@ -1,18 +1,27 @@ //! One-shot backfill of historical per-agent turn stats into the OTEL -//! collector as `claude_code.token.usage` metrics, so pre-OTEL history -//! shows up alongside the live telemetry stream. Driven by -//! `hivectl migrate-stats`. +//! collector, so pre-OTEL history shows up alongside the live telemetry +//! stream. Driven by `hivectl migrate-stats`. //! //! Reads each agent's `hyperhive-turn-stats.sqlite` (host path via -//! `Coordinator::agent_harness_dir`) and turns the four per-turn token -//! columns into **cumulative** `claude_code.token.usage` counter samples: -//! one series per `(type, model)`, value = the running total at each -//! turn's end time. Cumulative — not delta — because Prometheus/Mimir -//! family backends silently drop delta sums (the same reason the live -//! export forces cumulative temporality in harness-base.nix). The samples -//! are sent as OTLP/HTTP JSON to `/v1/metrics`. Every -//! datapoint carries `hyperhive-migration="true"` so the backfill is -//! distinguishable from the live stream. +//! `Coordinator::agent_harness_dir`) and emits two **cumulative** counter +//! metrics, matching the live Claude Code shape so the backfill merges +//! into the same series: +//! +//! - `claude_code.token.usage` — the four per-turn token columns, one +//! running-total series per `(type, model, session.id)`, sampled at each +//! turn's end time. +//! - `claude_code.session.count` — one increment per turn, split into +//! `start_type` = `fresh` (a session's first turn) / `continue` (the +//! rest), mirroring the per-turn `--continue` model. +//! +//! Cumulative — not delta — because Prometheus/Mimir family backends +//! silently drop delta sums (the same reason the live export forces +//! cumulative temporality in harness-base.nix). Sent as OTLP/HTTP JSON to +//! `/v1/metrics`. Every datapoint carries +//! `hyperhive-migration="true"` (so the backfill is distinguishable from +//! the live stream) and `terminal.type="non-interactive"`. `session.id` is +//! the db's own session id verbatim when present, else a synthetic +//! reproducible `---migration-`. //! //! Timestamps: `turn_stats.{started_at,ended_at}` are epoch *seconds* //! (see `hive-ag3nt` turn.rs), scaled to nanoseconds for OTLP. @@ -20,12 +29,14 @@ use anyhow::{Context, Result}; use rusqlite::Connection; use serde_json::{Value, json}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::Path; use crate::coordinator::Coordinator; -const METRIC: &str = "claude_code.token.usage"; +const TOKEN_METRIC: &str = "claude_code.token.usage"; +const SESSION_METRIC: &str = "claude_code.session.count"; +const TERMINAL_TYPE: &str = "non-interactive"; /// Cap datapoints per POST so a long-lived agent's history is chunked /// into reasonably-sized OTLP requests rather than one huge body. @@ -40,6 +51,12 @@ const TOKEN_TYPES: &[(&str, usize)] = &[ ("cacheCreation", 6), ]; +/// Datapoints collected from one agent's stats db, split by metric. +struct AgentPoints { + token: Vec, + session: Vec, +} + /// Run the migration. `endpoint` falls back to `OTEL_EXPORTER_OTLP_ENDPOINT` /// then `HYPERHIVE_OTEL_ENDPOINT`. `only_agent` limits to one agent. /// `dry_run` reports counts without sending. @@ -74,23 +91,35 @@ pub async fn run( if !db.exists() { continue; } - let points = - collect_agent_points(&db).with_context(|| format!("read turn_stats for {agent}"))?; - if points.is_empty() { + let id_prefix = format!("{swarm}-{hive}-{agent}"); + let pts = collect_agent_points(&db, &id_prefix) + .with_context(|| format!("read turn_stats for {agent}"))?; + let n = pts.token.len() + pts.session.len(); + if n == 0 { println!("{agent}: no token rows, skipping"); continue; } hit_agents += 1; - total_points += points.len(); - println!("{agent}: {} datapoints", points.len()); + total_points += n; + println!( + "{agent}: {} token + {} session datapoints", + pts.token.len(), + pts.session.len() + ); if dry_run { continue; } - for chunk in points.chunks(MAX_DATAPOINTS_PER_POST) { - let payload = build_payload(agent, &hive, &swarm, chunk); + for chunk in pts.token.chunks(MAX_DATAPOINTS_PER_POST) { + let payload = build_payload(agent, &hive, &swarm, TOKEN_METRIC, "tokens", chunk); post(&client, &url, &payload) .await - .with_context(|| format!("POST metrics for {agent}"))?; + .with_context(|| format!("POST token.usage for {agent}"))?; + } + for chunk in pts.session.chunks(MAX_DATAPOINTS_PER_POST) { + let payload = build_payload(agent, &hive, &swarm, SESSION_METRIC, "", chunk); + post(&client, &url, &payload) + .await + .with_context(|| format!("POST session.count for {agent}"))?; } } println!( @@ -104,15 +133,16 @@ pub async fn run( Ok(()) } -/// Build cumulative `claude_code.token.usage` datapoints for one agent's -/// stats db. One running-total series per `(type, model)`; a datapoint is -/// emitted whenever that series' value changes (token delta > 0). -fn collect_agent_points(db: &Path) -> Result> { +/// Build cumulative datapoints for one agent's stats db. `id_prefix` is +/// `--`, used to synthesise a `session.id` for rows +/// whose `session_id` FK is NULL (pre-session-tracking turns). +fn collect_agent_points(db: &Path, id_prefix: &str) -> Result { let conn = Connection::open_with_flags(db, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY) .context("open stats db read-only")?; + // session_id last so the token column indices (3..=6) stay stable. let mut stmt = conn.prepare( "SELECT started_at, ended_at, model, input_tokens, output_tokens, \ - cache_read_input_tokens, cache_creation_input_tokens \ + cache_read_input_tokens, cache_creation_input_tokens, session_id \ FROM turn_stats ORDER BY ended_at ASC", )?; let rows = stmt.query_map([], |r| { @@ -124,45 +154,91 @@ fn collect_agent_points(db: &Path) -> Result> { r.get::<_, i64>(4)?, r.get::<_, i64>(5)?, r.get::<_, i64>(6)?, + r.get::<_, Option>(7)?, )) })?; - // (type, model) -> (running total, series start time in nanos string) - let mut series: HashMap<(&'static str, String), (u64, String)> = HashMap::new(); - let mut points: Vec = Vec::new(); + // token.usage: (type, model, session.id) -> (running total, series start nanos). + let mut token_series: HashMap<(&'static str, String, String), (u64, String)> = HashMap::new(); + // session.count: start_type -> (running count, series start nanos). + let mut session_counts: HashMap<&'static str, (u64, String)> = HashMap::new(); + let mut seen_sessions: HashSet = HashSet::new(); + let mut migration_idx: u64 = 0; + let mut token = Vec::new(); + let mut session = Vec::new(); + for row in rows { - let (started, ended, model, inp, out, cache_read, cache_create) = row?; + let (started, ended, model, inp, out, cache_read, cache_create, sid) = row?; let cols = [inp, out, cache_read, cache_create]; let start_nanos = u64::try_from(started).unwrap_or(0) * 1_000_000_000; let time_nanos = (u64::try_from(ended).unwrap_or(0) * 1_000_000_000).to_string(); + + // session.id (verbatim db id, else synthetic) + start_type (a + // session's first turn is fresh; every later turn is continue). + let (session_id, is_fresh) = if let Some(id) = sid { + (id.to_string(), seen_sessions.insert(id)) + } else { + let s = format!("{id_prefix}-migration-{migration_idx}"); + migration_idx += 1; + (s, true) + }; + let start_type = if is_fresh { "fresh" } else { "continue" }; + + // session.count: +1 this turn on the start_type series. + let scount = session_counts + .entry(start_type) + .or_insert_with(|| (0u64, start_nanos.to_string())); + scount.0 += 1; + session.push(json!({ + "asInt": scount.0.to_string(), + "startTimeUnixNano": scount.1.clone(), + "timeUnixNano": time_nanos, + "attributes": [ + {"key": "start_type", "value": {"stringValue": start_type}}, + {"key": "terminal.type", "value": {"stringValue": TERMINAL_TYPE}}, + {"key": "hyperhive-migration", "value": {"stringValue": "true"}}, + ], + })); + + // token.usage: per (type, model, session.id) running total. for &(type_, idx) in TOKEN_TYPES { let val = cols[idx - 3]; if val <= 0 { continue; } - let entry = series - .entry((type_, model.clone())) + let entry = token_series + .entry((type_, model.clone(), session_id.clone())) .or_insert_with(|| (0u64, start_nanos.to_string())); entry.0 += u64::try_from(val).unwrap_or(0); - points.push(json!({ + token.push(json!({ "asInt": entry.0.to_string(), "startTimeUnixNano": entry.1.clone(), "timeUnixNano": time_nanos, "attributes": [ {"key": "type", "value": {"stringValue": type_}}, {"key": "model", "value": {"stringValue": model.clone()}}, + {"key": "session.id", "value": {"stringValue": session_id.clone()}}, + {"key": "terminal.type", "value": {"stringValue": TERMINAL_TYPE}}, {"key": "hyperhive-migration", "value": {"stringValue": "true"}}, ], })); } } - Ok(points) + Ok(AgentPoints { token, session }) } -/// Wrap a chunk of datapoints in an OTLP/HTTP JSON `ExportMetricsServiceRequest`. -/// Resource attributes mirror the live export (`hive-serve-otel`): a fixed -/// `service.name` plus this agent's name and the hive/swarm names. -fn build_payload(agent: &str, hive: &str, swarm: &str, points: &[Value]) -> Value { +/// Wrap a chunk of datapoints in an OTLP/HTTP JSON `ExportMetricsServiceRequest` +/// for one cumulative monotonic Sum metric. Resource attributes mirror the +/// live export (`hive-serve-otel`): a fixed `service.name` plus this agent's +/// name and the hive/swarm names. +fn build_payload( + agent: &str, + hive: &str, + swarm: &str, + metric: &str, + unit: &str, + points: &[Value], +) -> Value { json!({ "resourceMetrics": [{ "resource": {"attributes": [ @@ -174,8 +250,8 @@ fn build_payload(agent: &str, hive: &str, swarm: &str, points: &[Value]) -> Valu "scopeMetrics": [{ "scope": {"name": "hyperhive-migration"}, "metrics": [{ - "name": METRIC, - "unit": "tokens", + "name": metric, + "unit": unit, "sum": { // 2 = AGGREGATION_TEMPORALITY_CUMULATIVE "aggregationTemporality": 2, From a94fb9955f90ac1db28ba8bedab79014c2edacf1 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 24 Jun 2026 21:17:42 +0200 Subject: [PATCH 4/4] fix(#1974): pair token cols by array position (argus review) --- hive-c0re/src/otel_migrate.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/hive-c0re/src/otel_migrate.rs b/hive-c0re/src/otel_migrate.rs index e7d2342a..0fb8d687 100644 --- a/hive-c0re/src/otel_migrate.rs +++ b/hive-c0re/src/otel_migrate.rs @@ -42,13 +42,13 @@ const TERMINAL_TYPE: &str = "non-interactive"; /// into reasonably-sized OTLP requests rather than one huge body. const MAX_DATAPOINTS_PER_POST: usize = 1000; -/// `claude_code.token.usage` `type` attribute value paired with the -/// `turn_stats` column it maps to (column index in the SELECT below). +/// `claude_code.token.usage` `type` attribute value paired with its +/// position in the per-row `cols` array (built in row order below). const TOKEN_TYPES: &[(&str, usize)] = &[ - ("input", 3), - ("output", 4), - ("cacheRead", 5), - ("cacheCreation", 6), + ("input", 0), + ("output", 1), + ("cacheRead", 2), + ("cacheCreation", 3), ]; /// Datapoints collected from one agent's stats db, split by metric. @@ -201,8 +201,8 @@ fn collect_agent_points(db: &Path, id_prefix: &str) -> Result { })); // token.usage: per (type, model, session.id) running total. - for &(type_, idx) in TOKEN_TYPES { - let val = cols[idx - 3]; + for &(type_, pos) in TOKEN_TYPES { + let val = cols[pos]; if val <= 0 { continue; }