feat: add linux agent data into metrics-cache

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.

metrics-cache/src/handlers/ingest.rs

@@ -1,10 +1,13 @@
+use axum::body::Bytes;
+use axum::extract::Path;
 use axum::{Json, extract::State};
 use std::sync::Arc;
 use std::time::Instant;
 
 use crate::AppState;
 use crate::auth::kubernetes::TokenValidator;
 use crate::ingest::MetricsFetcherIngestion;
+use crate::ingest::SystemAgentOutput;
 use crate::ingest::kubelet_stats::StatsSummary;
 
 pub async fn kubelet_stats_summary(
@@ -22,3 +25,55 @@ pub async fn kubelet_stats_summary(
         .await;
     Json("ok".to_string())
 }
+
+/// Store the raw output of a machine-level agent (currently only Linux's
+/// `check_mk_agent`) for a node as-is, keyed by node 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 system_agent(
+    State(state): State<AppState<impl TokenValidator>>,
+    Path(node_name): Path<String>,
+    body: Bytes,
+) -> Json<String> {
+    let ingestion = MetricsFetcherIngestion {
+        received_at: Instant::now(),
+        payload: SystemAgentOutput(body),
+    };
+    state
+        .system_agent_cache
+        .insert(node_name, Arc::new(ingestion))
+        .await;
+    Json("ok".to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use axum::extract::{Path, State};
+
+    use super::*;
+    use crate::state::tests::test_app_state;
+
+    /// Exercises the handler directly (no router, no auth middleware) — this
+    /// is about whether the handler does what's expected of it, not whether
+    /// the route is wired up correctly; that's covered in `handlers::tests`.
+    #[tokio::test]
+    async fn system_agent_populates_cache_and_returns_ok() {
+        let state = test_app_state();
+        let cache = state.system_agent_cache.clone();
+
+        let Json(resp) = system_agent(
+            State(state),
+            Path("node-1".to_string()),
+            Bytes::from_static(b"<<<check_mk>>>\nVersion: 2.5.0\n"),
+        )
+        .await;
+
+        assert_eq!(resp, "ok");
+        cache.run_pending_tasks().await;
+        assert_eq!(
+            cache.get("node-1").await.map(|v| v.payload.0.clone()),
+            Some(Bytes::from_static(b"<<<check_mk>>>\nVersion: 2.5.0\n"))
+        );
+    }
+}

metrics-cache/src/handlers/mod.rs

@@ -36,6 +36,7 @@ pub fn ingest_app<V: TokenValidator>(state: AppState<V>) -> Router {
             "/kubelet_stats_summary",
             post(ingest::kubelet_stats_summary),
         )
+        .route("/system_agent/{node_name}", post(ingest::system_agent))
         .route_layer(middleware::from_fn_with_state(
             state.clone(),
             auth::kubernetes::authenticate,
@@ -46,3 +47,35 @@ 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;
+    use axum::http::{Request, StatusCode};
+    use tower::ServiceExt;
+
+    use super::*;
+    use crate::state::tests::test_app_state;
+
+    /// Sanity check that the ingest route is actually wired up behind the
+    /// kubernetes-token middleware; handler-level behavior is covered in
+    /// `handlers::ingest::tests`.
+    #[tokio::test]
+    async fn system_agent_ingest_requires_auth() {
+        let state = test_app_state();
+        let app = ingest_app(state);
+
+        let resp = app
+            .oneshot(
+                Request::builder()
+                    .method("POST")
+                    .uri("/ingest/system_agent/node-1")
+                    .body(Body::from("<<<check_mk>>>\n"))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
+    }
+}

metrics-cache/src/ingest/mod.rs

@@ -1,3 +1,4 @@
+use axum::body::Bytes;
 use std::time::Instant;
 
 pub mod kubelet_stats;
@@ -13,3 +14,11 @@ pub struct MetricsFetcherIngestion<T> {
     pub received_at: Instant,
     pub payload: T,
 }
+
+/// Raw output from a machine-level agent (currently only Linux's
+/// `check_mk_agent`, but named generically since a Windows agent could push
+/// here too some day). A distinct type rather than a bare [`Bytes`], so the
+/// cache's shape stays unambiguous if another `Bytes`-based payload is ever
+/// added.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SystemAgentOutput(pub Bytes);

metrics-cache/src/state.rs

@@ -9,6 +9,7 @@ use crate::cli_args::CliArgs;
 use crate::error::Result;
 use crate::host_settings::{AlwaysEmitted, AnnotationKeyPattern, HostSettings};
 use crate::ingest::MetricsFetcherIngestion;
+use crate::ingest::SystemAgentOutput;
 use crate::ingest::kubelet_stats::StatsSummary;
 use crate::ingest::reflectors::Stores;
 
@@ -22,6 +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 system_agent_cache: Cache<String, Arc<MetricsFetcherIngestion<SystemAgentOutput>>>,
     pub host_settings: Arc<HostSettings>,
 }
 
@@ -47,6 +49,9 @@ impl AppState<Client> {
             kubelet_stats_summary_cache: Cache::builder()
                 .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
                 .build(),
+            system_agent_cache: Cache::builder()
+                .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
+                .build(),
             host_settings: host_settings.into(),
         };
         Ok(state)
@@ -101,6 +106,10 @@ pub mod tests {
                 .time_to_live(Duration::from_secs(120))
                 .max_capacity(10000)
                 .build(),
+            system_agent_cache: Cache::builder()
+                .time_to_live(Duration::from_secs(120))
+                .max_capacity(10000)
+                .build(),
             host_settings: HostSettings {
                 cluster_name: "testcluster".to_string(),
                 cluster_host_name: "testclusterhost".to_string(),

metrics-fetcher/Cargo.toml

@@ -14,7 +14,7 @@ clap = { workspace = true }
 reqwest = { version = "0.13.3", features = ["blocking", "query", "json", "rustls-no-provider", "http2", "system-proxy"], default-features = false }
 rustls = { workspace = true }
 thiserror = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["process"] }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 

metrics-fetcher/src/error.rs

@@ -14,4 +14,13 @@ pub(crate) enum Error {
         #[source]
         source: std::env::VarError,
     },
+    #[error("check_mk_agent timed out after {0:?}")]
+    AgentTimeout(std::time::Duration),
+    #[error("check_mk_agent exited with {status}: {stderr}")]
+    AgentExitStatus {
+        status: std::process::ExitStatus,
+        stderr: String,
+    },
+    #[error("check_mk_agent produced empty output")]
+    AgentEmptyOutput,
 }

metrics-fetcher/src/linux_agent.rs

@@ -0,0 +1,80 @@
+use bytes::Bytes;
+use reqwest::Client;
+use std::process::Stdio;
+use std::sync::Arc;
+use tokio::process::Command;
+use tokio::time::{Duration, timeout};
+use tracing::{debug, trace};
+
+use crate::cli_args::CliArgs;
+use crate::error::{Error, Result};
+use crate::payload::Payload;
+use crate::scraper::Scraper;
+
+const AGENT_PATH: &str = "/usr/local/bin/check_mk_agent";
+const AGENT_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub(crate) struct LinuxAgentScraper {
+    relay_client: Client,
+    args: Arc<CliArgs>,
+}
+
+impl LinuxAgentScraper {
+    pub(crate) fn new(args: Arc<CliArgs>, metrics_cache_client: Client) -> LinuxAgentScraper {
+        LinuxAgentScraper {
+            relay_client: metrics_cache_client,
+            args,
+        }
+    }
+}
+
+impl Scraper for LinuxAgentScraper {
+    fn relay_client(&self) -> Client {
+        self.relay_client.clone()
+    }
+
+    fn args(&self) -> Arc<CliArgs> {
+        self.args.clone()
+    }
+
+    /// Run the local check_mk_agent script and capture its raw stdout.
+    ///
+    /// 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 = 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)
+            .env("PYTHONDONTWRITEBYTECODE", "1")
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
+            .kill_on_drop(true)
+            .spawn()?;
+
+        let output = timeout(AGENT_TIMEOUT, child.wait_with_output())
+            .await
+            .map_err(|_| Error::AgentTimeout(AGENT_TIMEOUT))??;
+
+        if !output.status.success() {
+            return Err(Error::AgentExitStatus {
+                status: output.status,
+                stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
+            });
+        }
+
+        if output.stdout.is_empty() {
+            return Err(Error::AgentEmptyOutput);
+        }
+
+        trace!(bytes = output.stdout.len(), "check_mk_agent run complete");
+        Ok(Payload::CheckmkLinuxAgent {
+            node_name,
+            body: Bytes::from(output.stdout),
+        })
+    }
+}

metrics-fetcher/src/main.rs

@@ -1,6 +1,7 @@
 mod cli_args;
 mod error;
 mod kubelet_stats_summary;
+mod linux_agent;
 mod payload;
 mod scraper;
 
@@ -13,6 +14,7 @@ use tracing_subscriber::EnvFilter;
 
 use crate::cli_args::CliArgs;
 use crate::kubelet_stats_summary::KubeletStatsSummaryScraper;
+use crate::linux_agent::LinuxAgentScraper;
 use crate::scraper::Scraper;
 
 #[tokio::main]
@@ -30,19 +32,33 @@ async fn main() -> Result<()> {
         .install_default()
         .expect("Failed to install rustls crypto provider");
 
-    // Client to communicate with metrics cache; we allocate it just once, up front.
+    // Client to communicate with metrics cache; we allocate it just once, up front,
+    // and share it between scrapers.
     let metrics_cache_client = match args.metrics_cache_ca_cert_file.as_deref() {
         Some(file) => {
             let pem = tokio::fs::read(file).await?;
             let ca = reqwest::Certificate::from_pem(&pem)?;
             ClientBuilder::new().tls_certs_only([ca])
         }
         None => ClientBuilder::new(),
-    };
+    }
+    .build()?;
 
+    let args = Arc::new(args);
     let kubelet_stats_summary_scraper =
-        KubeletStatsSummaryScraper::new(Arc::new(args), metrics_cache_client.build()?);
+        KubeletStatsSummaryScraper::new(args.clone(), metrics_cache_client.clone());
+    let linux_agent_scraper = LinuxAgentScraper::new(args.clone(), metrics_cache_client);
     let kubelet_scrape = tokio::spawn(kubelet_stats_summary_scraper.loop_push_scrape());
-    let _ = tokio::try_join!(kubelet_scrape);
-    Ok(())
+    let linux_agent_scrape = tokio::spawn(linux_agent_scraper.loop_push_scrape());
+
+    tokio::select! {
+        res = kubelet_scrape => {
+            tracing::error!(error = ?res, "kubelet stats summary scrape loop exited unexpectedly");
+            anyhow::bail!("kubelet stats summary scrape loop terminated unexpectedly");
+        }
+        res = linux_agent_scrape => {
+            tracing::error!(error = ?res, "linux agent scrape loop exited unexpectedly");
+            anyhow::bail!("linux agent scrape loop terminated unexpectedly");
+        }
+    }
 }

