feat: cmk Linux agent data emitted from metrics-cache

metrics-cache/src/handlers/debug.rs

@@ -8,7 +8,11 @@ use crate::section::writeable::frame;
 use crate::snapshot::Snapshot;
 
 pub async fn get(State(state): State<AppState<impl TokenValidator>>) -> Result<String, StatusCode> {
-    let snap = Snapshot::new(state.stores, state.kubelet_stats_summary_cache);
+    let snap = Snapshot::new(
+        state.stores,
+        state.kubelet_stats_summary_cache,
+        state.system_agent_cache,
+    );
     let sections = emit_all(&snap, &state.host_settings);
     let mut out = Vec::new();
     frame(&mut out, sections).map_err(|e| {

metrics-cache/src/piggyback/node.rs

@@ -56,7 +56,60 @@ impl PiggybackHost for Node<'_> {
         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(raw) = self.snapshot.system_agent_snapshot.get(self.meta.name) {
+            out.push(Ok(WriteableSection::from_raw(me.clone(), raw.clone())));
+        }
         out.extend(self.aggregation_sections(&me));
         out
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use axum::body::Bytes;
+    use std::time::Instant;
+
+    use crate::ingest::{MetricsFetcherIngestion, SystemAgentOutput};
+    use crate::section::writeable::SectionBody;
+    use crate::state::tests::test_app_state;
+    use crate::test_support;
+
+    #[tokio::test]
+    async fn emit_includes_raw_system_agent_output_keyed_by_bare_node_name() {
+        let state = test_app_state();
+        let cache = state.system_agent_cache.clone();
+        cache
+            .insert(
+                "node-1".to_string(),
+                Arc::new(MetricsFetcherIngestion {
+                    received_at: Instant::now(),
+                    payload: SystemAgentOutput(Bytes::from_static(b"<<<check_mk>>>\n")),
+                }),
+            )
+            .await;
+        cache.run_pending_tasks().await;
+
+        let api = test_support::node("node-1");
+        let host_settings = state.host_settings.clone();
+        let snapshot = Snapshot::new(
+            state.stores,
+            state.kubelet_stats_summary_cache,
+            state.system_agent_cache,
+        );
+        let node = Node::new(&api, &snapshot, &host_settings).unwrap();
+
+        let raw_section = node
+            .emit()
+            .into_iter()
+            .filter_map(Result::ok)
+            .find(|s| matches!(s.body, SectionBody::Raw(_)))
+            .expect("expected a raw system agent section");
+
+        assert_eq!(raw_section.piggyback_hostname, "node_testcluster_node-1");
+        match raw_section.body {
+            SectionBody::Raw(raw) => assert_eq!(raw, Bytes::from_static(b"<<<check_mk>>>\n")),
+            SectionBody::Json { .. } => unreachable!(),
+        }
+    }
+}

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.system_agent_cache.clone(),
     );
     let sections = emit_all(&snap, &state.host_settings);
     let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default());

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

@@ -0,0 +1,13 @@
+---
+source: metrics-cache/src/section/writeable.rs
+assertion_line: 120
+expression: "String::from_utf8(out).unwrap()"
+---
+<<<<json-host>>>>
+<<<test_section_v1:sep(0)>>>
+{"value":7}
+<<<<>>>>
+<<<<raw-host>>>>
+<<<check_mk>>>
+Version: 2.5.0
+<<<<>>>>

metrics-cache/src/section/writeable.rs

@@ -1,3 +1,4 @@
+use axum::body::Bytes;
 use std::collections::BTreeMap;
 use std::io::Write;
 
@@ -6,9 +7,13 @@ use crate::section::Section;
 #[derive(Debug)]
 pub struct WriteableSection {
     pub piggyback_hostname: String,
-    pub name: &'static str,
-    /// The JSON serialized body of the section
-    pub body: String,
+    pub body: SectionBody,
+}
+
+#[derive(Debug)]
+pub enum SectionBody {
+    Json { name: &'static str, body: String },
+    Raw(Bytes),
 }
 
 #[derive(Debug)]
@@ -25,10 +30,19 @@ impl WriteableSection {
         })?;
         Ok(Self {
             piggyback_hostname: piggyback_hostname.to_string(),
-            name: S::NAME,
-            body,
+            body: SectionBody::Json {
+                name: S::NAME,
+                body,
+            },
         })
     }
+
+    pub fn from_raw(piggyback_hostname: String, raw: Bytes) -> Self {
+        Self {
+            piggyback_hostname,
+            body: SectionBody::Raw(raw),
+        }
+    }
 }
 
 /// Take a collection of [`WriteableSection`]s and render them into something
