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

fix(hive-metric): use opentelemetry-otlp SDK instead of hand-rolled OTLP

Browse source
This commit is contained in:
damocles 2026-07-04 23:00:06 +02:00 • committed by mara
commit b4b5dd7fa4
2 changed files with 59 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

11
hive-metric/Cargo.toml
View file

@ -10,6 +10,13 @@ path = "src/main.rs"
[dependencies]
anyhow.workspace = true
clap.workspace = true
reqwest.workspace = true
serde_json.workspace = true
# OTEL SDK — only used by this crate, so kept local rather than in workspace.dependencies.
opentelemetry = "0.32"
opentelemetry_sdk = { version = "0.32", features = ["metrics"] }
opentelemetry-otlp = { version = "0.32", default-features = false, features = [
"metrics",
"http-json",
"reqwest-client",
"reqwest-rustls",
] }
tokio = { workspace = true, features = ["rt", "macros"] }

155
hive-metric/src/main.rs
View file

@ -1,28 +1,25 @@
//! `hive-metric` — push a single labeled metric to the OTEL collector.
//!
//! Reads `OTEL_EXPORTER_OTLP_ENDPOINT` and `OTEL_EXPORTER_OTLP_HEADERS`
//! from the environment (set by the harness via claude's managed settings
//! when `services.hyperhive.otel.enable = true`). Resource labels (agent,
//! hive, swarm, service.name) are inherited from `OTEL_RESOURCE_ATTRIBUTES`,
//! which the harness already populates per-agent — the caller does not need
//! to re-specify them.
//! Uses the OpenTelemetry Rust SDK with the OTLP HTTP/JSON exporter to emit
//! metrics. Standard OTEL env vars are read automatically by the SDK:
//!
//! - `OTEL_EXPORTER_OTLP_ENDPOINT` — collector URL (required)
//! - `OTEL_EXPORTER_OTLP_HEADERS` — auth headers (`Key=Value,...`)
//! - `OTEL_RESOURCE_ATTRIBUTES` — resource labels (`k=v,...`)
//!
//! The harness populates all of these per-agent when
//! `services.hyperhive.otel.enable = true`.
//!
//! # Usage
//! ```text
//! hive-metric <name> <value> [--type gauge|counter] [--labels key=value...]
//! ```
//!
//! # Examples
//! ```text
//! hive-metric tasks_completed 42
//! hive-metric errors_total 3 --type counter --labels phase=scan
//! hive-metric latency_ms 250.5 --labels model=sonnet --labels tier=api
//! ```
use anyhow::{Context, Result, bail};
use clap::Parser;
use serde_json::{Value, json};
use std::time::{SystemTime, UNIX_EPOCH};
use opentelemetry::KeyValue;
use opentelemetry::metrics::MeterProvider;
use opentelemetry_otlp::{Protocol, WithExportConfig};
#[derive(Parser)]
#[command(
@ -30,27 +27,26 @@ use std::time::{SystemTime, UNIX_EPOCH};
about = "Push a labeled metric to the OTEL collector (requires services.hyperhive.otel.enable = true)"
)]
struct Cli {
/// Metric name (e.g. `tasks_completed`, `errors_total`).
/// Metric name (e.g. `tasks_completed`, `latency_ms`).
name: String,
/// Numeric value (f64). Floats and integers both accepted.
value: f64,
/// Metric kind: `gauge` (instantaneous value, default) or `counter`
/// (monotonically increasing sum, cumulative temporality).
/// Metric kind: `gauge` (instantaneous, default) or `counter` (cumulative sum).
#[arg(long = "type", value_name = "TYPE", default_value = "gauge")]
metric_type: MetricKind,
/// Extra label(s) as `key=value` pairs. May be repeated.
/// Resource labels (agent, hive, swarm) are inherited automatically
/// from `OTEL_RESOURCE_ATTRIBUTES` and must not be duplicated here.
/// Resource labels (agent, hive, swarm) come from `OTEL_RESOURCE_ATTRIBUTES`
/// automatically — do not re-specify them here.
#[arg(long, value_name = "KEY=VALUE")]
labels: Vec<String>,
}
#[derive(Clone, Debug, clap::ValueEnum)]
enum MetricKind {
/// Gauge: represents a value at a point in time (default).
/// Gauge: point-in-time value (default).
Gauge,
/// Counter: monotonically increasing cumulative sum.
Counter,
@ -60,110 +56,51 @@ enum MetricKind {
async fn main() -> Result<()> {
let cli = Cli::parse();
let endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").context(
"OTEL_EXPORTER_OTLP_ENDPOINT not set — enable OTEL in NixOS config \
(services.hyperhive.otel.enable = true)",
)?;
let metrics_url = format!("{}/v1/metrics", endpoint.trim_end_matches('/'));
// Fail early with a clear message when OTEL is not configured.
if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err() {
bail!(
"OTEL_EXPORTER_OTLP_ENDPOINT not set — \
enable OTEL in NixOS config (services.hyperhive.otel.enable = true)"
);
}
let now_ns = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
.to_string();
let labels = parse_labels(&cli.labels)?;
let resource_attrs = parse_kv_list(&std::env::var("OTEL_RESOURCE_ATTRIBUTES").unwrap_or_default(), ',');
let data_attrs = parse_label_args(&cli.labels)?;
let exporter = opentelemetry_otlp::MetricExporter::builder()
.with_http()
.with_protocol(Protocol::HttpJson)
.build()
.context("failed to build OTLP metric exporter")?;
let data_point = json!({
"attributes": data_attrs,
"startTimeUnixNano": now_ns,
"timeUnixNano": now_ns,
"asDouble": cli.value,
});
// SDK automatically reads OTEL_RESOURCE_ATTRIBUTES for resource labels.
let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder()
.with_periodic_exporter(exporter)
.build();
let metric = match cli.metric_type {
MetricKind::Gauge => json!({
"name": cli.name,
"gauge": { "dataPoints": [data_point] },
}),
MetricKind::Counter => json!({
"name": cli.name,
"sum": {
"dataPoints": [data_point],
// AGGREGATION_TEMPORALITY_CUMULATIVE = 2
"aggregationTemporality": 2,
"isMonotonic": true,
},
}),
};
let meter = provider.meter("hive-metric");
let payload = json!({
"resourceMetrics": [{
"resource": { "attributes": resource_attrs },
"scopeMetrics": [{
"scope": { "name": "hive-metric", "version": "0.1.0" },
"metrics": [metric],
}],
}]
});
let mut builder = reqwest::Client::new()
.post(&metrics_url)
.header("content-type", "application/json")
.json(&payload);
// OTEL_EXPORTER_OTLP_HEADERS: `Key=Value,Key2=Value2` (comma-separated).
if let Ok(h) = std::env::var("OTEL_EXPORTER_OTLP_HEADERS") {
for (k, v) in parse_kv_list(&h, ',')
.into_iter()
.filter_map(|kv| {
let k = kv.get("key")?.as_str()?.to_owned();
let v = kv
.get("value")?
.get("stringValue")?
.as_str()?
.to_owned();
Some((k, v))
})
{
builder = builder.header(k, v);
match cli.metric_type {
MetricKind::Gauge => {
meter.f64_gauge(cli.name).build().record(cli.value, &labels);
}
MetricKind::Counter => {
meter.f64_counter(cli.name).build().add(cli.value, &labels);
}
}
let resp = builder
.send()
.await
.with_context(|| format!("POST to {metrics_url} failed"))?;
if !resp.status().is_success() {
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
bail!("OTEL endpoint {metrics_url} returned {status}: {body}");
}
// Flush pending metrics and shut down cleanly.
provider.shutdown().context("failed to flush OTLP metrics")?;
Ok(())
}
/// Parse a `key=value` delimited list into OTLP attribute objects.
/// Entries that lack `=` are silently skipped.
fn parse_kv_list(s: &str, sep: char) -> Vec<Value> {
s.split(sep)
.filter_map(|entry| {
let (k, v) = entry.split_once('=')?;
Some(json!({ "key": k.trim(), "value": { "stringValue": v.trim() } }))
})
.collect()
}
/// Parse `--labels k=v` CLI args into OTLP attribute objects.
fn parse_label_args(args: &[String]) -> Result<Vec<Value>> {
fn parse_labels(args: &[String]) -> Result<Vec<KeyValue>> {
args.iter()
.map(|s| {
let (k, v) = s
.split_once('=')
.with_context(|| format!("--labels value must be key=value, got {s:?}"))?;
Ok(json!({ "key": k.trim(), "value": { "stringValue": v.trim() } }))
.with_context(|| format!("--labels must be key=value, got {s:?}"))?;
Ok(KeyValue::new(k.trim().to_owned(), v.trim().to_owned()))
})
.collect()
}