review: address relrod's feedback on linux agent ingestion

racicLuka · racicLuka · commit b052b7c2658e · 2026-08-09T17:57:53.000+02:00
- linux_agent_cache stores MetricsFetcherIngestion&lt;Bytes&gt; instead of a
  bare String/Arc&lt;String&gt;, matching kubelet_stats_summary_cache and
  giving self-health reporting a received_at timestamp to use later
- switch the ingest handler and cache to Bytes throughout, since
  check_mk_agent plugin output isn't guaranteed to be valid UTF-8; drop
  the false "charset=utf-8" claim on the content-type header sent by
  metrics-fetcher
- simplify NODE_NAME lookup and agent-timeout handling in
  metrics-fetcher's LinuxAgentScraper per suggested diffs
- fix metrics-fetcher main() silently exiting 0 if either scrape loop
  panics, by select!+log+bail on the join handles like metrics-cache's
  main() does
- move the linux_agent ingest router tests into handlers/mod.rs, since
  they exercise routing/middleware rather than the handler itself; add
  a manual, fail-closed Default for PullAgentMiddlewareConfig instead
  of an ad-hoc no_pull_agent() test helper
- drop the payload.rs unit tests that only asserted match arms
diff --git a/metrics-cache/src/auth/pull_agent.rs b/metrics-cache/src/auth/pull_agent.rs
@@ -31,6 +31,20 @@ pub struct PullAgentMiddlewareConfig {
     pub shared_secret: Option<String>,
 }
 
+impl Default for PullAgentMiddlewareConfig {
+    /// Defaults closed: auth enabled with no configured secret, so
+    /// [`authorized`] rejects every request. This is only meant as a
+    /// placeholder for tests that don't exercise pull-agent auth at all; if it
+    /// were ever reached in real code, it fails safe instead of leaving the
+    /// endpoint open.
+    fn default() -> Self {
+        Self {
+            auth_enabled: true,
+            shared_secret: None,
+        }
+    }
+}
+
 /// Perform authentication, specifically for pull-agent endpoints.
 ///
 /// If `config.shared_secret` is `None` or empty, we _REJECT_ all requests.
diff --git a/metrics-cache/src/handlers/ingest.rs b/metrics-cache/src/handlers/ingest.rs
@@ -1,3 +1,4 @@
+use axum::body::Bytes;
 use axum::extract::Path;
 use axum::{Json, extract::State};
 use std::sync::Arc;
@@ -25,99 +26,21 @@ pub async fn kubelet_stats_summary(
 }
 
 /// Store the raw check_mk_agent output for a node, verbatim, keyed by node
-/// name. No parsing/validation is done here or by the caller.
+/// name. No parsing/validation is done here or by the caller. Kept as
+/// [`Bytes`] rather than [`String`] since agent plugins are not guaranteed to
+/// produce valid UTF-8.
 pub async fn linux_agent(
     State(state): State<AppState<impl TokenValidator>>,
     Path(node_name): Path<String>,
-    body: String,
+    body: Bytes,
 ) -> Json<String> {
+    let ingestion = MetricsFetcherIngestion {
+        received_at: Instant::now(),
+        payload: body,
+    };
     state
         .linux_agent_cache
-        .insert(node_name, Arc::new(body))
+        .insert(node_name, Arc::new(ingestion))
         .await;
     Json("ok".to_string())
 }
