vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore(#1931): remove non-working hivectl migrate-stats verb

Browse source
This commit is contained in:
damocles 2026-06-26 18:06:46 +02:00
commit b0c89af817
4 changed files with 0 additions and 350 deletions
Download patch file Download diff file Expand all files Collapse all files

16
docs/tools/hivectl-cli.md
View file

@ -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 <name>`). 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 <ENDPOINT>` — OTLP collector base URL (e.g. `http://10.42.0.1:4318`). Defaults to `$OTEL_EXPORTER_OTLP_ENDPOINT`, then `$HYPERHIVE_OTEL_ENDPOINT`
* `--agent <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.

25
hive-c0re/src/bin/hivectl.rs
View file

@ -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<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):
@ -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,

1
hive-c0re/src/lib.rs
View file

@ -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;

308
hive-c0re/src/otel_migrate.rs
View file

@ -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
//! `<endpoint>/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 `<swarm>-<hive>-<agent>-migration-<n>`.
//!
//! 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<Value>,
session: Vec<Value>,
}
/// 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;
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
/// `<swarm>-<hive>-<agent>`, 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<AgentPoints> {
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<i64>>(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<i64> = 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 `<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(())
}