KubeNodeKubeletV1 added

metrics-cache/src/handlers/debug.rs

@@ -12,6 +12,7 @@ pub async fn get(State(state): State<AppState<impl TokenValidator>>) -> Result<B
     let snap = Snapshot::new(
         state.stores,
         state.kubelet_stats_summary_cache,
+        state.kubelet_health_cache,
         state.system_agent_cache,
     );
     let sections = emit_all(&snap, &state.host_settings);

metrics-cache/src/handlers/ingest.rs

@@ -1,14 +1,14 @@
-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_health::KubeletHealth;
 use crate::ingest::kubelet_stats::StatsSummary;
+use axum::body::Bytes;
+use axum::extract::Path;
+use axum::{Json, extract::State};
+use std::sync::Arc;
+use std::time::Instant;
 
 pub async fn kubelet_stats_summary(
     State(state): State<AppState<impl TokenValidator>>,
@@ -26,6 +26,22 @@ pub async fn kubelet_stats_summary(
     Json("ok".to_string())
 }
 
+pub async fn kubelet_health(
+    State(state): State<AppState<impl TokenValidator>>,
+    Path(node_name): Path<String>,
+    Json(health): Json<KubeletHealth>,
+) -> Json<String> {
+    let ingestion = MetricsFetcherIngestion {
+        received_at: Instant::now(),
+        payload: health,
+    };
+    state
+        .kubelet_health_cache
+        .insert(node_name, Arc::new(ingestion))
+        .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`]

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("/kubelet_health/{node_name}", post(ingest::kubelet_health))
         .route("/system_agent/{node_name}", post(ingest::system_agent))
         .route_layer(middleware::from_fn_with_state(
             state.clone(),

metrics-cache/src/ingest/kubelet_health.rs

@@ -0,0 +1,8 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum KubeletHealth {
+    Response { status_code: u16, response: String },
+    ConnectionError { message: String },
+}

metrics-cache/src/ingest/mod.rs

@@ -1,6 +1,7 @@
 use axum::body::Bytes;
 use std::time::Instant;
 
+pub mod kubelet_health;
 pub mod kubelet_stats;
 pub mod reflectors;
 

metrics-cache/src/piggyback/node.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use crate::host_settings::HostSettings;
 use crate::piggyback::{AggregationHost, Meta, PiggybackHost};
 use crate::section::node::KubeNodeInfoV1;
+use crate::section::node_kubelet::KubeNodeKubeletV1;
 use crate::section::writeable::{SectionError, WriteableSection};
 use crate::snapshot::Snapshot;
 
@@ -53,9 +54,16 @@ impl PiggybackHost for Node<'_> {
     fn emit(&self) -> Vec<Result<WriteableSection, SectionError>> {
         let me = self.meta.piggyback_hostname(&self.settings.cluster_name);
         let mut out = Vec::new();
+
         if let Some(kube_node_info_v1) = KubeNodeInfoV1::from_node(self.api, self.settings) {
             out.push(WriteableSection::of(&me, &kube_node_info_v1));
         }
+
+        if let Some(kube_node_kubelet_v1) =
+            KubeNodeKubeletV1::from_node(self.api, &self.snapshot.kubelet_health)
+        {
+            out.push(WriteableSection::of(&me, &kube_node_kubelet_v1));
+        }
         if let Some(ingestion) = self.snapshot.system_agent_snapshot.get(self.meta.name) {
             out.push(Ok(WriteableSection::raw(&me, ingestion.payload.0.clone())));
         }
@@ -95,6 +103,7 @@ mod tests {
         let snapshot = Snapshot::new(
             state.stores,
             state.kubelet_stats_summary_cache,
+            state.kubelet_health_cache,
             state.system_agent_cache,
         );
         let node = Node::new(&api, &snapshot, &host_settings).unwrap();

metrics-cache/src/push/mod.rs

@@ -48,6 +48,7 @@ async fn push_cycle(
     let snap = Snapshot::new(
         state.stores.clone(),
         state.kubelet_stats_summary_cache.clone(),
+        state.kubelet_health_cache.clone(),
         state.system_agent_cache.clone(),
     );
     let sections = emit_all(&snap, &state.host_settings);

metrics-cache/src/section/mod.rs

@@ -2,6 +2,7 @@ pub mod common;
 pub mod cronjob;
 pub mod namespace;
 pub mod node;
+pub mod node_kubelet;
 pub mod performance;
 pub mod pod;
 pub mod pvc;

metrics-cache/src/section/node_kubelet.rs

@@ -0,0 +1,91 @@
+use k8s_openapi::api::core::v1::Node;
+use serde::Serialize;
+
+use crate::ingest::kubelet_health::KubeletHealth;
+use crate::section::Section;
+use crate::snapshot::kubelet_health::KubeletHealths;
+
+/// Kubelet version and health. (`kube_node_kubelet_v1`)
+#[derive(Serialize)]
+pub(crate) struct KubeNodeKubeletV1<'a> {
+    pub version: &'a str,
+    pub health: &'a KubeletHealth,
+}
+
+impl<'a> KubeNodeKubeletV1<'a> {
+    pub fn from_node(
+        node: &'a Node,
+        kubelet_healths: &'a KubeletHealths,
+    ) -> Option<KubeNodeKubeletV1<'a>> {
+        let node_info = node.status.as_ref()?.node_info.as_ref()?;
+        let node_name = node.metadata.name.as_deref()?;
+        let health = kubelet_healths.get(node_name)?;
+
+        Some(KubeNodeKubeletV1 {
+            version: &node_info.kubelet_version,
+            health,
+        })
+    }
+}
+
+impl Section for KubeNodeKubeletV1<'_> {
+    const NAME: &'static str = "kube_node_kubelet_v1";
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+    use std::time::Instant;
+
+    use super::*;
+
+    use crate::ingest::MetricsFetcherIngestion;
+    use crate::test_support::*;
+
+    fn kubelet_healths(entries: &[(&str, KubeletHealth)]) -> KubeletHealths {
+        KubeletHealths {
+            by_node: entries
+                .iter()
+                .map(|(name, health)| {
+                    let ingestion = MetricsFetcherIngestion {
+                        received_at: Instant::now(),
+                        payload: health.clone(),
+                    };
+                    (name.to_string(), Arc::new(ingestion))
+                })
+                .collect(),
+        }
+    }
+
+    #[test]
+    fn kube_node_kubelet_v1_response() {
+        let node = node_prefilled("node01");
+        let healths = kubelet_healths(&[(
+            "node01",
+            KubeletHealth::Response {
+                status_code: 200,
+                response: "ok".to_string(),
+            },
+        )]);
+        insta::assert_json_snapshot!(KubeNodeKubeletV1::from_node(&node, &healths));
+    }
+
+    #[test]
+    fn kube_node_kubelet_v1_connection_error() {
+        let node = node_prefilled("node01");
+        let healths = kubelet_healths(&[(
+            "node01",
+            KubeletHealth::ConnectionError {
+                message: "connection refused".to_string(),
+            },
+        )]);
+        insta::assert_json_snapshot!(KubeNodeKubeletV1::from_node(&node, &healths));
+    }
+
+    #[test]
+    fn kube_node_kubelet_v1_no_health_reported() {
+        let node = node_prefilled("node01");
+        let healths = kubelet_healths(&[]);
+        assert!(KubeNodeKubeletV1::from_node(&node, &healths).is_none());
+    }
+}

metrics-cache/src/section/snapshots/metrics_cache__section__node_kubelet__tests__kube_node_kubelet_v1_connection_error.snap

@@ -0,0 +1,10 @@
+---
+source: metrics-cache/src/section/node_kubelet.rs
+expression: "KubeNodeKubeletV1::from_node(&node, &healths)"
+---
+{
+  "version": "v1.34.0",
+  "health": {
+    "message": "connection refused"
+  }
+}

metrics-cache/src/section/snapshots/metrics_cache__section__node_kubelet__tests__kube_node_kubelet_v1_response.snap

@@ -0,0 +1,11 @@
+---
+source: metrics-cache/src/section/node_kubelet.rs
+expression: "KubeNodeKubeletV1::from_node(&node, &healths)"
+---
+{
+  "version": "v1.34.0",
+  "health": {
+    "status_code": 200,
+    "response": "ok"
+  }
+}

metrics-cache/src/snapshot/kubelet_health.rs

@@ -0,0 +1,30 @@
+use moka::future::Cache;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use crate::ingest::MetricsFetcherIngestion;
+use crate::ingest::kubelet_health::KubeletHealth;
+
+/// Kubelet `/healthz` results pushed by metrics-fetcher, indexed by node name.
+#[derive(Debug)]
+pub struct KubeletHealths {
+    pub by_node: HashMap<String, Arc<MetricsFetcherIngestion<KubeletHealth>>>,
+}
+
+impl KubeletHealths {
+    pub fn from_cache(cache: &Cache<String, Arc<MetricsFetcherIngestion<KubeletHealth>>>) -> Self {
+        Self {
+            by_node: cache
+                .iter()
+                .map(|(name, ingestion)| (name.to_string(), ingestion))
+                .collect(),
+        }
+    }
+
+    /// Get the Kubelet `/healthz` result last reported for a node, if any.
+    pub fn get(&self, node_name: &str) -> Option<&KubeletHealth> {
+        self.by_node
+            .get(node_name)
+            .map(|ingestion| &ingestion.payload)
+    }
+}

metrics-cache/src/snapshot/mod.rs

@@ -1,4 +1,5 @@
 pub mod indexes;
+pub mod kubelet_health;
 pub mod metric_tables;
 pub mod owner_graph;
 pub mod self_health;
@@ -11,9 +12,11 @@ use std::time::Instant;
 
 use crate::ingest::MetricsFetcherIngestion;
 use crate::ingest::SystemAgentOutput;
+use crate::ingest::kubelet_health::KubeletHealth;
 use crate::ingest::kubelet_stats::StatsSummary;
 use crate::ingest::reflectors::{FrozenStores, Stores};
 use crate::snapshot::indexes::Indexes;
+use crate::snapshot::kubelet_health::KubeletHealths;
 use crate::snapshot::metric_tables::MetricTables;
 use crate::snapshot::owner_graph::OwnerGraph;
 use crate::snapshot::self_health::SelfHealth;
@@ -40,6 +43,7 @@ pub struct Snapshot {
     pub metrics: MetricTables,
     pub indexes: Indexes,
     pub self_health: SelfHealth,
+    pub kubelet_health: KubeletHealths,
     pub system_agent_snapshot: HashMap<String, Arc<MetricsFetcherIngestion<SystemAgentOutput>>>,
 }
 
@@ -49,6 +53,7 @@ impl Snapshot {
     pub fn new(
         stores: Stores,
         kubelet_stats_summary_cache: Cache<String, Arc<MetricsFetcherIngestion<StatsSummary>>>,
+        kubelet_health_cache: Cache<String, Arc<MetricsFetcherIngestion<KubeletHealth>>>,
         system_agent_cache: Cache<String, Arc<MetricsFetcherIngestion<SystemAgentOutput>>>,
     ) -> Self {
         let instant = Instant::now();
@@ -63,6 +68,7 @@ impl Snapshot {
             reflector_healths,
             &kubelet_stats_summary_cache,
         );
+        let kubelet_health = KubeletHealths::from_cache(&kubelet_health_cache);
         let system_agent_snapshot: HashMap<
             String,
             Arc<MetricsFetcherIngestion<SystemAgentOutput>>,
@@ -77,6 +83,7 @@ impl Snapshot {
             metrics,
             indexes,
             self_health,
+            kubelet_health,
             system_agent_snapshot,
         }
     }

metrics-cache/src/state.rs

@@ -10,6 +10,7 @@ use crate::error::Result;
 use crate::host_settings::{AlwaysEmitted, AnnotationKeyPattern, HostSettings};
 use crate::ingest::MetricsFetcherIngestion;
 use crate::ingest::SystemAgentOutput;
+use crate::ingest::kubelet_health::KubeletHealth;
 use crate::ingest::kubelet_stats::StatsSummary;
 use crate::ingest::reflectors::Stores;
 
@@ -23,6 +24,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 kubelet_health_cache: Cache<String, Arc<MetricsFetcherIngestion<KubeletHealth>>>,
     pub system_agent_cache: Cache<String, Arc<MetricsFetcherIngestion<SystemAgentOutput>>>,
     pub host_settings: Arc<HostSettings>,
 }
@@ -50,6 +52,9 @@ impl AppState<Client> {
                 .time_to_live(args.kubelet_stats_cache_ttl)
                 .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
                 .build(),
+            kubelet_health_cache: Cache::builder()
+                .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
+                .build(),
             system_agent_cache: Cache::builder()
                 .time_to_live(args.system_agent_cache_ttl)
                 .max_capacity(MAX_SUPPORTED_KUBERNETES_NODES)
@@ -108,6 +113,10 @@ pub mod tests {
                 .time_to_live(Duration::from_secs(120))
                 .max_capacity(10000)
                 .build(),
+            kubelet_health_cache: Cache::builder()
+                .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)

metrics-fetcher/Cargo.toml

@@ -13,6 +13,8 @@ bytes = "1.11.1"
 clap = { workspace = true }
 reqwest = { version = "0.13.3", features = ["blocking", "query", "json", "rustls-no-provider", "http2", "system-proxy"], default-features = false }
 rustls = { workspace = true }
+serde = { workspace = true }
+serde_json = "1.0.150"
 thiserror = { workspace = true }
 tokio = { workspace = true, features = ["process"] }
 tracing = { workspace = true }