@@ -47,9 +61,19 @@ pub fn frame<W: Write>(writer: &mut W, sections: Vec<WriteableSection>) -> std::
             writeln!(writer, "<<<<{host}>>>>")?;
         }
         for section in host_sections {
-            writeln!(writer, "<<<{}:sep(0)>>>", section.name)?;
-            writer.write_all(section.body.as_bytes())?;
-            writeln!(writer)?;
+            match &section.body {
+                SectionBody::Json { name, body } => {
+                    writeln!(writer, "<<<{name}:sep(0)>>>")?;
+                    writer.write_all(body.as_bytes())?;
+                    writeln!(writer)?;
+                }
+                SectionBody::Raw(raw) => {
+                    writer.write_all(raw)?;
+                    if !raw.ends_with(b"\n") && !raw.is_empty() {
+                        writeln!(writer)?;
+                    }
+                }
+            }
         }
         if !bare {
             writeln!(writer, "<<<<>>>>")?;
@@ -81,4 +105,46 @@ mod tests {
         frame(&mut out, sections).unwrap();
         insta::assert_snapshot!(String::from_utf8(out).unwrap());
     }
+
+    #[test]
+    fn frame_raw_and_json_mixed() {
+        let sections = vec![
+            WriteableSection::of("json-host", &TestSectionV1 { value: 7 }).unwrap(),
+            WriteableSection::from_raw(
+                "raw-host".to_string(),
+                Bytes::from_static(b"<<<check_mk>>>\nVersion: 2.5.0\n"),
+            ),
+        ];
+        let mut out = Vec::new();
+        frame(&mut out, sections).unwrap();
+        insta::assert_snapshot!(String::from_utf8(out).unwrap());
+    }
+
+    #[test]
+    fn frame_raw_without_trailing_newline_gets_one_added() {
+        let sections = vec![WriteableSection::from_raw(
+            "raw-host".to_string(),
+            Bytes::from_static(b"<<<check_mk>>>\nVersion: 2.5.0"),
+        )];
+        let mut out = Vec::new();
+        frame(&mut out, sections).unwrap();
+        assert_eq!(
+            String::from_utf8(out).unwrap(),
+            "<<<<raw-host>>>>\n<<<check_mk>>>\nVersion: 2.5.0\n<<<<>>>>\n"
+        );
+    }
+
+    #[test]
+    fn frame_raw_empty_produces_no_spurious_blank_line() {
+        let sections = vec![WriteableSection::from_raw(
+            "raw-host".to_string(),
+            Bytes::new(),
+        )];
+        let mut out = Vec::new();
+        frame(&mut out, sections).unwrap();
+        assert_eq!(
+            String::from_utf8(out).unwrap(),
+            "<<<<raw-host>>>>\n<<<<>>>>\n"
+        );
+    }
 }

metrics-cache/src/snapshot/mod.rs

@@ -3,12 +3,15 @@ pub mod metric_tables;
 pub mod owner_graph;
 pub mod self_health;
 
+use axum::body::Bytes;
 use moka::future::Cache;
 use std::borrow::Borrow;
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Instant;
 
 use crate::ingest::MetricsFetcherIngestion;
+use crate::ingest::SystemAgentOutput;
 use crate::ingest::kubelet_stats::StatsSummary;
 use crate::ingest::reflectors::{FrozenStores, Stores};
 use crate::snapshot::indexes::Indexes;
@@ -38,6 +41,7 @@ pub struct Snapshot {
     pub metrics: MetricTables,
     pub indexes: Indexes,
     pub self_health: SelfHealth,
+    pub system_agent_snapshot: HashMap<String, Bytes>,
 }
 
 impl Snapshot {
@@ -46,6 +50,7 @@ impl Snapshot {
     pub fn new(
         stores: Stores,
         kubelet_stats_summary_cache: Cache<String, Arc<MetricsFetcherIngestion<StatsSummary>>>,
+        system_agent_cache: Cache<String, Arc<MetricsFetcherIngestion<SystemAgentOutput>>>,
     ) -> Self {
         let instant = Instant::now();
         let reflector_healths = stores.freeze_healths();
@@ -59,13 +64,18 @@ impl Snapshot {
             reflector_healths,
             &kubelet_stats_summary_cache,
         );
+        let system_agent_snapshot: HashMap<String, Bytes> = system_agent_cache
+            .iter()
+            .map(|(name, ingestion)| (name.to_string(), ingestion.payload.0.clone()))
+            .collect();
         Snapshot {
             instant,
             stores,
             owner_graph,
             metrics,
             indexes,
             self_health,
+            system_agent_snapshot,
         }
     }
 }