diff --git a/src/handlers/http/resource_check.rs b/src/handlers/http/resource_check.rs index aaf3595df..4cbcdb225 100644 --- a/src/handlers/http/resource_check.rs +++ b/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> = 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; diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 57ca55a03..a705b89e2 100644 --- a/src/metrics/mod.rs +++ b/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 = Lazy::new(|| { .expect("metric can be created") }); +pub static PROCESS_CPU_USAGE_PERCENT: Lazy = 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 = 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 = + 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 = 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"); diff --git a/src/metrics/prom_utils.rs b/src/metrics/prom_utils.rs index 3f04d89f6..9d63bd78a 100644 --- a/src/metrics/prom_utils.rs +++ b/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;