diff --git a/docs/tools/hivectl-cli.md b/docs/tools/hivectl-cli.md index 6349c71b..49b83751 100644 --- a/docs/tools/hivectl-cli.md +++ b/docs/tools/hivectl-cli.md @@ -35,7 +35,6 @@ 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` @@ -58,7 +57,6 @@ 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:** @@ -515,20 +513,6 @@ 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. diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index 474907da..4379f1bf 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -204,26 +204,6 @@ 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): @@ -628,11 +608,6 @@ 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 257995f7..53f66241 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -42,7 +42,6 @@ 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 deleted file mode 100644 index 2a8acb78..00000000 --- a/hive-c0re/src/otel_migrate.rs +++ /dev/null @@ -1,308 +0,0 @@ -//! One-shot backfill of historical per-agent turn stats into the OTEL -//! 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 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. - -use anyhow::{Context, Result}; -use rusqlite::Connection; -use serde_json::{Value, json}; -use std::collections::{HashMap, HashSet}; -use std::path::Path; - -use crate::coordinator::Coordinator; - -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. -const MAX_DATAPOINTS_PER_POST: usize = 1000; - -/// `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", 0), - ("output", 1), - ("cacheRead", 2), - ("cacheCreation", 3), -]; - -/// 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. -/// -/// # 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; - let mut failed = 0usize; - for agent in &agents { - let db = Coordinator::agent_harness_dir(agent).join("hyperhive-turn-stats.sqlite"); - if !db.exists() { - continue; - } - let id_prefix = format!("{swarm}-{hive}-{agent}"); - // Skip + warn on a bad db (e.g. an old schema) rather than aborting - // the whole migration on the first failed agent. - let pts = match collect_agent_points(&db, &id_prefix) { - Ok(p) => p, - Err(e) => { - eprintln!("{agent}: skipped — {e:#}"); - failed += 1; - continue; - } - }; - let n = pts.token.len() + pts.session.len(); - if n == 0 { - println!("{agent}: no token rows, skipping"); - continue; - } - hit_agents += 1; - total_points += n; - println!( - "{agent}: {} token + {} session datapoints", - pts.token.len(), - pts.session.len() - ); - if dry_run { - continue; - } - 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 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!( - "migrate-stats: {hit_agents} agent(s) with stats, {total_points} datapoints {}{}", - if dry_run { - "(dry-run, nothing sent)" - } else { - "sent" - }, - if failed > 0 { - format!(" ({failed} agent(s) skipped on error)") - } else { - String::new() - } - ); - Ok(()) -} - -/// 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")?; - // Older dbs predate the `session_id` migration. Probe for the column and - // fall back to a `NULL AS session_id` placeholder so the row shape (col 7) - // is identical either way (those rows get a synthetic session.id below). - // session_id is selected last so the token column indices stay fixed. - let has_session = conn - .prepare("SELECT session_id FROM turn_stats LIMIT 1") - .is_ok(); - let sql = if has_session { - "SELECT started_at, ended_at, model, input_tokens, output_tokens, \ - cache_read_input_tokens, cache_creation_input_tokens, session_id \ - FROM turn_stats ORDER BY ended_at ASC" - } else { - "SELECT started_at, ended_at, model, input_tokens, output_tokens, \ - cache_read_input_tokens, cache_creation_input_tokens, NULL AS session_id \ - FROM turn_stats ORDER BY ended_at ASC" - }; - let mut stmt = conn.prepare(sql)?; - 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)?, - r.get::<_, Option>(7)?, - )) - })?; - - // 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, 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_, pos) in TOKEN_TYPES { - let val = cols[pos]; - if val <= 0 { - continue; - } - 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); - 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(AgentPoints { token, session }) -} - -/// 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": [ - {"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": unit, - "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(()) -}