feat: add linux agent data into metrics-cache

CMK-36613

Author: racicLuka

metrics-cache/src/handlers/ingest.rs

@@ -1,3 +1,4 @@
+use axum::extract::Path;
 use axum::{Json, extract::State};
 use std::sync::Arc;
 use std::time::Instant;
@@ -22,3 +23,101 @@ pub async fn kubelet_stats_summary(
         .await;
     Json("ok".to_string())
 }
+
+/// 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.
+pub async fn linux_agent(
+    State(state): State<AppState<impl TokenValidator>>,
+    Path(node_name): Path<String>,
+    body: String,
+) -> Json<String> {
+    state
+        .linux_agent_cache
+        .insert(node_name, Arc::new(body))
+        .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);
+    }
+}

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("/linux_agent/{node_name}", post(ingest::linux_agent))
         .route_layer(middleware::from_fn_with_state(
             state.clone(),
             auth::kubernetes::authenticate,

metrics-cache/src/state.rs

@@ -22,6 +22,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 host_settings: Arc<HostSettings>,
 }
 
@@ -47,6 +48,9 @@ impl AppState<Client> {
             kubelet_stats_summary_cache: Cache::builder()
                 .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
                 .build(),
+            linux_agent_cache: Cache::builder()
+                .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
+                .build(),
             host_settings: host_settings.into(),
         };
         Ok(state)
@@ -101,6 +105,10 @@ pub mod tests {
                 .time_to_live(Duration::from_secs(120))
                 .max_capacity(10000)
                 .build(),
+            linux_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,86 @@
+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 = match std::env::var("NODE_NAME") {
+            Ok(val) => val,
+            Err(err) => {
+                return Err(Error::EnvVar {
+                    name: "NODE_NAME".to_string(),
+                    source: err,
+                });
+            }
+        };
+
+        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 = match timeout(AGENT_TIMEOUT, child.wait_with_output()).await {
+            Ok(result) => result?,
+            Err(_elapsed) => return 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,28 @@ 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!("/linux_agent/{node_name}"),
+        }
+    }
+
+    fn content_type(&self) -> &'static str {
+        match self {
+            Self::KubeletStatsSummary(_) => "application/json",
+            Self::CheckmkLinuxAgent { .. } => "text/plain; charset=utf-8",
         }
     }
 
     fn extract(&self) -> Bytes {
         match self {
             Self::KubeletStatsSummary(bytes) => bytes.clone(),
+            Self::CheckmkLinuxAgent { body, .. } => body.clone(),
         }
     }
 
@@ -54,7 +62,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() {
@@ -65,3 +73,32 @@ 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"));
+        }
+    }
+}