-
-#[cfg(test)]
-mod tests {
-    use axum::body::Body;
-    use axum::http::{Request, StatusCode};
-    use k8s_openapi::api::authentication::v1::{TokenReview, TokenReviewStatus, UserInfo};
-    use tower::ServiceExt;
-
-    use crate::auth::pull_agent::PullAgentMiddlewareConfig;
-    use crate::handlers::app;
-    use crate::state::tests::{MockValidator, test_app_state_with_validator};
-
-    fn no_pull_agent() -> PullAgentMiddlewareConfig {
-        PullAgentMiddlewareConfig {
-            auth_enabled: false,
-            shared_secret: None,
-        }
-    }
-
-    fn authenticated_review(username: &str) -> TokenReview {
-        TokenReview {
-            status: Some(TokenReviewStatus {
-                authenticated: Some(true),
-                user: Some(UserInfo {
-                    username: Some(username.to_string()),
-                    ..Default::default()
-                }),
-                ..Default::default()
-            }),
-            ..Default::default()
-        }
-    }
-
-    #[tokio::test]
-    async fn linux_agent_ingest_populates_cache_and_returns_ok() {
-        let state = test_app_state_with_validator(MockValidator {
-            response: Ok(authenticated_review(
-                "system:serviceaccount:test-ns:test-writer",
-            )),
-        });
-        let cache = state.linux_agent_cache.clone();
-        let app = app(state, no_pull_agent());
-
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .method("POST")
-                    .uri("/ingest/linux_agent/node-1")
-                    .header("Authorization", "Bearer test-token")
-                    .body(Body::from("<<<check_mk>>>\nVersion: 2.5.0\n"))
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-
-        assert_eq!(resp.status(), StatusCode::OK);
-        cache.run_pending_tasks().await;
-        assert_eq!(
-            cache.get("node-1").await.map(|v| (*v).clone()),
-            Some("<<<check_mk>>>\nVersion: 2.5.0\n".to_string())
-        );
-    }
-
-    #[tokio::test]
-    async fn linux_agent_ingest_requires_auth() {
-        let state = test_app_state_with_validator(MockValidator {
-            response: Ok(TokenReview::default()),
-        });
-        let app = app(state, no_pull_agent());
-
-        let resp = app
-            .oneshot(
-                Request::builder()
-                    .method("POST")
-                    .uri("/ingest/linux_agent/node-1")
-                    .body(Body::from("<<<check_mk>>>\n"))
-                    .unwrap(),
-            )
-            .await
-            .unwrap();
-
-        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
-    }
-}
diff --git a/metrics-cache/src/handlers/mod.rs b/metrics-cache/src/handlers/mod.rs
@@ -47,3 +47,77 @@ pub fn ingest_app<V: TokenValidator>(state: AppState<V>) -> Router {
         .layer(tower_http::compression::CompressionLayer::new())
         .with_state(state)
 }
