From 1e8c3981323231ac169267ca573ab444c9919eaa Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 16 Oct 2026 20:53:59 +0200 Subject: [PATCH] swarm-controller: get its otlp pushes past the collector, log failures The collector's oidc/swarm authenticator verifies the bearer offline as a JWT. The controller's authelia client was registered without accessTokenSignedResponseAlg, so authelia mints it an opaque authelia_at_ handle, which that authenticator cannot verify: by config, no push from the vcs or jobq exporter can be accepted. Register it with RS256 like every other client whose token lands on an oidc/* receiver. The SDK logs a failed export at debug only, so a refused push never reached the journal. AuthenticatedHttpClient now warns on a failed token mint, a transport error, or a non-2xx answer (status plus a body snippet), once per kind of failure and again every 10 minutes with a count, and logs once when a push succeeds after failing. Refs #5393 --- docs/scheduler/observability.md | 7 + .../glue-controller-oidc-client.nix | 4 + nix/module-eval/core-toggle.nix | 13 +- swarm-controller/src/otel_http_client.rs | 378 +++++++++++++++++- 4 files changed, 381 insertions(+), 21 deletions(-) diff --git a/docs/scheduler/observability.md b/docs/scheduler/observability.md index 91788e96..3a84cb93 100644 --- a/docs/scheduler/observability.md +++ b/docs/scheduler/observability.md @@ -494,6 +494,13 @@ crate's own doc comment) — `swarm-controller` sets `service.name = swarm-controller` directly rather than reading it from the container environment, since it's a standalone daemon, not a per-agent harness process. +Both exporters push through `swarm-controller`'s own authenticated client, +and the collector's `oidc/swarm` receiver checks the bearer. When a push +fails, `swarm-controller` logs `otel export: push to the collector failed` +at warn, with the kind of failure (`token`, `transport` or `status `) +and the start of the response body. It repeats a persisting failure every +10 minutes with a count, and logs once at info when pushes succeed again. + ## Agent-emitted custom metrics (`hive-metric`) Agents can push arbitrary labeled metrics to the same OTEL collector via the diff --git a/nix/host-modules/glue-controller-oidc-client.nix b/nix/host-modules/glue-controller-oidc-client.nix index b17868e9..255add9b 100644 --- a/nix/host-modules/glue-controller-oidc-client.nix +++ b/nix/host-modules/glue-controller-oidc-client.nix @@ -42,6 +42,10 @@ in # same id; either half without the other is a 401 several layers from # this line. audience = [ queueClientId ]; + # `oidc/swarm` verifies the token offline against `/jwks.json`, which + # authelia's default opaque handle fails. The queue and the authelia + # bridge introspect, which reads a JWT just as well. + accessTokenSignedResponseAlg = "RS256"; } ]; }; diff --git a/nix/module-eval/core-toggle.nix b/nix/module-eval/core-toggle.nix index c10f2b40..c66c2d28 100644 --- a/nix/module-eval/core-toggle.nix +++ b/nix/module-eval/core-toggle.nix @@ -157,11 +157,18 @@ let let env = m: m.systemd.services.swarm-controller.environment; otel = allLocal.services.hyperhive.swarm.otel; + clientId = allLocal.services.hyperhive.swarm.controller.queueClientId; + # ../host-modules/glue-controller-oidc-client.nix: the collector's + # `oidc/swarm` reads only a signed access token. + client = lib.findFirst ( + c: c.id == clientId + ) null allLocal.services.hyperhive.swarm.authelia.oidc.clients; in (env allLocal).OTEL_EXPORTER_OTLP_ENDPOINT or null == "https://${otel.domain}/${otel.producerName}" - && - (env allLocal).SWARM_CONTROLLER_OTEL_AUDIENCE or null - == allLocal.services.hyperhive.swarm.controller.queueClientId + && (env allLocal).SWARM_CONTROLLER_OTEL_AUDIENCE or null == clientId + && client != null + && lib.elem clientId client.audience + && client.accessTokenSignedResponseAlg == "RS256" && !(env shared.controllerAlone ? OTEL_EXPORTER_OTLP_ENDPOINT) && !(env shared.controllerAlone ? SWARM_CONTROLLER_OTEL_AUDIENCE); } diff --git a/swarm-controller/src/otel_http_client.rs b/swarm-controller/src/otel_http_client.rs index 004219be..c6955ef0 100644 --- a/swarm-controller/src/otel_http_client.rs +++ b/swarm-controller/src/otel_http_client.rs @@ -28,11 +28,23 @@ //! before it started, and that failure is intermittent — the worst kind to //! debug through a metrics pipeline that is itself the thing being fixed. +use std::fmt; +use std::sync::{Mutex, PoisonError}; +use std::time::{Duration, Instant}; + use async_trait::async_trait; use bytes::Bytes; use http::{Request, Response}; use opentelemetry_http::{HttpClient, HttpError}; -use swarm_queue_client::{QueueConfig, mint_token_for_blocking}; +use swarm_queue_client::{QueueConfig, chain, mint_token_for_blocking}; + +/// How long a failure of the same kind stays quiet after it was logged. +/// Every push in between is counted and the count rides on the next line. +const REPEAT_INTERVAL: Duration = Duration::from_mins(10); + +/// The most of a refusal's response body a log line carries. The collector +/// and authelia both answer with a one-line JSON error, well under this. +const BODY_SNIPPET_CHARS: usize = 200; /// See the module doc. Wraps a plain `reqwest::blocking::Client` — deliberately /// not `opentelemetry_http`'s own blanket impl for that type (behind its @@ -47,6 +59,7 @@ pub struct AuthenticatedHttpClient { /// `swarm_queue_client::mint_token_for`'s) for why this has to be asked /// for explicitly rather than left to whatever the identity's default is. audience: String, + failures: Mutex, } impl AuthenticatedHttpClient { @@ -65,6 +78,82 @@ impl AuthenticatedHttpClient { inner, cfg, audience: audience.into(), + failures: Mutex::default(), + } + } + + /// One push: mint, then post. A failure carries what to log beside the + /// error the SDK gets back. + fn send(&self, request: Request) -> Result, Failed> { + let token = mint_token_for_blocking(&self.cfg, Some(&self.audience), None) + .map_err(|e| Failed::new(Failure::Token, chain(&e), e))?; + self.post(&token, request) + } + + /// The HTTP half of [`Self::send`], with the token already minted. + fn post(&self, token: &str, mut request: Request) -> Result, Failed> { + let header = http::HeaderValue::from_str(&format!("Bearer {token}")) + .map_err(|e| Failed::new(Failure::Token, e.to_string(), e))?; + request + .headers_mut() + .insert(http::header::AUTHORIZATION, header); + + let req = request + .try_into() + .map_err(|e: reqwest::Error| Failed::new(Failure::Transport, chain(&e), e))?; + let mut response = self + .inner + .execute(req) + .map_err(|e| Failed::new(Failure::Transport, chain(&e), e))?; + let status = response.status(); + // Same header/body reassembly `opentelemetry_http`'s own blocking + // impl does — `reqwest::blocking::Response` and `http::Response` + // are different types, so this is the required conversion, not an + // extra step invented here. + let headers = std::mem::take(response.headers_mut()); + let body = response + .bytes() + .map_err(|e| Failed::new(Failure::Transport, chain(&e), e))?; + if !status.is_success() { + let detail = snippet(&body); + let error = format!("the collector answered {status}: {detail}"); + return Err(Failed::new(Failure::Status(status.as_u16()), detail, error)); + } + let mut http_response = Response::builder() + .status(status) + .body(body) + .map_err(|e| Failed::new(Failure::Transport, e.to_string(), e))?; + *http_response.headers_mut() = headers; + Ok(http_response) + } + + /// The SDK reports a failed export only at debug level, so this is the + /// one place a refused push shows up: a warning per kind of failure, + /// repeated every [`REPEAT_INTERVAL`] while it persists, and one info + /// line when a push succeeds after failing. + fn log(&self, outcome: Result<(), &Failed>) { + let mut failures = self.failures.lock().unwrap_or_else(PoisonError::into_inner); + match outcome { + Ok(()) => { + if failures.succeeded() == Verdict::Recovered { + tracing::info!( + client_id = %self.cfg.client_id, + "otel export: push to the collector succeeded after failing" + ); + } + } + Err(failed) => { + if let Verdict::Warn { suppressed } = failures.failed(failed.kind, Instant::now()) { + tracing::warn!( + failure = %failed.kind, + detail = %failed.detail, + suppressed, + client_id = %self.cfg.client_id, + audience = %self.audience, + "otel export: push to the collector failed" + ); + } + } } } } @@ -77,24 +166,277 @@ impl HttpClient for AuthenticatedHttpClient { /// need that, not because this one does, and the module doc explains /// why this specific caller deliberately runs on a thread with nothing /// to yield to. - async fn send_bytes(&self, mut request: Request) -> Result, HttpError> { - let token = mint_token_for_blocking(&self.cfg, Some(&self.audience), None)?; - request.headers_mut().insert( - http::header::AUTHORIZATION, - http::HeaderValue::from_str(&format!("Bearer {token}"))?, + async fn send_bytes(&self, request: Request) -> Result, HttpError> { + let outcome = self.send(request); + self.log(outcome.as_ref().map(|_| ())); + outcome.map_err(|failed| failed.error) + } +} + +/// What stopped one push. Coarse on purpose: a collector down for an hour +/// is one kind of failure, logged once, not sixty. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Failure { + /// No token: authelia refused the request or could not be reached. + Token, + /// No answer from the collector: DNS, connect, TLS, or a cut-off body. + Transport, + /// The collector, or the gateway in front of it, answered non-2xx. + Status(u16), +} + +impl fmt::Display for Failure { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Token => f.write_str("token"), + Self::Transport => f.write_str("transport"), + Self::Status(code) => write!(f, "status {code}"), + } + } +} + +struct Failed { + kind: Failure, + detail: String, + error: HttpError, +} + +impl Failed { + fn new(kind: Failure, detail: String, error: impl Into) -> Self { + Self { + kind, + detail, + error: error.into(), + } + } +} + +/// Whether a push outcome is worth a log line. +#[derive(Debug, PartialEq, Eq)] +enum Verdict { + Quiet, + /// Log this failure; `suppressed` pushes failed the same way unlogged + /// since the last line about it. + Warn { + suppressed: u64, + }, + /// The first success after a logged failure. + Recovered, +} + +/// The failure streak in progress, if any. +#[derive(Debug, Default)] +struct FailureLog { + streak: Option, +} + +#[derive(Debug)] +struct Streak { + kind: Failure, + logged_at: Instant, + suppressed: u64, +} + +impl FailureLog { + fn failed(&mut self, kind: Failure, now: Instant) -> Verdict { + match &mut self.streak { + Some(streak) if streak.kind == kind => { + if now.duration_since(streak.logged_at) < REPEAT_INTERVAL { + streak.suppressed += 1; + return Verdict::Quiet; + } + streak.logged_at = now; + Verdict::Warn { + suppressed: std::mem::take(&mut streak.suppressed), + } + } + _ => { + self.streak = Some(Streak { + kind, + logged_at: now, + suppressed: 0, + }); + Verdict::Warn { suppressed: 0 } + } + } + } + + fn succeeded(&mut self) -> Verdict { + if self.streak.take().is_some() { + Verdict::Recovered + } else { + Verdict::Quiet + } + } +} + +/// The start of a response body as one log-safe line. +fn snippet(body: &[u8]) -> String { + let text = String::from_utf8_lossy(body); + let text = text.trim(); + let mut out: String = text + .chars() + .take(BODY_SNIPPET_CHARS) + .map(|c| if c.is_control() { ' ' } else { c }) + .collect(); + if text.chars().count() > BODY_SNIPPET_CHARS { + out.push('…'); + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::PathBuf; + use swarm_queue_client::ClientSecret; + + fn client() -> AuthenticatedHttpClient { + let inner = reqwest::blocking::Client::builder() + .no_proxy() + .build() + .expect("client"); + let cfg = QueueConfig { + url: "nats://127.0.0.1:4222".to_owned(), + token_endpoint: "http://127.0.0.1:9/token".to_owned(), + client_id: "swarm-controller".to_owned(), + client_secret: ClientSecret::File(PathBuf::from("/nonexistent")), + ca_file: None, + }; + AuthenticatedHttpClient::new(inner, cfg, "swarm-controller") + } + + fn request(url: &str) -> Request { + Request::post(url) + .body(Bytes::from_static(b"{}")) + .expect("request") + } + + /// A stub collector answering every request with `status` and `body`. + async fn collector(status: http::StatusCode, body: &'static str) -> String { + let app = axum::Router::new().fallback(move || async move { (status, body) }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let url = format!("http://{}/v1/metrics", listener.local_addr().expect("addr")); + tokio::spawn(async move { axum::serve(listener, app).await }); + url + } + + async fn post(url: String) -> Result, (Failure, String)> { + tokio::task::spawn_blocking(move || { + client() + .post("a-token", request(&url)) + .map_err(|f| (f.kind, f.detail)) + }) + .await + .expect("joined") + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_refusal_is_classified_by_status_with_the_body() { + let url = collector( + http::StatusCode::UNAUTHORIZED, + r#"{"code":16,"message":"authentication didn't succeed"}"#, + ) + .await; + let (kind, detail) = post(url).await.expect_err("refused"); + assert_eq!(kind, Failure::Status(401)); + assert_eq!( + detail, + r#"{"code":16,"message":"authentication didn't succeed"}"# + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn an_accepted_push_is_returned_as_is() { + let url = collector(http::StatusCode::OK, "{}").await; + let response = post(url).await.expect("accepted"); + assert_eq!(response.status(), http::StatusCode::OK); + assert_eq!(response.body().as_ref(), b"{}"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn an_unreachable_collector_is_a_transport_failure() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let url = format!("http://{}/v1/metrics", listener.local_addr().expect("addr")); + drop(listener); + let (kind, _) = post(url).await.expect_err("unreachable"); + assert_eq!(kind, Failure::Transport); + } + + /// The mint runs first: no token, no request, and the failure says so. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_failed_mint_is_a_token_failure() { + let url = collector(http::StatusCode::OK, "{}").await; + let kind = tokio::task::spawn_blocking(move || client().send(request(&url)).err()) + .await + .expect("joined") + .expect("no token") + .kind; + assert_eq!(kind, Failure::Token); + } + + #[test] + fn a_repeated_failure_is_logged_once_per_interval_with_the_count() { + let mut log = FailureLog::default(); + let t0 = Instant::now(); + let kind = Failure::Status(401); + assert_eq!(log.failed(kind, t0), Verdict::Warn { suppressed: 0 }); + for minute in 1..10 { + assert_eq!( + log.failed(kind, t0 + Duration::from_mins(minute)), + Verdict::Quiet + ); + } + assert_eq!( + log.failed(kind, t0 + REPEAT_INTERVAL), + Verdict::Warn { suppressed: 9 } + ); + assert_eq!( + log.failed(kind, t0 + REPEAT_INTERVAL + Duration::from_mins(1)), + Verdict::Quiet + ); + } + + #[test] + fn a_different_failure_is_logged_at_once() { + let mut log = FailureLog::default(); + let t0 = Instant::now(); + assert_eq!( + log.failed(Failure::Token, t0), + Verdict::Warn { suppressed: 0 } + ); + assert_eq!( + log.failed(Failure::Status(401), t0), + Verdict::Warn { suppressed: 0 } + ); + assert_eq!( + log.failed(Failure::Status(404), t0), + Verdict::Warn { suppressed: 0 } + ); + } + + #[test] + fn recovery_is_logged_once_and_only_after_a_failure() { + let mut log = FailureLog::default(); + let t0 = Instant::now(); + assert_eq!(log.succeeded(), Verdict::Quiet); + log.failed(Failure::Transport, t0); + assert_eq!(log.succeeded(), Verdict::Recovered); + assert_eq!(log.succeeded(), Verdict::Quiet); + assert_eq!( + log.failed(Failure::Transport, t0), + Verdict::Warn { suppressed: 0 } ); + } - let req = request.try_into()?; - let mut response = self.inner.execute(req)?.error_for_status()?; - // Same header/body reassembly `opentelemetry_http`'s own blocking - // impl does — `reqwest::blocking::Response` and `http::Response` - // are different types, so this is the required conversion, not an - // extra step invented here. - let headers = std::mem::take(response.headers_mut()); - let mut http_response = Response::builder() - .status(response.status()) - .body(response.bytes()?)?; - *http_response.headers_mut() = headers; - Ok(http_response) + #[test] + fn a_long_body_is_cut_on_a_character_boundary() { + let body = "é".repeat(BODY_SNIPPET_CHARS + 5); + let cut = snippet(body.as_bytes()); + assert_eq!(cut.chars().count(), BODY_SNIPPET_CHARS + 1); + assert!(cut.ends_with('…')); + assert_eq!(snippet(b" short\nbody \n"), "short body"); } }