Add node metrics receiver for host-level system metrics

.github/workflows/ci.yml

@@ -50,7 +50,9 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: nextest
-          args: run --target ${{ matrix.target }} --verbose
+          # node_metrics_receiver pulls in no extra dependencies, so building it here is
+          # nearly free and keeps its code and tests compiled on every PR.
+          args: run --target ${{ matrix.target }} --features node_metrics_receiver --verbose
       - name: run test processor-sdk
         uses: actions-rs/cargo@v1
         with:
@@ -83,8 +85,10 @@ jobs:
       - name: build (no default features)
         uses: actions-rs/cargo@v1
         with:
+          # node_metrics_receiver is built here too: it reads /proc and /sys through libc,
+          # whose statfs field widths differ on 32-bit targets.
+          args: --target ${{ matrix.target }} --no-default-features --features file_receiver,node_metrics_receiver
           command: build
-          args: --target ${{ matrix.target }} --no-default-features --features file_receiver
 
   # Check formatting with rustfmt
   formatting:

Cargo.toml

@@ -132,6 +132,7 @@ default = ["rdkafka", "aws_iam", "file_receiver", "rust_processor"]
 fluent_receiver = ["dep:rmpv", "dep:rmp-serde"]
 file_receiver = ["dep:glob", "dep:notify", "dep:notify-debouncer-mini"]
 kmsg_receiver = ["dep:tempfile"]
+node_metrics_receiver = []
 pprof = ["dep:pprof"]
 pyo3 = ["dep:pyo3", "dep:rotel_python_processor_sdk"]
 rust_processor = ["dep:rotel_rust_processor_sdk", "dep:abi_stable"]

build.rs

@@ -3,9 +3,11 @@ use std::io::Result;
 fn main() -> Result<()> {
     println!("cargo::rustc-check-cfg=cfg(kafka_integration_tests, values(\"true\"))");
     println!("cargo::rustc-check-cfg=cfg(kmsg_integration_tests, values(\"true\"))");
+    println!("cargo::rustc-check-cfg=cfg(node_metrics_integration_tests, values(\"true\"))");
     println!("cargo::rustc-check-cfg=cfg(rust_processor_integration_tests, values(\"true\"))");
     println!("cargo::rerun-if-env-changed=KAFKA_INTEGRATION_TESTS");
     println!("cargo::rerun-if-env-changed=KMSG_INTEGRATION_TESTS");
+    println!("cargo::rerun-if-env-changed=NODE_METRICS_INTEGRATION_TESTS");
     println!("cargo::rerun-if-env-changed=RUST_PROCESSOR_INTEGRATION_TESTS");
     println!("cargo::rerun-if-changed=proto/datadog");
     prost_build::compile_protos(&["proto/datadog/agent_payload.proto"], &["proto/datadog"])?;
@@ -16,6 +18,9 @@ fn main() -> Result<()> {
     let env = std::env::var("KMSG_INTEGRATION_TESTS").unwrap_or("false".to_string());
     println!("cargo:rustc-cfg=kmsg_integration_tests=\"{}\"", env);
 
+    let env = std::env::var("NODE_METRICS_INTEGRATION_TESTS").unwrap_or("false".to_string());
+    println!("cargo:rustc-cfg=node_metrics_integration_tests=\"{}\"", env);
+
     let env = std::env::var("RUST_PROCESSOR_INTEGRATION_TESTS").unwrap_or("false".to_string());
     println!(
         "cargo:rustc-cfg=rust_processor_integration_tests=\"{}\"",

src/init/activation.rs

@@ -72,6 +72,8 @@ fn all_traces_receivers_disabled(rc: &HashMap<Receiver, ReceiverConfig>) -> bool
             ReceiverConfig::File(_) => {} // File receiver doesn't handle traces
             #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
             ReceiverConfig::Kmsg(_) => {} // Kmsg receiver doesn't handle traces
+            #[cfg(feature = "node_metrics_receiver")]
+            ReceiverConfig::NodeMetrics(_) => {} // Node metrics receiver doesn't handle traces
         }
     }
     true
@@ -98,6 +100,8 @@ fn all_metrics_receivers_disabled(rc: &HashMap<Receiver, ReceiverConfig>) -> boo
             ReceiverConfig::File(_) => {} // File receiver doesn't handle metrics
             #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
             ReceiverConfig::Kmsg(_) => {} // Kmsg receiver doesn't handle metrics
+            #[cfg(feature = "node_metrics_receiver")]
+            ReceiverConfig::NodeMetrics(_) => return false, // Node metrics receiver handles metrics
         }
     }
     true
