feat: expose averaged process CPU and memory per cluster node

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(10);
+
 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

@@ -25,7 +25,8 @@ 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};
+use std::sync::atomic::{AtomicU64, Ordering};
 
 pub const METRICS_NAMESPACE: &str = env!("CARGO_PKG_NAME");
 
@@ -175,6 +176,79 @@ pub static STAGING_FILES: Lazy<IntGaugeVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
+pub static PROCESS_CPU_USAGE_PERCENT: Lazy<Gauge> = Lazy::new(|| {
+    Gauge::with_opts(
+        Opts::new(
+            "process_cpu_usage_percent",
+            "Current CPU usage percent for this Parseable process",
+        )
+        .namespace(METRICS_NAMESPACE),
+    )
+    .expect("metric can be created")
+});
+
+pub static PROCESS_MEMORY_BYTES: Lazy<Gauge> = Lazy::new(|| {
+    Gauge::with_opts(
+        Opts::new(
+            "process_memory_bytes",
+            "Current resident memory used by this Parseable process in bytes",
+        )
+        .namespace(METRICS_NAMESPACE),
+    )
+    .expect("metric can be created")
+});
+
+const CPU_USAGE_PRECISION: f64 = 1_000.0;
+
+#[derive(Default)]
+struct ProcessMetricsAccumulator {
+    cpu_usage_sum: AtomicU64,
+    memory_bytes_sum: AtomicU64,
+    sample_count: AtomicU64,
+}
+
+impl ProcessMetricsAccumulator {
+    fn record(&self, cpu_usage_percent: f64, memory_bytes: u64) -> (f64, f64) {
+        self.cpu_usage_sum.fetch_add(
+            (cpu_usage_percent * CPU_USAGE_PRECISION).round() as u64,
+            Ordering::Relaxed,
+        );
+        self.memory_bytes_sum
+            .fetch_add(memory_bytes, Ordering::Relaxed);
+        let sample_count = self.sample_count.fetch_add(1, Ordering::Relaxed) + 1;
+
+        (
+            self.cpu_usage_sum.load(Ordering::Relaxed) as f64
+                / sample_count as f64
+                / CPU_USAGE_PRECISION,
+            self.memory_bytes_sum.load(Ordering::Relaxed) as f64 / sample_count as f64,
+        )
+    }
+}
+
+static PROCESS_METRICS_ACCUMULATOR: Lazy<ProcessMetricsAccumulator> =
+    Lazy::new(ProcessMetricsAccumulator::default);
+
+pub fn record_process_metrics_sample(cpu_usage_percent: f64, memory_bytes: u64) {
+    let (average_cpu_usage, average_memory_bytes) =
+        PROCESS_METRICS_ACCUMULATOR.record(cpu_usage_percent, memory_bytes);
+    PROCESS_CPU_USAGE_PERCENT.set(average_cpu_usage);
+    PROCESS_MEMORY_BYTES.set(average_memory_bytes);
+}
+
+#[cfg(test)]
+mod process_metrics_tests {
+    use super::ProcessMetricsAccumulator;
+
+    #[test]
+    fn averages_process_metric_samples() {
+        let accumulator = ProcessMetricsAccumulator::default();
+
+        assert_eq!(accumulator.record(10.0, 100), (10.0, 100.0));
+        assert_eq!(accumulator.record(20.0, 300), (15.0, 200.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 +737,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.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(PROCESS_MEMORY_BYTES.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: f64,
+    process_memory_bytes: 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: 0.0,
+            process_memory_bytes: 0.0,
         }
     }
 }
@@ -113,6 +117,8 @@ impl Metrics {
             event_time: Utc::now().naive_utc(),
             commit: "".to_string(),
             staging: "".to_string(),
+            process_cpu_usage_percent: 0.0,
+            process_memory_bytes: 0.0,
         }
     }
 }
@@ -187,6 +193,10 @@ impl Metrics {
                     "process_resident_memory_bytes" => {
                         prom_dress.process_resident_memory_bytes += val
                     }
+                    "parseable_process_cpu_usage_percent" => {
+                        prom_dress.process_cpu_usage_percent += val
+                    }
+                    "parseable_process_memory_bytes" => prom_dress.process_memory_bytes += val,
                     "parseable_storage_size" => {
                         if sample.labels.get("type").expect("type is present") == "staging" {
                             prom_dress.parseable_storage_size.staging += val;