added resource monitoring with configurable CPU and memory thresholds

Author: vkhinvasara

src/cli.rs

@@ -317,6 +317,25 @@ pub struct Options {
     )]
     pub parquet_compression: Compression,
 
+    // Resource monitoring
+    #[arg(
+        long,
+        env = "P_CPU_THRESHOLD",
+        default_value = "80.0",
+        value_parser = validation::validate_percentage,
+        help = "CPU utilization threshold percentage (0.0-100.0) for resource monitoring"
+    )]
+    pub cpu_utilization_threshold: f32,
+
+    #[arg(
+        long,
+        env = "P_MEMORY_THRESHOLD", 
+        default_value = "80.0",
+        value_parser = validation::validate_percentage,
+        help = "Memory utilization threshold percentage (0.0-100.0) for resource monitoring"
+    )]
+    pub memory_utilization_threshold: f32,
+
     // Integration features
     #[arg(
         long,

src/handlers/http/modal/ingest_server.rs

@@ -39,7 +39,7 @@ use crate::{
         http::{
             base_path, ingest, logstream,
             middleware::{DisAllowRootUser, RouteExt},
-            role,
+            resource_check, role,
         },
     },
     migration,
@@ -126,12 +126,18 @@ impl ParseableServer for IngestServer {
         let (cancel_tx, cancel_rx) = oneshot::channel();
         thread::spawn(|| sync::handler(cancel_rx));
 
+        // Start resource monitor
+        let (resource_shutdown_tx, resource_shutdown_rx) = oneshot::channel();
+        resource_check::spawn_resource_monitor(resource_shutdown_rx);
+
         tokio::spawn(airplane::server());
 
         // Ingestors shouldn't have to deal with OpenId auth flow
         let result = self.start(shutdown_rx, prometheus.clone(), None).await;
         // Cancel sync jobs
         cancel_tx.send(()).expect("Cancellation should not fail");
+        // Shutdown resource monitor
+        let _ = resource_shutdown_tx.send(());
         if let Err(join_err) = startup_sync_handle.await {
             tracing::warn!("startup sync task panicked: {join_err}");
         }

src/handlers/http/modal/server.rs

@@ -25,6 +25,7 @@ use crate::handlers::http::alerts;
 use crate::handlers::http::base_path;
 use crate::handlers::http::health_check;
 use crate::handlers::http::prism_base_path;
+use crate::handlers::http::resource_check;
 use crate::handlers::http::query;
 use crate::handlers::http::users::dashboards;
 use crate::handlers::http::users::filters;
@@ -138,6 +139,10 @@ impl ParseableServer for Server {
         let (cancel_tx, cancel_rx) = oneshot::channel();
         thread::spawn(|| sync::handler(cancel_rx));
 
+        // Start resource monitor
+        let (resource_shutdown_tx, resource_shutdown_rx) = oneshot::channel();
+        resource_check::spawn_resource_monitor(resource_shutdown_rx);
+
         if PARSEABLE.options.send_analytics {
             analytics::init_analytics_scheduler()?;
         }
@@ -150,6 +155,8 @@ impl ParseableServer for Server {
             .await;
         // Cancel sync jobs
         cancel_tx.send(()).expect("Cancellation should not fail");
+        // Shutdown resource monitor
+        let _ = resource_shutdown_tx.send(());
         if let Err(join_err) = startup_sync_handle.await {
             tracing::warn!("startup sync task panicked: {join_err}");
         }

src/handlers/http/resource_check.rs

@@ -23,51 +23,102 @@ use actix_web::{
     error::ErrorServiceUnavailable,
     middleware::Next,
 };
-use sysinfo::System;
-use tracing::warn;
+use tokio::{select, time::{interval, Duration}};
+use tokio::sync::RwLock;
+use tracing::{warn, trace, info};
 
-const CPU_UTILIZATION_THRESHOLD: f32 = 90.0;
-const MEMORY_UTILIZATION_THRESHOLD: f32 = 90.0;
+use crate::analytics::{SYS_INFO, refresh_sys_info};
+use crate::parseable::PARSEABLE;
+
+static RESOURCE_CHECK_ENABLED: RwLock<bool> = RwLock::const_new(true);
+
+/// Spawn a background task to monitor system resources
+pub fn spawn_resource_monitor(shutdown_rx: tokio::sync::oneshot::Receiver<()>) {
+    tokio::spawn(async move {
+        let mut check_interval = interval(Duration::from_secs(30));
+        let mut shutdown_rx = shutdown_rx;
+        
+        let cpu_threshold = PARSEABLE.options.cpu_utilization_threshold;
+        let memory_threshold = PARSEABLE.options.memory_utilization_threshold;
+        
+        info!("Resource monitor started with thresholds - CPU: {:.1}%, Memory: {:.1}%", 
+              cpu_threshold, memory_threshold);
+        loop {
+            select! {
+                _ = check_interval.tick() => {
+                    trace!("Checking system resource utilization...");
+                    
+                    refresh_sys_info();
+                    let (used_memory, total_memory, cpu_usage) = {
+                        let sys = SYS_INFO.lock().unwrap();
+                        let used_memory = sys.used_memory() as f32;
+                        let total_memory = sys.total_memory() as f32;
+                        let cpu_usage = sys.global_cpu_usage();
+                        (used_memory, total_memory, cpu_usage)
+                    };
+                    
+                    let mut resource_ok = true;
+                    
+                    // Calculate memory usage percentage
+                    let memory_usage = if total_memory > 0.0 {
+                        (used_memory / total_memory) * 100.0
+                    } else {
+                        0.0
+                    };
+                    
+                    // Log current resource usage every few checks for debugging
+                    info!("Current resource usage - CPU: {:.1}%, Memory: {:.1}% ({:.1}GB/{:.1}GB)", 
+                          cpu_usage, memory_usage, 
+                          used_memory / 1024.0 / 1024.0 / 1024.0, 
+                          total_memory / 1024.0 / 1024.0 / 1024.0);
+                    
+                    // Check memory utilization
+                    if memory_usage > memory_threshold {
+                        warn!("High memory usage detected: {:.1}% (threshold: {:.1}%)", 
+                              memory_usage, memory_threshold);
+                        resource_ok = false;
+                    }
+                    
+                    // Check CPU utilization
+                    if cpu_usage > cpu_threshold {
+                        warn!("High CPU usage detected: {:.1}% (threshold: {:.1}%)", 
+                              cpu_usage, cpu_threshold);
+                        resource_ok = false;
+                    }
+                    
+                    let previous_state = *RESOURCE_CHECK_ENABLED.read().await;
+                    *RESOURCE_CHECK_ENABLED.write().await = resource_ok;
+                    
+                    // Log state changes
+                    if previous_state != resource_ok {
+                        if resource_ok {
+                            info!("Resource utilization back to normal - requests will be accepted");
+                        } else {
+                            warn!("Resource utilization too high - requests will be rejected");
+                        }
+                    }
+                },
+                _ = &mut shutdown_rx => {
+                    trace!("Resource monitor shutting down");
+                    break;
+                }
+            }
+        }
+    });
+}
 
 /// Middleware to check system resource utilization before processing requests
-/// Returns 503 Service Unavailable if CPU or memory usage exceeds thresholds
+/// Returns 503 Service Unavailable if resources are over-utilized
 pub async fn check_resource_utilization_middleware(
     req: ServiceRequest,
     next: Next<impl MessageBody>,
 ) -> Result<ServiceResponse<impl MessageBody>, Error> {
 
-    let mut sys = System::new_all();
-    sys.refresh_cpu_usage();
-    sys.refresh_memory();
+    let resource_ok = *RESOURCE_CHECK_ENABLED.read().await;
 
-    let used_memory = sys.used_memory() as f32;
-    let total_memory = sys.total_memory() as f32;
-    
-    // Check memory utilization
-    if total_memory > 0.0 {
-        let memory_usage = (used_memory / total_memory) * 100.0;
-        if memory_usage > MEMORY_UTILIZATION_THRESHOLD {
-            let error_msg = format!("Memory is over-utilized: {:.1}%", memory_usage);
-            warn!(
-                "Rejecting request to {} due to high memory usage: {:.1}% (threshold: {:.1}%)", 
-                req.path(), 
-                memory_usage,
-                MEMORY_UTILIZATION_THRESHOLD
-            );
-            return Err(ErrorServiceUnavailable(error_msg));
-        }
-    }
-
-    // Check CPU utilization
-    let cpu_usage = sys.global_cpu_usage();
-    if cpu_usage > CPU_UTILIZATION_THRESHOLD {
-        let error_msg = format!("CPU is over-utilized: {:.1}%", cpu_usage);
-        warn!(
-            "Rejecting request to {} due to high CPU usage: {:.1}% (threshold: {:.1}%)", 
-            req.path(), 
-            cpu_usage,
-            CPU_UTILIZATION_THRESHOLD
-        );
+    if !resource_ok {
+        let error_msg = "Server resources over-utilized";
+        warn!("Rejecting request to {} due to resource constraints", req.path());
         return Err(ErrorServiceUnavailable(error_msg));
     }
 

src/option.rs

@@ -175,6 +175,18 @@ pub mod validation {
         }
     }
 
+    pub fn validate_percentage(percentage: &str) -> Result<f32, String> {
+        if let Ok(percentage) = percentage.parse::<f32>() {
+            if (0.0..=100.0).contains(&percentage) {
+                Ok(percentage)
+            } else {
+                Err("Invalid percentage value. It should be between 0.0 and 100.0".to_string())
+            }
+        } else {
+            Err("Invalid percentage value. It should be a decimal number like 80.0".to_string())
+        }
+    }
+
     pub fn validate_dataset_fields_allowed_limit(s: &str) -> Result<usize, String> {
         if let Ok(size) = s.parse::<usize>() {
             if (1..=DATASET_FIELD_COUNT_LIMIT).contains(&size) {