@@ -123,7 +127,78 @@ fn all_logs_receivers_disabled(rc: &HashMap<Receiver, ReceiverConfig>) -> bool {
             ReceiverConfig::File(_) => return false, // File receiver handles logs
             #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
             ReceiverConfig::Kmsg(_) => return false, // Kmsg receiver handles logs
+            #[cfg(feature = "node_metrics_receiver")]
+            ReceiverConfig::NodeMetrics(_) => {} // Node metrics receiver doesn't handle logs
         }
     }
     true
 }
+
+#[cfg(all(test, feature = "node_metrics_receiver"))]
+mod tests {
+    use super::*;
+    use crate::init::config::ExporterConfig;
+    use crate::receivers::node_metrics::config::NodeMetricsConfig;
+
+    /// A receiver map holding only the node metrics receiver
+    fn node_metrics_only() -> HashMap<Receiver, ReceiverConfig> {
+        HashMap::from([(
+            Receiver::NodeMetrics,
+            ReceiverConfig::NodeMetrics(NodeMetricsConfig::default()),
+        )])
+    }
+
+    /// The node metrics receiver is the only source of metrics in its own pipeline.
+    ///
+    /// This single match arm decides whether the metrics pipeline is activated at all:
+    /// folded in with the neighbouring receivers that do not produce metrics, the
+    /// receiver would scrape forever into a pipeline that was never started.
+    #[test]
+    fn test_node_metrics_receiver_keeps_the_metrics_pipeline_enabled() {
+        let receivers = node_metrics_only();
+
+        assert!(
+            !all_metrics_receivers_disabled(&receivers),
+            "a node metrics receiver must keep the metrics pipeline enabled"
+        );
+
+        // The `false` above has to come from the node metrics arm, not from an
+        // unconditionally false result: with no receivers at all, metrics stay disabled.
+        assert!(all_metrics_receivers_disabled(&HashMap::new()));
+
+        // It produces neither traces nor logs, so those pipelines stay disabled — the
+        // arms that would activate a whole pipeline for nothing.
+        assert!(all_traces_receivers_disabled(&receivers));
+        assert!(all_logs_receivers_disabled(&receivers));
+    }
+
+    /// The same decision as seen by the agent, which acts on `TelemetryActivation`.
+    #[test]
+    fn test_activation_with_only_a_node_metrics_receiver() {
+        let exporters = ExporterConfigs {
+            metrics: vec![ExporterConfig::Blackhole],
+            ..Default::default()
+        };
+
+        let activation = TelemetryActivation::from_config(&node_metrics_only(), &exporters, false);
+
+        // Metrics are the only pipeline this receiver feeds: it has both a receiver and
+        // an exporter, so it is the only one that comes up.
+        assert_eq!(activation.metrics, TelemetryState::Active);
+        assert_eq!(activation.traces, TelemetryState::Disabled);
+        assert_eq!(activation.logs, TelemetryState::Disabled);
+    }
+
+    /// Without a metrics exporter there is nowhere to send scrapes, which is reported as
+    /// `NoListeners` rather than as a disabled receiver.
+    #[test]
+    fn test_activation_without_a_metrics_exporter() {
+        let activation = TelemetryActivation::from_config(
+            &node_metrics_only(),
+            &ExporterConfigs::default(),
+            false,
+        );
+
+        assert_eq!(activation.metrics, TelemetryState::NoListeners);
+    }
+}

src/init/agent.rs

@@ -31,6 +31,8 @@ use crate::receivers::kafka::offset_ack_committer::KafkaOffsetCommitter;
 use crate::receivers::kafka::receiver::KafkaReceiver;
 #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
 use crate::receivers::kmsg::receiver::KmsgReceiver;
+#[cfg(feature = "node_metrics_receiver")]
+use crate::receivers::node_metrics::NodeMetricsReceiver;
 use crate::receivers::otlp::otlp_grpc::OTLPGrpcServer;
 use crate::receivers::otlp::otlp_http::OTLPHttpServer;
 use crate::receivers::otlp_output::OTLPOutput;
@@ -1082,6 +1084,17 @@ impl Agent {
                         }
                     });
                 }
