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

@@ -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 <code>`)
+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

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";
       }
     ];
   };

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 == clientId
-          (env allLocal).SWARM_CONTROLLER_OTEL_AUDIENCE or null
+        && client != null
-          == allLocal.services.hyperhive.swarm.controller.queueClientId
+        && lib.elem clientId client.audience
+        && client.accessTokenSignedResponseAlg == "RS256"
         && !(env shared.controllerAlone ? OTEL_EXPORTER_OTLP_ENDPOINT)
         && !(env shared.controllerAlone ? SWARM_CONTROLLER_OTEL_AUDIENCE);
     }

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<FailureLog>,
 }
 
 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<Bytes>) -> Result<Response<Bytes>, 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<Bytes>) -> Result<Response<Bytes>, 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<Bytes>) -> Result<Response<Bytes>, HttpError> {
+    async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
-        let token = mint_token_for_blocking(&self.cfg, Some(&self.audience), None)?;
+        let outcome = self.send(request);
-        request.headers_mut().insert(
+        self.log(outcome.as_ref().map(|_| ()));
-            http::header::AUTHORIZATION,
+        outcome.map_err(|failed| failed.error)
-            http::HeaderValue::from_str(&format!("Bearer {token}"))?,
+    }
-        );
+}
-
+
-        let req = request.try_into()?;
+/// What stopped one push. Coarse on purpose: a collector down for an hour
-        let mut response = self.inner.execute(req)?.error_for_status()?;
+/// is one kind of failure, logged once, not sixty.
-        // Same header/body reassembly `opentelemetry_http`'s own blocking
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-        // impl does — `reqwest::blocking::Response` and `http::Response`
+enum Failure {
-        // are different types, so this is the required conversion, not an
+    /// No token: authelia refused the request or could not be reached.
-        // extra step invented here.
+    Token,
-        let headers = std::mem::take(response.headers_mut());
+    /// No answer from the collector: DNS, connect, TLS, or a cut-off body.
-        let mut http_response = Response::builder()
+    Transport,
-            .status(response.status())
+    /// The collector, or the gateway in front of it, answered non-2xx.
-            .body(response.bytes()?)?;
+    Status(u16),
-        *http_response.headers_mut() = headers;
+}
-        Ok(http_response)
+
+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<HttpError>) -> 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<Streak>,
+}
+
+#[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<Bytes> {
+        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<Response<Bytes>, (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 }
+        );
+    }
+
+    #[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");
     }
 }