metrics-fetcher/src/payload.rs

@@ -7,20 +7,30 @@ use crate::error::Result;
 #[derive(Debug)]
 pub(crate) enum Payload {
     KubeletStatsSummary(Bytes),
-    // TODO:
-    // CheckmkLinuxAgent(Bytes),
+    CheckmkLinuxAgent { node_name: String, body: Bytes },
 }
 
 impl Payload {
-    fn metrics_cache_endpoint(&self) -> &str {
+    fn metrics_cache_endpoint(&self) -> String {
         match self {
-            Self::KubeletStatsSummary(_) => "/kubelet_stats_summary",
+            Self::KubeletStatsSummary(_) => "/kubelet_stats_summary".to_string(),
+            Self::CheckmkLinuxAgent { node_name, .. } => format!("/system_agent/{node_name}"),
+        }
+    }
+
+    fn content_type(&self) -> &'static str {
+        match self {
+            Self::KubeletStatsSummary(_) => "application/json",
+            // Not text/plain: a patched image's plugin can make check_mk_agent
+            // output non-UTF-8, even non-textual, so we don't claim otherwise.
+            Self::CheckmkLinuxAgent { .. } => "application/octet-stream",
         }
     }
 
     fn extract(&self) -> Bytes {
         match self {
             Self::KubeletStatsSummary(bytes) => bytes.clone(),
+            Self::CheckmkLinuxAgent { body, .. } => body.clone(),
         }
     }
 
@@ -54,7 +64,7 @@ impl Payload {
             .post(&url)
             .bearer_auth(token.trim())
             .body(self.extract())
-            .header("content-type", "application/json")
+            .header("content-type", self.content_type())
             .send()
             .await?;
         if response.status().is_success() {