+                #[cfg(feature = "node_metrics_receiver")]
+                ReceiverConfig::NodeMetrics(config) => {
+                    let node_metrics_receiver =
+                        NodeMetricsReceiver::new(config.clone(), metrics_output.clone())?;
+
+                    // Spawned directly into the shared task set rather than into its own
+                    // supervised set: this receiver owns a single task, so there is
+                    // nothing to supervise or drain. It runs until cancelled, so an early
+                    // completion here legitimately means something went wrong.
+                    node_metrics_receiver.start(&mut receivers_task_set, &receivers_cancel)?;
+                }
             }
         }
 

src/init/args.rs

@@ -13,6 +13,8 @@ use crate::init::kafka_exporter::KafkaExporterArgs;
 use crate::init::kafka_receiver::KafkaReceiverArgs;
 #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
 use crate::init::kmsg_receiver::KmsgReceiverArgs;
+#[cfg(feature = "node_metrics_receiver")]
+use crate::init::node_metrics_receiver::NodeMetricsReceiverArgs;
 use crate::init::otlp_exporter::OTLPExporterArgs;
 use crate::init::otlp_receiver::OTLPReceiverArgs;
 #[cfg(feature = "prometheus")]
@@ -82,6 +84,10 @@ pub struct AgentRun {
     #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
     pub kmsg_receiver: KmsgReceiverArgs,
 
+    #[command(flatten)]
+    #[cfg(feature = "node_metrics_receiver")]
+    pub node_metrics_receiver: NodeMetricsReceiverArgs,
+
     /// Single receiver (type)
     #[arg(value_enum, long, env = "ROTEL_RECEIVER")]
     pub receiver: Option<Receiver>,
@@ -214,6 +220,8 @@ impl Default for AgentRun {
             file_receiver: FileReceiverArgs::default(),
             #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
             kmsg_receiver: KmsgReceiverArgs::default(),
+            #[cfg(feature = "node_metrics_receiver")]
+            node_metrics_receiver: NodeMetricsReceiverArgs::default(),
             otlp_with_trace_processor: Vec::new(),
             otlp_with_logs_processor: Vec::new(),
             otlp_with_metrics_processor: Vec::new(),
@@ -317,6 +325,9 @@ pub enum Receiver {
     File,
     #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
     Kmsg,
+    #[cfg(feature = "node_metrics_receiver")]
+    #[clap(name = "node_metrics")]
+    NodeMetrics,
 }
 
 impl FromStr for Receiver {
@@ -332,6 +343,8 @@ impl FromStr for Receiver {
             "file" => Ok(Receiver::File),
             #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
             "kmsg" => Ok(Receiver::Kmsg),
+            #[cfg(feature = "node_metrics_receiver")]
+            "node_metrics" => Ok(Receiver::NodeMetrics),
             _ => Err("Unknown receiver"),
         }
     }

src/init/config.rs

@@ -28,6 +28,8 @@ use crate::receivers::fluent::config::FluentReceiverConfig;
 use crate::receivers::kafka::config::KafkaReceiverConfig;
 #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
 use crate::receivers::kmsg::config::KmsgReceiverConfig;
+#[cfg(feature = "node_metrics_receiver")]
+use crate::receivers::node_metrics::config::NodeMetricsConfig;
 use crate::receivers::otlp::OTLPReceiverConfig;
 use figment::{Figment, providers::Env};
 use gethostname::gethostname;
@@ -214,6 +216,8 @@ pub(crate) enum ReceiverConfig {
     File(FileReceiverConfig),
     #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
     Kmsg(KmsgReceiverConfig),
+    #[cfg(feature = "node_metrics_receiver")]
+    NodeMetrics(NodeMetricsConfig),
 }
 
 impl TryIntoConfig for ExporterArgs {
@@ -650,6 +654,10 @@ fn get_receiver_config(config: &AgentRun, receiver: Receiver) -> Result<Receiver
         Receiver::File => ReceiverConfig::File(config.file_receiver.build_config()),
         #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
         Receiver::Kmsg => ReceiverConfig::Kmsg(config.kmsg_receiver.build_config()),
+        #[cfg(feature = "node_metrics_receiver")]
+        Receiver::NodeMetrics => {
+            ReceiverConfig::NodeMetrics(config.node_metrics_receiver.build_config()?)
+        }
     })
 }
 

src/init/mod.rs

@@ -17,6 +17,9 @@ mod file_receiver;
 #[cfg(all(target_os = "linux", feature = "kmsg_receiver"))]
 mod kmsg_receiver;
 
+#[cfg(feature = "node_metrics_receiver")]
+mod node_metrics_receiver;
+
 mod awsemf_exporter;
 mod clickhouse_exporter;
 mod datadog_exporter;