feat: expose averaged process CPU and memory per cluster node (#1743)

* feat: expose averaged process CPU and memory per cluster node

* rename: clarify process CPU/memory metrics as lifetime averages

* EWMA instead of lifetime avg

* coderabbit suggestion

---------

Co-authored-by: Anant Vindal <anant.v09@protonmail.com>

Author: praveen5959

src/handlers/http/resource_check.rs

@@ -32,8 +32,11 @@ use tokio::{
 use tracing::{info, trace, warn};
 
 use crate::analytics::{SYS_INFO, refresh_sys_info};
+use crate::metrics::record_process_metrics_sample;
 use crate::parseable::PARSEABLE;
 
+const PROCESS_METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(5);
+
 static RESOURCE_CHECK_ENABLED: LazyLock<Arc<AtomicBool>> =
     LazyLock::new(|| Arc::new(AtomicBool::new(false)));
 
@@ -42,6 +45,7 @@ pub fn spawn_resource_monitor(shutdown_rx: tokio::sync::oneshot::Receiver<()>) {
     tokio::spawn(async move {
         let resource_check_interval = PARSEABLE.options.resource_check_interval;
         let mut check_interval = interval(Duration::from_secs(resource_check_interval));
+        let mut process_metrics_interval = interval(PROCESS_METRICS_SAMPLE_INTERVAL);
         let mut shutdown_rx = shutdown_rx;
 
         let cpu_threshold = PARSEABLE.options.cpu_utilization_threshold;
@@ -106,6 +110,20 @@ pub fn spawn_resource_monitor(shutdown_rx: tokio::sync::oneshot::Receiver<()>) {
                         }
                     }
                 },
+                _ = process_metrics_interval.tick() => {
+                    refresh_sys_info();
+                    let process_metrics = tokio::task::spawn_blocking(|| {
+                        let sys = SYS_INFO.lock().unwrap();
+                        sysinfo::get_current_pid()
+                            .ok()
+                            .and_then(|pid| sys.process(pid))
+                            .map(|process| (process.cpu_usage() as f64, process.memory()))
+                    }).await.unwrap();
+
+                    if let Some((cpu_usage, memory_bytes)) = process_metrics {
+                        record_process_metrics_sample(cpu_usage, memory_bytes);
+                    }
+                },
                 _ = &mut shutdown_rx => {
                     trace!("Resource monitor shutting down");
                     break;

src/metrics/mod.rs

@@ -17,6 +17,8 @@
  */
 
 pub mod prom_utils;
+use std::sync::OnceLock;
+
 use crate::{
     handlers::{TelemetryType, http::metrics_path},
     stats::FullStats,
@@ -25,7 +27,10 @@ use actix_web::Responder;
 use actix_web_prometheus::{PrometheusMetrics, PrometheusMetricsBuilder};
 use error::MetricsError;
 use once_cell::sync::Lazy;
-use prometheus::{HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts, Registry};
+use prometheus::{
+    Gauge, HistogramOpts, HistogramVec, IntCounterVec, IntGaugeVec, Opts, Registry,
+    core::{Atomic, AtomicF64},
+};
 
 pub const METRICS_NAMESPACE: &str = env!("CARGO_PKG_NAME");
 
@@ -175,6 +180,97 @@ pub static STAGING_FILES: Lazy<IntGaugeVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
+pub static PROCESS_CPU_USAGE_PERCENT_AVG: Lazy<Gauge> = Lazy::new(|| {
+    Gauge::with_opts(
+        Opts::new(
+            "process_cpu_usage_percent_avg",
+            "Lifetime average CPU usage percent for this Parseable process",
+        )
+        .namespace(METRICS_NAMESPACE),
+    )
+    .expect("metric can be created")
+});
+
+pub static PROCESS_MEMORY_BYTES_AVG: Lazy<Gauge> = Lazy::new(|| {
+    Gauge::with_opts(
+        Opts::new(
+            "process_memory_bytes_avg",
+            "Lifetime average resident memory used by this Parseable process in bytes",
+        )
+        .namespace(METRICS_NAMESPACE),
+    )
+    .expect("metric can be created")
+});
+pub static PROCESS_METRICS_INIT: OnceLock<(f64, u64)> = OnceLock::new();
+struct ProcessMetricsAccumulator {
+    cpu_usage_avg: AtomicF64,
+    memory_bytes_avg: AtomicF64,
+}
+
+impl Default for ProcessMetricsAccumulator {
+    fn default() -> Self {
+        // PROCESS_METRICS_INIT must be initialized by now
+        let (cpu, mem) = *PROCESS_METRICS_INIT.get().unwrap();
+        Self {
+            cpu_usage_avg: AtomicF64::new(cpu),
+            memory_bytes_avg: AtomicF64::new(mem as f64),
+        }
+    }
+}
+
+impl ProcessMetricsAccumulator {
+    fn record(&self, cpu_usage_percent: f64, memory_bytes: u64) -> (f64, f64) {
+        // Exponentially Weighted Moving Average is better than
+        // a lifetime average
+        // A spike which occurred 5 days ago should not affect the average utilization
+        // for the last minute
+        // α = 1 - exp(-Δt / τ) = 1 - exp(-5/60) ≈ 0.0800
+        // S_new = S_old + α * (x_new - S_old)
+        let s_cpu_old = self.cpu_usage_avg.get();
+        let s_cpu_new = s_cpu_old + 0.08 * (cpu_usage_percent - s_cpu_old);
+
+        let s_mem_old = self.memory_bytes_avg.get();
+        let s_mem_new = s_mem_old + 0.08 * (memory_bytes as f64 - s_mem_old);
+
+        // update accumulator
+        self.cpu_usage_avg.set(s_cpu_new);
+        self.memory_bytes_avg.set(s_mem_new);
+
+        (s_cpu_new, s_mem_new)
+    }
+}
+
+static PROCESS_METRICS_ACCUMULATOR: Lazy<ProcessMetricsAccumulator> =
+    Lazy::new(ProcessMetricsAccumulator::default);
+
+pub fn record_process_metrics_sample(cpu_usage_percent: f64, memory_bytes: u64) {
+    if PROCESS_METRICS_INIT.get().is_none() {
+        // first measurement
+        let _ = PROCESS_METRICS_INIT.set((cpu_usage_percent, memory_bytes));
+    }
+    let (average_cpu_usage, average_memory_bytes) =
+        PROCESS_METRICS_ACCUMULATOR.record(cpu_usage_percent, memory_bytes);
+    PROCESS_CPU_USAGE_PERCENT_AVG.set(average_cpu_usage);
+    PROCESS_MEMORY_BYTES_AVG.set(average_memory_bytes);
+}
+
+#[cfg(test)]
+mod process_metrics_tests {
+    use crate::metrics::PROCESS_METRICS_INIT;
+
+    use super::ProcessMetricsAccumulator;
+
+    #[test]
+    fn averages_process_metric_samples() {
+        // init PROCESS_METRICS_INIT
+        PROCESS_METRICS_INIT.get_or_init(|| (10.0, 100));
+        let accumulator = ProcessMetricsAccumulator::default();
+
+        assert_eq!(accumulator.record(10.0, 100), (10.0, 100.0));
+        assert_eq!(accumulator.record(20.0, 300), (10.8, 116.0));
+    }
+}
+
 pub static QUERY_EXECUTE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     HistogramVec::new(
         HistogramOpts::new("query_execute_time", "Query execute time").namespace(METRICS_NAMESPACE),
@@ -663,6 +759,12 @@ fn custom_metrics(registry: &Registry) {
     registry
         .register(Box::new(STAGING_FILES.clone()))
         .expect("metric can be registered");
+    registry
+        .register(Box::new(PROCESS_CPU_USAGE_PERCENT_AVG.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(PROCESS_MEMORY_BYTES_AVG.clone()))
+        .expect("metric can be registered");
     registry
         .register(Box::new(QUERY_EXECUTE_TIME.clone()))
         .expect("metric can be registered");

src/metrics/prom_utils.rs

@@ -53,6 +53,8 @@ pub struct Metrics {
     event_time: NaiveDateTime,
     commit: String,
     staging: String,
+    process_cpu_usage_percent_avg: f64,
+    process_memory_bytes_avg: f64,
 }
 
 #[derive(Debug, Serialize, Default, Clone)]
@@ -89,6 +91,8 @@ impl Default for Metrics {
             event_time: Utc::now().naive_utc(),
             commit: "".to_string(),
             staging: "".to_string(),
+            process_cpu_usage_percent_avg: 0.0,
+            process_memory_bytes_avg: 0.0,
         }
     }
 }
@@ -113,6 +117,8 @@ impl Metrics {
             event_time: Utc::now().naive_utc(),
             commit: "".to_string(),
             staging: "".to_string(),
+            process_cpu_usage_percent_avg: 0.0,
+            process_memory_bytes_avg: 0.0,
         }
     }
 }
@@ -187,6 +193,12 @@ impl Metrics {
                     "process_resident_memory_bytes" => {
                         prom_dress.process_resident_memory_bytes += val
                     }
+                    "parseable_process_cpu_usage_percent_avg" => {
+                        prom_dress.process_cpu_usage_percent_avg += val
+                    }
+                    "parseable_process_memory_bytes_avg" => {
+                        prom_dress.process_memory_bytes_avg += val
+                    }
                     "parseable_storage_size" => {
                         if sample.labels.get("type").expect("type is present") == "staging" {
                             prom_dress.parseable_storage_size.staging += val;