From 652c0e96143e69a4fe32106c91d2f9e5e315b8d7 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 4 Nov 2026 22:59:29 +0100 Subject: [PATCH] swarm-claude-login: export login step, duration, expiry and active session metrics --- Cargo.lock | 2 + docs/swarm/claude-login.md | 18 +++ nix/host-modules/swarm-claude-login.nix | 9 ++ swarm-claude-login/Cargo.toml | 10 +- swarm-claude-login/src/login.rs | 13 +- swarm-claude-login/src/main.rs | 57 ++++--- swarm-claude-login/src/metrics.rs | 205 ++++++++++++++++++++++++ 7 files changed, 292 insertions(+), 22 deletions(-) create mode 100644 swarm-claude-login/src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index 57a9137f..77480d01 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4970,6 +4970,8 @@ dependencies = [ "futures-util", "hive-log", "hive-types", + "opentelemetry", + "opentelemetry_sdk", "serde_json", "swarm-queue-client", "swarm-secret-client", diff --git a/docs/swarm/claude-login.md b/docs/swarm/claude-login.md index f06d0115..84de3bcb 100644 --- a/docs/swarm/claude-login.md +++ b/docs/swarm/claude-login.md @@ -84,3 +84,21 @@ newer one. The agent's own web UI login (`/login`) still works, unchanged. It goes away once the operator has confirmed the swarm login works. + +## Observability + +The minter pushes OTLP metrics through its container's forwarder to the swarm +collector, labelled `service.name=hyperhive-claude-login`. With no +`OTEL_EXPORTER_OTLP_ENDPOINT` it exports nothing. The client-certificate +gauges (`bao.client_cert.*`) come with them. + +| Metric | Unit | Kind | Attributes | +| ----------------------------------------- | ---- | --------- | --------------------------------------------------------------------------------------------------- | +| `hyperhive.claude_login.steps` | — | counter | `step`: `start`, `code`, `cancel`, `invalid`; `outcome`: `started`, `stored`, `cancelled`, `failed` | +| `hyperhive.claude_login.step.duration` | `s` | histogram | `step` | +| `hyperhive.claude_login.sessions.expired` | — | counter | — | +| `hyperhive.claude_login.sessions.active` | — | gauge | logins waiting for a code, as of the last step | + +`invalid` is a request that did not decode or named no agent. A `start` spawns +`claude auth login` and a `code` stores the credentials, so those two are the +slow steps. `expired` counts logins stopped because their code never came. diff --git a/nix/host-modules/swarm-claude-login.nix b/nix/host-modules/swarm-claude-login.nix index c38eff80..53f95692 100644 --- a/nix/host-modules/swarm-claude-login.nix +++ b/nix/host-modules/swarm-claude-login.nix @@ -53,6 +53,9 @@ let # Loopback in the container's own netns, and distinct from the other service # containers' forwarders all the same. forwarderTelemetryPort = 8893; + # The standard OTLP/HTTP port, on the container's loopback: the forwarder + # takes the minter's metrics there and pushes them to the swarm collector. + forwarderOtlpPort = 4318; rootfsCheck = import ./lib/container-rootfs-check.nix { inherit config; }; @@ -201,6 +204,7 @@ in inherit clientId; tokenUrl = "${autheliaCfg.url}/api/oidc/token"; telemetryPort = forwarderTelemetryPort; + otlpMetricsPort = forwarderOtlpPort; store = { package = baoDeploy.package; addr = "https://${baoCfg.domain}:${toString baoCfg.port}"; @@ -229,6 +233,11 @@ in SWARM_CLAUDE_LOGIN_OIDC_TOKEN_ENDPOINT = "${autheliaCfg.url}/api/oidc/token"; SWARM_CLAUDE_LOGIN_OIDC_CLIENT_ID = clientId; SWARM_CLAUDE_LOGIN_CLAUDE = "${claudeCode pkgs.claude-code}/bin/claude"; + OTEL_EXPORTER_OTLP_ENDPOINT = "http://127.0.0.1:${toString forwarderOtlpPort}"; + OTEL_EXPORTER_OTLP_PROTOCOL = "http/protobuf"; + OTEL_RESOURCE_ATTRIBUTES = "service.name=hyperhive-claude-login,swarm=${ + if hyperhiveCfg.swarm.name == null then "unknown" else hyperhiveCfg.swarm.name + }"; } // lib.optionalAttrs (storeFiles ? claude-login-bao-server-ca) { BAO_CACERT = credential "claude-login-bao-server-ca"; diff --git a/swarm-claude-login/Cargo.toml b/swarm-claude-login/Cargo.toml index b1af0ca1..923f5f7e 100644 --- a/swarm-claude-login/Cargo.toml +++ b/swarm-claude-login/Cargo.toml @@ -14,17 +14,25 @@ async-nats.workspace = true # StreamExt::next on the subscription. futures-util.workspace = true hive-log.workspace = true +# The metrics the minter exports (`metrics`), pushed through the OTLP exporter +# `swarm-secret-client`'s `meter_provider` builds. +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true serde_json.workspace = true # The login subject and the shapes the controller sends and reads back. # The agent name a request carries is parsed as the controller parses it. hive-types.workspace = true swarm-queue-client.workspace = true # The credentials and queue client secret paths, and the store login. -swarm-secret-client.workspace = true +swarm-secret-client = { workspace = true, features = ["otel"] } # The login subject and the requests and replies it carries. swarm-wire.workspace = true tokio.workspace = true tracing.workspace = true +[dev-dependencies] +# `InMemoryMetricExporter`, to read back what `metrics` records. +opentelemetry_sdk = { workspace = true, features = ["testing"] } + [lints] workspace = true diff --git a/swarm-claude-login/src/login.rs b/swarm-claude-login/src/login.rs index 545071ac..dac9bb88 100644 --- a/swarm-claude-login/src/login.rs +++ b/swarm-claude-login/src/login.rs @@ -217,14 +217,21 @@ impl Logins { Reply::Cancelled } + /// How many logins wait for a code. + pub async fn active(&self) -> usize { + self.running.lock().await.len() + } + /// Stop `agent`'s login `id` if it is still the one running: called once - /// its time to wait for a code is up. - pub async fn expire(&self, agent: &str, id: u64) { + /// its time to wait for a code is up. Whether it stopped one. + pub async fn expire(&self, agent: &str, id: u64) -> bool { let mut running = self.running.lock().await; - if running.get(agent).is_some_and(|r| r.id == id) { + let stop = running.get(agent).is_some_and(|r| r.id == id); + if stop { running.remove(agent); tracing::info!(%agent, "a login waited too long for its code; stopped"); } + stop } fn spawn(&self, agent: &str, id: u64) -> anyhow::Result { diff --git a/swarm-claude-login/src/main.rs b/swarm-claude-login/src/main.rs index 9cc7cd64..67cbc265 100644 --- a/swarm-claude-login/src/main.rs +++ b/swarm-claude-login/src/main.rs @@ -13,15 +13,18 @@ use std::path::PathBuf; use std::sync::Arc; +use std::time::Instant; use anyhow::{Context, Result, bail}; use futures_util::StreamExt as _; +use metrics::Metrics; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::claude_login::{Credentials, client_secret_path, credentials_path}; use swarm_secret_client::stored_secret::StoredSecret; use swarm_wire::claude_login::{Reply, Request, SESSION_DEADLINE, STEP_DEADLINE, SUBJECT}; mod login; +mod metrics; /// The cert-auth role this service logs in to the store as. `swarm-bao.nix` /// writes it, matching the leaf's subject. @@ -57,6 +60,7 @@ async fn main() -> Result<()> { std::env::var(&full).with_context(|| format!("{full} is unset")) }; let client_id = var("OIDC_CLIENT_ID")?; + let metrics = Arc::new(Metrics::from_env()); let logins = Arc::new(login::Logins::new( PathBuf::from(var("CLAUDE")?), std::env::temp_dir(), @@ -97,8 +101,9 @@ async fn main() -> Result<()> { }; let queue = queue.clone(); let logins = Arc::clone(&logins); + let metrics = Arc::clone(&metrics); tokio::spawn(async move { - let reply = handle(&msg.payload, &logins).await; + let reply = handle(&msg.payload, &logins, &metrics).await; match serde_json::to_vec(&reply) { Ok(body) => { if let Err(e) = queue.publish(reply_to, body.into()).await { @@ -112,15 +117,28 @@ async fn main() -> Result<()> { bail!("the subscription to {SUBJECT} ended") } -/// Decode one step, run it and log it. Never logs the code or the URL. -async fn handle(payload: &[u8], logins: &Arc) -> Reply { +/// Answer one step and count it. +async fn handle(payload: &[u8], logins: &Arc, metrics: &Arc) -> Reply { + let started = Instant::now(); + let (step, reply) = run_step(payload, logins, metrics).await; + metrics.step(step, &reply, started.elapsed()); + metrics.active(logins.active().await); + reply +} + +/// Decode one step, run it and log it, with the kind of step it was. Never +/// logs the code or the URL. +async fn run_step( + payload: &[u8], + logins: &Arc, + metrics: &Arc, +) -> (&'static str, Reply) { let request: Request = match serde_json::from_slice(payload) { Ok(r) => r, Err(e) => { tracing::warn!(error = %e, "an undecodable login step; answering with the error"); - return Reply::Failed { - error: format!("the minter could not decode the request: {e}"), - }; + let error = format!("the minter could not decode the request: {e}"); + return ("invalid", Reply::Failed { error }); } }; // Before anything else: the name becomes a scratch directory's name and @@ -129,31 +147,34 @@ async fn handle(payload: &[u8], logins: &Arc) -> Reply { &request; if let Err(reason) = hive_types::Ident::parse(agent) { tracing::warn!(agent = ?agent, reason, "a login step for an invalid agent name; refused"); - return Reply::Failed { - error: format!("{agent:?} is not an agent name: {reason}"), - }; + let error = format!("{agent:?} is not an agent name: {reason}"); + return ("invalid", Reply::Failed { error }); } - let (agent, reply) = match request { + let (step, agent, reply) = match request { Request::Start { agent, operator } => { tracing::info!(%agent, %operator, "starting a claude login"); let (reply, id) = logins.start(&agent).await; if let Some(id) = id { let logins = Arc::clone(logins); let expiring = agent.clone(); + let metrics = Arc::clone(metrics); tokio::spawn(async move { tokio::time::sleep(SESSION_DEADLINE).await; - logins.expire(&expiring, id).await; + if logins.expire(&expiring, id).await { + metrics.expired(); + metrics.active(logins.active().await); + } }); } - (agent, reply) + ("start", agent, reply) } Request::Code { agent, code } => { let reply = logins.code(&agent, &code, &AgentLeaf).await; - (agent, reply) + ("code", agent, reply) } Request::Cancel { agent } => { let reply = logins.cancel(&agent).await; - (agent, reply) + ("cancel", agent, reply) } }; match &reply { @@ -162,7 +183,7 @@ async fn handle(payload: &[u8], logins: &Arc) -> Reply { Reply::Cancelled => tracing::info!(%agent, "claude login cancelled"), Reply::Failed { error } => tracing::warn!(%agent, %error, "claude login step failed"), } - reply + (step, reply) } async fn client_secret(client_id: &str) -> Result { @@ -192,7 +213,7 @@ mod tests { use swarm_wire::claude_login::{Reply, Request}; - use super::{handle, login}; + use super::{Metrics, handle, login}; #[tokio::test] async fn a_step_for_an_invalid_agent_name_is_refused_before_anything_is_made() { @@ -223,7 +244,7 @@ mod tests { }, ] { let payload = serde_json::to_vec(&request).expect("encodes"); - let reply = handle(&payload, &logins).await; + let reply = handle(&payload, &logins, &Arc::new(Metrics::noop())).await; assert!( matches!(reply, Reply::Failed { .. }), "{agent:?}: {reply:?}" @@ -247,7 +268,7 @@ mod tests { }) .expect("encodes"); assert!(matches!( - handle(&payload, &logins).await, + handle(&payload, &logins, &Arc::new(Metrics::noop())).await, Reply::Started { .. } )); drop(logins); diff --git a/swarm-claude-login/src/metrics.rs b/swarm-claude-login/src/metrics.rs new file mode 100644 index 00000000..a47b8e40 --- /dev/null +++ b/swarm-claude-login/src/metrics.rs @@ -0,0 +1,205 @@ +//! What the login minter exports for its dashboard, through the swarm +//! collector. +//! +//! - `hyperhive.claude_login.steps`: login steps answered, by `step` (`start`, +//! `code`, `cancel`, `invalid` for a request that did not decode or named no +//! agent) and `outcome` (`started`, `stored`, `cancelled`, `failed`). +//! - `hyperhive.claude_login.step.duration`: seconds to answer one step, by +//! `step`. A `start` spawns `claude auth login` and a `code` stores the +//! credentials, so these are the slow ones. +//! - `hyperhive.claude_login.sessions.expired`: logins stopped because their +//! code never came. +//! - `hyperhive.claude_login.sessions.active`: logins waiting for a code now, +//! as of the last step. +//! +//! The provider is built from the standard `OTEL_*` environment, so without a +//! collector the instruments do nothing. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider as _}; +use opentelemetry_sdk::metrics::SdkMeterProvider; +use swarm_wire::claude_login::Reply; + +const METER: &str = "hyperhive.claude_login"; + +/// Seconds. A `start` waits for `claude auth login` to print its URL and a +/// `code` for the login to finish and the credentials to be stored, each up to +/// the step deadline (a minute); `cancel` and `invalid` answer at once. The SDK's own +/// bounds start at 5 and are made for milliseconds. +const DURATION_BOUNDARIES: [f64; 10] = [0.01, 0.05, 0.25, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0]; + +/// The minter's instruments, and the provider that exports them. +pub struct Metrics { + /// Kept alive: it exports only while it lives. + _provider: Option, + steps: Counter, + duration: Histogram, + expired: Counter, + active: Arc, +} + +impl Metrics { + /// The instruments on the provider the environment names, or no-op ones. + #[must_use] + pub fn from_env() -> Self { + let provider = swarm_secret_client::meter_provider::from_env(METER, "claude-login"); + let meter = provider + .as_ref() + .map_or_else(|| opentelemetry::global::meter(METER), |p| p.meter(METER)); + Self::on(&meter, provider) + } + + fn on(meter: &Meter, provider: Option) -> Self { + let active = Arc::new(AtomicU64::new(0)); + let gauge = Arc::clone(&active); + meter + .u64_observable_gauge(format!("{METER}.sessions.active")) + .with_callback(move |obs| obs.observe(gauge.load(Ordering::Relaxed), &[])) + .build(); + Self { + _provider: provider, + steps: meter.u64_counter(format!("{METER}.steps")).build(), + duration: meter + .f64_histogram(format!("{METER}.step.duration")) + .with_unit("s") + .with_boundaries(DURATION_BOUNDARIES.to_vec()) + .build(), + expired: meter + .u64_counter(format!("{METER}.sessions.expired")) + .build(), + active, + } + } + + /// Instruments that export nothing, for the tests that do not read them. + #[cfg(test)] + pub fn noop() -> Self { + Self::on(&opentelemetry::global::meter(METER), None) + } + + /// One step of kind `step` answered with `reply`, after `took`. + pub fn step(&self, step: &'static str, reply: &Reply, took: Duration) { + let outcome = match reply { + Reply::Started { .. } => "started", + Reply::Stored => "stored", + Reply::Cancelled => "cancelled", + Reply::Failed { .. } => "failed", + }; + self.steps.add( + 1, + &[ + KeyValue::new("step", step), + KeyValue::new("outcome", outcome), + ], + ); + self.duration + .record(took.as_secs_f64(), &[KeyValue::new("step", step)]); + } + + /// A login stopped because its code never came. + pub fn expired(&self) { + self.expired.add(1, &[]); + } + + /// How many logins wait for a code now. + pub fn active(&self, n: usize) { + self.active + .store(u64::try_from(n).unwrap_or(u64::MAX), Ordering::Relaxed); + } +} + +#[cfg(test)] +mod tests { + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader}; + + use super::*; + + fn recorded() -> (Metrics, SdkMeterProvider, InMemoryMetricExporter) { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let metrics = Metrics::on(&provider.meter("test"), None); + (metrics, provider, exporter) + } + + /// Every data point of the u64 sum or gauge `name`: its labels and value. + fn points(exporter: &InMemoryMetricExporter, name: &str) -> Vec<(Vec<(String, String)>, u64)> { + let mut out = Vec::new(); + for resource in exporter.get_finished_metrics().unwrap() { + for scope in resource.scope_metrics() { + for metric in scope.metrics().filter(|m| m.name() == name) { + let labels = |attrs: &mut dyn Iterator| { + attrs + .map(|kv| (kv.key.to_string(), kv.value.to_string())) + .collect::>() + }; + match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + for p in sum.data_points() { + out.push((labels(&mut p.attributes()), p.value())); + } + } + AggregatedMetrics::U64(MetricData::Gauge(gauge)) => { + for p in gauge.data_points() { + out.push((labels(&mut p.attributes()), p.value())); + } + } + _ => {} + } + } + } + } + out + } + + fn pair(k: &str, v: &str) -> (String, String) { + (k.to_owned(), v.to_owned()) + } + + #[test] + fn a_step_counts_under_its_kind_and_the_reply() { + let (metrics, provider, exporter) = recorded(); + let took = Duration::from_millis(3); + metrics.step("code", &Reply::Stored, took); + metrics.step("code", &Reply::Stored, took); + metrics.step( + "code", + &Reply::Failed { + error: "no".to_owned(), + }, + took, + ); + provider.force_flush().unwrap(); + let mut got = points(&exporter, "hyperhive.claude_login.steps"); + got.sort(); + assert_eq!( + got, + vec![ + (vec![pair("step", "code"), pair("outcome", "failed")], 1), + (vec![pair("step", "code"), pair("outcome", "stored")], 2), + ] + ); + } + + #[test] + fn expiries_count_and_the_active_gauge_reports_the_last_count() { + let (metrics, provider, exporter) = recorded(); + metrics.expired(); + metrics.active(3); + metrics.active(1); + provider.force_flush().unwrap(); + assert_eq!( + points(&exporter, "hyperhive.claude_login.sessions.expired"), + vec![(vec![], 1)] + ); + assert_eq!( + points(&exporter, "hyperhive.claude_login.sessions.active"), + vec![(vec![], 1)] + ); + } +}