feat(#1974): hivectl migrate-stats — backfill per-agent token history to otel

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<String>,
+        /// Limit to a single agent (default: every agent with a state dir).
+        #[arg(long)]
+        agent: Option<String>,
+        /// 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,

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;

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 `<endpoint>/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<String>,
+    only_agent: Option<String>,
+    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<Vec<Value>> {
+    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<Value> = 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 `<endpoint>/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(())
+}