+
+#[cfg(test)]
+mod tests {
+    use axum::body::{Body, Bytes};
+    use axum::http::{Request, StatusCode};
+    use k8s_openapi::api::authentication::v1::{TokenReview, TokenReviewStatus, UserInfo};
+    use tower::ServiceExt;
+
+    use super::*;
+    use crate::state::tests::{MockValidator, test_app_state, test_app_state_with_validator};
+
+    fn authenticated_review(username: &str) -> TokenReview {
+        TokenReview {
+            status: Some(TokenReviewStatus {
+                authenticated: Some(true),
+                user: Some(UserInfo {
+                    username: Some(username.to_string()),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            }),
+            ..Default::default()
+        }
+    }
+
+    #[tokio::test]
+    async fn linux_agent_ingest_populates_cache_and_returns_ok() {
+        let state = test_app_state_with_validator(MockValidator {
+            response: Ok(authenticated_review(
+                "system:serviceaccount:test-ns:test-writer",
+            )),
+        });
+        let cache = state.linux_agent_cache.clone();
+        let app = ingest_app(state);
+
+        let resp = app
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/ingest/linux_agent/node-1")
+                    .header("Authorization", "Bearer test-token")
+                    .body(Body::from("<<<check_mk>>>\nVersion: 2.5.0\n"))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(resp.status(), StatusCode::OK);
+        cache.run_pending_tasks().await;
+        assert_eq!(
+            cache.get("node-1").await.map(|v| v.payload.clone()),
+            Some(Bytes::from_static(b"<<<check_mk>>>\nVersion: 2.5.0\n"))
+        );
+    }
+
+    #[tokio::test]
+    async fn linux_agent_ingest_requires_auth() {
+        let state = test_app_state();
+        let app = ingest_app(state);
+
+        let resp = app
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/ingest/linux_agent/node-1")
+                    .body(Body::from("<<<check_mk>>>\n"))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
+    }
+}
diff --git a/metrics-cache/src/state.rs b/metrics-cache/src/state.rs
@@ -1,3 +1,4 @@
+use axum::body::Bytes;
 use kube::Client;
 use moka::future::Cache;
 use std::sync::Arc;
@@ -22,7 +23,7 @@ pub struct AppState<V: TokenValidator> {
     pub reader_allowlist: Vec<String>,
     pub writer_allowlist: Vec<String>,
     pub kubelet_stats_summary_cache: Cache<String, Arc<MetricsFetcherIngestion<StatsSummary>>>,
-    pub linux_agent_cache: Cache<String, Arc<String>>,
+    pub linux_agent_cache: Cache<String, Arc<MetricsFetcherIngestion<Bytes>>>,
     pub host_settings: Arc<HostSettings>,
 }
 
diff --git a/metrics-fetcher/src/linux_agent.rs b/metrics-fetcher/src/linux_agent.rs
@@ -42,15 +42,10 @@ impl Scraper for LinuxAgentScraper {
     /// We do not parse the output or perform any calculations on it here,
     /// leaving these tasks for metrics-cache (and later, Checkmk itself) to do.
     async fn scrape(&self) -> Result<Payload> {
-        let node_name = match std::env::var("NODE_NAME") {
-            Ok(val) => val,
-            Err(err) => {
-                return Err(Error::EnvVar {
-                    name: "NODE_NAME".to_string(),
-                    source: err,
-                });
-            }
-        };
+        let node_name = std::env::var("NODE_NAME").map_err(|e| Error::EnvVar {
+            name: "NODE_NAME".to_string(),
+            source: e,
+        })?;
 
         debug!("running check_mk_agent");
         let child = Command::new(AGENT_PATH)
@@ -61,10 +56,9 @@ impl Scraper for LinuxAgentScraper {
             .kill_on_drop(true)
             .spawn()?;
 
-        let output = match timeout(AGENT_TIMEOUT, child.wait_with_output()).await {
-            Ok(result) => result?,
-            Err(_elapsed) => return Err(Error::AgentTimeout(AGENT_TIMEOUT)),
-        };
+        let output = timeout(AGENT_TIMEOUT, child.wait_with_output())
+            .await
+            .map_err(|_| Error::AgentTimeout(AGENT_TIMEOUT))??;
 
         if !output.status.success() {
             return Err(Error::AgentExitStatus {
diff --git a/metrics-fetcher/src/payload.rs b/metrics-fetcher/src/payload.rs
@@ -21,7 +21,9 @@ impl Payload {
     fn content_type(&self) -> &'static str {
         match self {
             Self::KubeletStatsSummary(_) => "application/json",
-            Self::CheckmkLinuxAgent { .. } => "text/plain; charset=utf-8",
+            // No charset: plugins can make check_mk_agent output non-UTF-8,
+            // so we don't assert one.
+            Self::CheckmkLinuxAgent { .. } => "text/plain",
         }
     }
 
@@ -73,32 +75,3 @@ impl Payload {
         Ok(response)
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn kubelet_stats_summary_endpoint_and_content_type() {
-        let payload = Payload::KubeletStatsSummary(Bytes::from_static(b"{}"));
-        assert_eq!(payload.metrics_cache_endpoint(), "/kubelet_stats_summary");
-        assert_eq!(payload.content_type(), "application/json");
-        assert_eq!(payload.extract(), Bytes::from_static(b"{}"));
-    }
-
-    #[test]
-    fn checkmk_linux_agent_endpoint_and_content_type() {
-        for node_name in ["node-1", "node-with-dashes.example.com"] {
-            let payload = Payload::CheckmkLinuxAgent {
-                node_name: node_name.to_string(),
-                body: Bytes::from_static(b"<<<check_mk>>>\n"),
-            };
-            assert_eq!(
-                payload.metrics_cache_endpoint(),
-                format!("/linux_agent/{node_name}")
-            );
-            assert_eq!(payload.content_type(), "text/plain; charset=utf-8");
-            assert_eq!(payload.extract(), Bytes::from_static(b"<<<check_mk>>>\n"));
-        }
-    }
-}
