feat(host_metrics source): Implement process collection for host metrics

Cargo.toml

@@ -373,6 +373,7 @@ heim = { git = "https://github.com/vectordotdev/heim.git", branch = "update-nix"
 
 # make sure to update the external docs when the Lua version changes
 mlua = { version = "0.10.1", default-features = false, features = ["lua54", "send", "vendored", "macros"], optional = true }
+sysinfo = "0.32.0"
 
 [target.'cfg(windows)'.dependencies]
 windows-service = "0.7.0"

LICENSE-3rdparty.csv

@@ -155,6 +155,7 @@ crc-catalog,https://github.com/akhilles/crc-catalog,MIT OR Apache-2.0,Akhil Vela
 crc32c,https://github.com/zowens/crc32c,Apache-2.0 OR MIT,Zack Owens
 crc32fast,https://github.com/srijs/rust-crc32fast,MIT OR Apache-2.0,"Sam Rijs <[email protected]>, Alex Crichton <[email protected]>"
 crossbeam-channel,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-channel Authors
+crossbeam-deque,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-deque Authors
 crossbeam-epoch,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-epoch Authors
 crossbeam-queue,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-queue Authors
 crossbeam-utils,https://github.com/crossbeam-rs/crossbeam,MIT OR Apache-2.0,The crossbeam-utils Authors
@@ -484,6 +485,7 @@ rand_xorshift,https://github.com/rust-random/rngs,MIT OR Apache-2.0,"The Rand Pr
 ratatui,https://github.com/ratatui/ratatui,MIT,"Florian Dehau <[email protected]>, The Ratatui Developers"
 raw-cpuid,https://github.com/gz/rust-cpuid,MIT,Gerd Zellweger <[email protected]>
 raw-window-handle,https://github.com/rust-windowing/raw-window-handle,MIT OR Apache-2.0 OR Zlib,Osspial <[email protected]>
+rayon,https://github.com/rayon-rs/rayon,MIT OR Apache-2.0,"Niko Matsakis <[email protected]>, Josh Stone <[email protected]>"
 rdkafka,https://github.com/fede1024/rust-rdkafka,MIT,Federico Giraud <[email protected]>
 redis,https://github.com/redis-rs/redis-rs,BSD-3-Clause,The redis Authors
 redox_syscall,https://gitlab.redox-os.org/redox-os/syscall,MIT,Jeremy Soller <[email protected]>
@@ -592,6 +594,7 @@ syn,https://github.com/dtolnay/syn,MIT OR Apache-2.0,David Tolnay <dtolnay@gmail
 syn_derive,https://github.com/Kyuuhachi/syn_derive,MIT OR Apache-2.0,Kyuuhachi <[email protected]>
 sync_wrapper,https://github.com/Actyx/sync_wrapper,Apache-2.0,Actyx AG <[email protected]>
 synstructure,https://github.com/mystor/synstructure,MIT,Nika Layzell <[email protected]>
+sysinfo,https://github.com/GuillaumeGomez/sysinfo,MIT,Guillaume Gomez <[email protected]>
 syslog,https://github.com/Geal/rust-syslog,MIT,[email protected]
 syslog_loose,https://github.com/FungusHumungus/syslog-loose,MIT,Stephen Wakely <[email protected]>
 system-configuration,https://github.com/mullvad/system-configuration-rs,MIT OR Apache-2.0,Mullvad VPN

changelog.d/process_host_metrics.feature.md

@@ -0,0 +1,4 @@
+Allows for process metrics collection on host metrics
+This process metrics can be configured via the `process` option for sources with `type = "host_metrics"`.
+
+authors: leeteng2001

src/sources/host_metrics/mod.rs

@@ -34,6 +34,7 @@ mod disk;
 mod filesystem;
 mod memory;
 mod network;
+mod process;
 
 /// Collector types.
 #[serde_as]
@@ -49,6 +50,9 @@ pub enum Collector {
     /// Metrics related to CPU utilization.
     Cpu,
 
+    /// Metrics related to Process utilization.
+    Process,
+
     /// Metrics related to disk I/O utilization.
     Disk,
 
@@ -125,6 +129,10 @@ pub struct HostMetricsConfig {
     #[configurable(derived)]
     #[serde(default)]
     pub network: network::NetworkConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub process: process::ProcessConfig,
 }
 
 /// Options for the cgroups (controller groups) metrics collector.
@@ -192,6 +200,7 @@ fn default_collectors() -> Option<Vec<Collector>> {
         Collector::Host,
         Collector::Memory,
         Collector::Network,
+        Collector::Process,
     ];
 
     #[cfg(target_os = "linux")]
@@ -220,6 +229,20 @@ fn default_all_devices() -> FilterList {
     }
 }
 
+fn example_processes() -> FilterList {
+    FilterList {
+        includes: Some(vec!["docker".try_into().unwrap()]),
+        excludes: None,
+    }
+}
+
+fn default_all_processes() -> FilterList {
+    FilterList {
+        includes: Some(vec!["*".try_into().unwrap()]),
+        excludes: None,
+    }
+}
+
 const fn default_levels() -> usize {
     100
 }
@@ -354,6 +377,9 @@ impl HostMetrics {
         if self.config.has_collector(Collector::Cpu) {
             self.cpu_metrics(&mut buffer).await;
         }
+        if self.config.has_collector(Collector::Process) {
+            self.process_metrics(&mut buffer).await;
+        }
         if self.config.has_collector(Collector::Disk) {
             self.disk_metrics(&mut buffer).await;
         }
@@ -708,6 +734,7 @@ mod tests {
             #[cfg(target_os = "linux")]
             Collector::CGroups,
             Collector::Cpu,
+            Collector::Process,
             Collector::Disk,
             Collector::Filesystem,
             Collector::Load,

src/sources/host_metrics/process.rs

@@ -0,0 +1,73 @@
+use super::{default_all_processes, example_processes, FilterList, HostMetrics};
+use std::ffi::OsStr;
+use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, UpdateKind};
+use vector_lib::configurable::configurable_component;
+#[cfg(target_os = "linux")]
+use vector_lib::metric_tags;
+
+/// Options for the process metrics collector.
+#[configurable_component]
+#[derive(Clone, Debug, Default)]
+pub struct ProcessConfig {
+    /// Lists of device name patterns to include or exclude in gathering
+    /// network utilization metrics.
+    #[serde(default = "default_all_processes")]
+    #[configurable(metadata(docs::examples = "example_processes()"))]
+    processes: FilterList,
+}
+
+const RUNTIME: &str = "process_runtime";
+const CPU_USAGE: &str = "process_cpu_usage";
+const MEMORY_USAGE: &str = "process_memory_usage";
+
+impl HostMetrics {
+    pub async fn process_metrics(&self, output: &mut super::MetricsBuffer) {
+        let mut sys = System::new();
+        sys.refresh_processes_specifics(
+            ProcessesToUpdate::All,
+            true,
+            ProcessRefreshKind::new()
+                .with_memory()
+                .with_cpu()
+                .with_cmd(UpdateKind::OnlyIfNotSet),
+        );
+        output.name = "process";
+        let sep = OsStr::new(" ");
+        for (pid, process) in sys.processes().iter().filter(|&(_, proc)| {
+            self.config
+                .process
+                .processes
+                .contains_str(proc.name().to_str())
+        }) {
+            let tags = || {
+                metric_tags!(
+                "pid" => pid.as_u32().to_string(),
+                "name" => process.name().to_str().unwrap_or("unknown"),
+                "command" => process.cmd().join(sep).to_str().unwrap_or(""))
+            };
+            output.gauge(CPU_USAGE, process.cpu_usage().into(), tags());
+            output.gauge(MEMORY_USAGE, process.memory() as f64, tags());
+            output.counter(RUNTIME, process.run_time() as f64, tags());
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::{HostMetrics, HostMetricsConfig, MetricsBuffer};
+
+    #[tokio::test]
+    async fn generates_process_metrics() {
+        let mut buffer = MetricsBuffer::new(None);
+        HostMetrics::new(HostMetricsConfig::default())
+            .process_metrics(&mut buffer)
+            .await;
+        let metrics = buffer.metrics;
+        assert!(!metrics.is_empty());
+
+        // All metrics are named process_*
+        assert!(!metrics
+            .iter()
+            .any(|metric| !metric.name().starts_with("process_")));
+    }
+}