diff --git a/src/banner.rs b/src/banner.rs index dd1775687..26d3f3e23 100644 --- a/src/banner.rs +++ b/src/banner.rs @@ -63,11 +63,6 @@ fn status_info(config: &Parseable, scheme: &str, id: Uid) { credentials = "\"Using default creds admin, admin. Please set credentials with P_USERNAME and P_PASSWORD.\"".red().to_string(); } - let llm_status = match &config.options.open_ai_key { - Some(_) => "OpenAI Configured".green(), - None => "Not Configured".grey(), - }; - eprintln!( " Welcome to Parseable Server! Deployment UID: \"{}\"", @@ -79,13 +74,11 @@ fn status_info(config: &Parseable, scheme: &str, id: Uid) { {} Address: {} Credentials: {} - Server Mode: \"{}\" - LLM Status: \"{}\"", + Server Mode: \"{}\"", "Server:".to_string().bold(), address, credentials, config.get_server_mode_string(), - llm_status ); } diff --git a/src/cli.rs b/src/cli.rs index 1cfc796ac..1c986067a 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -512,21 +512,12 @@ pub struct Options { #[arg( long, env = "P_RESOURCE_CHECK_INTERVAL", - default_value = "15", + default_value = "5", value_parser = validation::validate_seconds, help = "Resource monitoring check interval in seconds" )] pub resource_check_interval: u64, - #[arg( - long, - env = "P_CPU_THRESHOLD", - default_value = "100.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", @@ -536,13 +527,13 @@ pub struct Options { )] pub memory_utilization_threshold: f32, - // Integration features #[arg( long, - env = "P_OPENAI_API_KEY", - help = "OpenAI key to enable llm features" + env = "P_RESOURCE_CHECK_ENABLED", + default_value = "false", + help = "Flag to enable resource check" )] - pub open_ai_key: Option, + pub resource_check_enabled: bool, #[arg( long, diff --git a/src/handlers/http/about.rs b/src/handlers/http/about.rs index 816f9de3e..fcf869e87 100644 --- a/src/handlers/http/about.rs +++ b/src/handlers/http/about.rs @@ -66,8 +66,6 @@ pub async fn about() -> Json { let grpc_port = PARSEABLE.options.grpc_port; let store_endpoint = PARSEABLE.storage.get_endpoint(); - let is_llm_active = &PARSEABLE.options.open_ai_key.is_some(); - let llm_provider = is_llm_active.then_some("OpenAI"); let is_oidc_active = PARSEABLE.options.openid().is_some(); let ui_version = option_env!("UI_VERSION").unwrap_or("development"); @@ -99,8 +97,6 @@ pub async fn about() -> Json { "deploymentId": deployment_id, "updateAvailable": update_available, "latestVersion": latest_release, - "llmActive": is_llm_active, - "llmProvider": llm_provider, "oidcActive": is_oidc_active, "license": license_info, "mode": mode, diff --git a/src/handlers/http/modal/ingest_server.rs b/src/handlers/http/modal/ingest_server.rs index 1910c7d1a..26e158dea 100644 --- a/src/handlers/http/modal/ingest_server.rs +++ b/src/handlers/http/modal/ingest_server.rs @@ -20,6 +20,7 @@ use std::sync::Arc; use std::thread; use actix_web::Scope; +use actix_web::middleware::from_fn; use actix_web::web; use actix_web_prometheus::PrometheusMetrics; use async_trait::async_trait; @@ -31,6 +32,7 @@ use tokio::sync::oneshot; use crate::handlers::http::middleware::IntraClusterRequest; use crate::handlers::http::modal::NodeType; +use crate::handlers::http::resource_check; use crate::sync::sync_start; use crate::{ Server, analytics, @@ -67,7 +69,9 @@ impl ParseableServer for IngestServer { .service( // Base path "{url}/api/v1" web::scope(&base_path()) - .service(Server::get_ingest_factory()) + .service(Server::get_ingest_factory().wrap(from_fn( + resource_check::check_resource_utilization_middleware, + ))) .service(Self::logstream_api()) .service(Server::get_about_factory()) .service(Self::analytics_factory()) @@ -78,7 +82,9 @@ impl ParseableServer for IngestServer { .service(Server::get_readiness_factory()) .service(Server::get_otel_generator_ingest_webscope()), ) - .service(Server::get_ingest_otel_factory()); + .service(Server::get_ingest_otel_factory().wrap(from_fn( + resource_check::check_resource_utilization_middleware, + ))); } async fn load_metadata(&self) -> anyhow::Result> { @@ -231,7 +237,10 @@ impl IngestServer { .route( web::post() .to(ingest::post_event) - .authorize_for_resource(Action::Ingest), + .authorize_for_resource(Action::Ingest) + .wrap(from_fn( + resource_check::check_resource_utilization_middleware, + )), ), ) .service( diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs index 266402ee9..97a22c64a 100644 --- a/src/handlers/http/modal/server.rs +++ b/src/handlers/http/modal/server.rs @@ -34,6 +34,7 @@ use crate::handlers::http::otel_generator::{ use crate::handlers::http::prism_base_path; use crate::handlers::http::query; use crate::handlers::http::query_context; +use crate::handlers::http::resource_check; use crate::handlers::http::targets; use crate::handlers::http::users::dashboards; use crate::handlers::http::users::filters; @@ -51,6 +52,7 @@ use crate::sync::sync_start; use crate::handlers::http::alert_target_policy; use actix_web::Resource; use actix_web::Scope; +use actix_web::middleware::from_fn; use actix_web::web; use actix_web::web::resource; use actix_web_prometheus::PrometheusMetrics; @@ -86,7 +88,9 @@ impl ParseableServer for Server { web::scope(&base_path()) .service(Self::get_query_factory()) .service(Self::get_query_context_factory()) - .service(Self::get_ingest_factory()) + .service(Self::get_ingest_factory().wrap(from_fn( + resource_check::check_resource_utilization_middleware, + ))) .service(Self::get_liveness_factory()) .service(Self::get_readiness_factory()) .service(Self::get_about_factory()) @@ -115,7 +119,9 @@ impl ParseableServer for Server { .service(Self::get_traces_webscope()) .service(Self::get_dataset_stats_webscope()), ) - .service(Self::get_ingest_otel_factory()) + .service(Self::get_ingest_otel_factory().wrap(from_fn( + resource_check::check_resource_utilization_middleware, + ))) .service(Self::get_generated()); } @@ -529,7 +535,10 @@ impl Server { .route( web::post() .to(ingest::post_event) - .authorize_for_resource(Action::Ingest), + .authorize_for_resource(Action::Ingest) + .wrap(from_fn( + resource_check::check_resource_utilization_middleware, + )), ) // DELETE "/logstream/{logstream}" ==> Delete log stream .route( diff --git a/src/handlers/http/resource_check.rs b/src/handlers/http/resource_check.rs index 77dc1f87e..09c841192 100644 --- a/src/handlers/http/resource_check.rs +++ b/src/handlers/http/resource_check.rs @@ -25,6 +25,7 @@ use actix_web::{ error::ErrorServiceUnavailable, middleware::Next, }; +use sysinfo::{MemoryRefreshKind, RefreshKind, System}; use tokio::{ select, time::{Duration, interval}, @@ -37,8 +38,7 @@ use crate::parseable::PARSEABLE; const PROCESS_METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(5); -static RESOURCE_CHECK_ENABLED: LazyLock> = - LazyLock::new(|| Arc::new(AtomicBool::new(false))); +static SERVER_OK: LazyLock> = LazyLock::new(|| Arc::new(AtomicBool::new(true))); /// Spawn a background task to monitor system resources pub fn spawn_resource_monitor(shutdown_rx: tokio::sync::oneshot::Receiver<()>) { @@ -48,61 +48,33 @@ pub fn spawn_resource_monitor(shutdown_rx: tokio::sync::oneshot::Receiver<()>) { let mut process_metrics_interval = interval(PROCESS_METRICS_SAMPLE_INTERVAL); let mut shutdown_rx = shutdown_rx; - let cpu_threshold = PARSEABLE.options.cpu_utilization_threshold; - let memory_threshold = PARSEABLE.options.memory_utilization_threshold; + let memory_threshold = (PARSEABLE.options.memory_utilization_threshold / 100.0) as f64; - info!( - "Resource monitor started with thresholds - CPU: {:.1}%, Memory: {:.1}%", - cpu_threshold, memory_threshold - ); loop { select! { _ = check_interval.tick() => { - trace!("Checking system resource utilization..."); - + if !PARSEABLE.options.resource_check_enabled { + continue; + } refresh_sys_info(); - let (used_memory, total_memory, cpu_usage) = tokio::task::spawn_blocking(|| { - let sys = SYS_INFO.lock().unwrap(); - let (used_memory, total_memory) = if let Some(cgroup) = sys.cgroup_limits() { - (cgroup.rss as f32,cgroup.total_memory as f32) - } else { - (sys.used_memory() as f32,sys.total_memory() as f32) - }; - let cpu_usage = sys.global_cpu_usage(); - (used_memory, total_memory, cpu_usage) - }).await.unwrap(); let mut resource_ok = true; - // Calculate memory usage percentage - let memory_usage = if total_memory > 0.0 { - (used_memory / total_memory) * 100.0 + let mut s = System::new_with_specifics( + RefreshKind::nothing().with_memory(MemoryRefreshKind::everything()), + ); + if let Some(cgroup) = s.cgroup_limits() { + if (cgroup.rss as f64) > memory_threshold * (cgroup.total_memory as f64) { + resource_ok = false; + } } 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; + s.refresh_memory(); + if (s.used_memory() as f64) > memory_threshold * (s.total_memory() as f64) { + resource_ok = false; + } } - - let previous_state = RESOURCE_CHECK_ENABLED.load(std::sync::atomic::Ordering::SeqCst); - RESOURCE_CHECK_ENABLED.store(resource_ok, std::sync::atomic::Ordering::SeqCst); + let previous_state = SERVER_OK.load(std::sync::atomic::Ordering::SeqCst); + SERVER_OK.store(resource_ok, std::sync::atomic::Ordering::SeqCst); // Log state changes if previous_state != resource_ok { @@ -117,14 +89,18 @@ pub fn spawn_resource_monitor(shutdown_rx: tokio::sync::oneshot::Receiver<()>) { refresh_sys_info(); let process_metrics = tokio::task::spawn_blocking(|| { let sys = SYS_INFO.lock().unwrap(); + let total_mem = if let Some(cgroup) = sys.cgroup_limits() { + cgroup.total_memory + } else { + sys.total_memory() + }; sysinfo::get_current_pid() .ok() .and_then(|pid| sys.process(pid)) - .map(|process| (process.cpu_usage() as f64, process.memory())) + .map(|process| (process.cpu_usage() as f64, process.memory(), total_mem)) }).await.unwrap(); - - if let Some((cpu_usage, memory_bytes)) = process_metrics { - record_process_metrics_sample(cpu_usage, memory_bytes); + if let Some((cpu_usage, memory_bytes, total_mem)) = process_metrics { + record_process_metrics_sample(cpu_usage, memory_bytes, total_mem); } }, _ = &mut shutdown_rx => { @@ -142,7 +118,7 @@ pub async fn check_resource_utilization_middleware( req: ServiceRequest, next: Next, ) -> Result, Error> { - let resource_ok = RESOURCE_CHECK_ENABLED.load(std::sync::atomic::Ordering::SeqCst); + let resource_ok = SERVER_OK.load(std::sync::atomic::Ordering::SeqCst); if !resource_ok { let error_msg = "Server resources over-utilized"; diff --git a/src/lib.rs b/src/lib.rs index 47fb92407..d2e0856c1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -74,10 +74,10 @@ pub use openid; use parseable::PARSEABLE; use reqwest::{Client, ClientBuilder}; pub use rustls; +pub use sysinfo; pub use utils as parseable_utils; pub use {clap, tracing_actix_web, tracing_opentelemetry, tracing_subscriber}; pub use {opentelemetry, opentelemetry_otlp, opentelemetry_proto, opentelemetry_sdk}; - // It is very unlikely that panic will occur when dealing with locks. pub const LOCK_EXPECT: &str = "Thread shouldn't panic while holding a lock"; diff --git a/src/main.rs b/src/main.rs index 34a16e07f..75522e8ef 100644 --- a/src/main.rs +++ b/src/main.rs @@ -20,8 +20,12 @@ use std::process::exit; #[cfg(feature = "kafka")] use parseable::connectors; use parseable::{ - IngestServer, ParseableServer, QueryServer, Server, banner, metrics, option::Mode, - parseable::PARSEABLE, rbac, storage, + IngestServer, ParseableServer, QueryServer, Server, + analytics::{SYS_INFO, refresh_sys_info}, + banner, metrics, + option::Mode, + parseable::PARSEABLE, + rbac, storage, }; use tokio::signal::ctrl_c; use tokio::sync::oneshot; @@ -31,6 +35,8 @@ use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use tracing_subscriber::{EnvFilter, Registry, fmt}; +use crate::metrics::record_process_metrics_sample; + #[actix_web::main] #[cfg_attr(feature = "hotpath", hotpath::main)] async fn main() -> anyhow::Result<()> { @@ -89,6 +95,27 @@ async fn main() -> anyhow::Result<()> { }); let prometheus = metrics::build_metrics_handler(); + // init process metrics + refresh_sys_info(); + let process_metrics = tokio::task::spawn_blocking(|| { + let sys = SYS_INFO.lock().unwrap(); + let total_mem = if let Some(cgroup) = sys.cgroup_limits() { + cgroup.total_memory + } else { + sys.total_memory() + }; + sysinfo::get_current_pid() + .ok() + .and_then(|pid| sys.process(pid)) + .map(|process| (process.cpu_usage() as f64, process.memory(), total_mem)) + }) + .await + .unwrap(); + // first measurement + if let Some((cpu_usage, memory_bytes, total_mem)) = process_metrics { + record_process_metrics_sample(cpu_usage, memory_bytes, total_mem); + } + // Start servers #[cfg(feature = "kafka")] { diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index fae16303a..d4420724d 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -201,19 +201,21 @@ pub static PROCESS_MEMORY_BYTES_AVG: Lazy = Lazy::new(|| { ) .expect("metric can be created") }); -pub static PROCESS_METRICS_INIT: OnceLock<(f64, u64)> = OnceLock::new(); -struct ProcessMetricsAccumulator { +pub static PROCESS_METRICS_INIT: OnceLock<(f64, u64, u64)> = OnceLock::new(); +pub struct ProcessMetricsAccumulator { cpu_usage_avg: AtomicF64, memory_bytes_avg: AtomicF64, + total_memory: AtomicF64, } impl Default for ProcessMetricsAccumulator { fn default() -> Self { // PROCESS_METRICS_INIT must be initialized by now - let (cpu, mem) = *PROCESS_METRICS_INIT.get().unwrap(); + let (cpu, mem, total_mem) = *PROCESS_METRICS_INIT.get().unwrap(); Self { cpu_usage_avg: AtomicF64::new(cpu), memory_bytes_avg: AtomicF64::new(mem as f64), + total_memory: AtomicF64::new(total_mem as f64), } } } @@ -238,15 +240,27 @@ impl ProcessMetricsAccumulator { (s_cpu_new, s_mem_new) } + + pub fn get_cpu(&self) -> f64 { + self.cpu_usage_avg.get() + } + + pub fn get_mem(&self) -> f64 { + self.memory_bytes_avg.get() + } + + pub fn get_total_mem(&self) -> f64 { + self.total_memory.get() + } } -static PROCESS_METRICS_ACCUMULATOR: Lazy = +pub static PROCESS_METRICS_ACCUMULATOR: Lazy = Lazy::new(ProcessMetricsAccumulator::default); -pub fn record_process_metrics_sample(cpu_usage_percent: f64, memory_bytes: u64) { +pub fn record_process_metrics_sample(cpu_usage_percent: f64, memory_bytes: u64, total_mem: u64) { if PROCESS_METRICS_INIT.get().is_none() { // first measurement - let _ = PROCESS_METRICS_INIT.set((cpu_usage_percent, memory_bytes)); + let _ = PROCESS_METRICS_INIT.set((cpu_usage_percent, memory_bytes, total_mem)); } let (average_cpu_usage, average_memory_bytes) = PROCESS_METRICS_ACCUMULATOR.record(cpu_usage_percent, memory_bytes); @@ -263,7 +277,7 @@ mod process_metrics_tests { #[test] fn averages_process_metric_samples() { // init PROCESS_METRICS_INIT - PROCESS_METRICS_INIT.get_or_init(|| (10.0, 100)); + PROCESS_METRICS_INIT.get_or_init(|| (10.0, 100, 100)); let accumulator = ProcessMetricsAccumulator::default(); assert_eq!(accumulator.record(10.0, 100), (10.0, 100.0)); diff --git a/src/query/mod.rs b/src/query/mod.rs index 2d4f38742..3d0dc14b9 100644 --- a/src/query/mod.rs +++ b/src/query/mod.rs @@ -172,18 +172,16 @@ async fn enough_available_memory() -> Result<(), ExecuteError> { let mut s = System::new_with_specifics( RefreshKind::nothing().with_memory(MemoryRefreshKind::everything()), ); - s.refresh_all(); let threshold = (PARSEABLE.options.query_mem_threshold / 100.0) as f64; - let f = async { loop { if let Some(cgroup) = s.cgroup_limits() { - if (cgroup.rss as f64) < threshold * (cgroup.total_memory as f64) { + if (cgroup.rss as f64) > threshold * (cgroup.total_memory as f64) { return; } } else { s.refresh_memory(); - if (s.used_memory() as f64) < threshold * (s.total_memory() as f64) { + if (s.used_memory() as f64) > threshold * (s.total_memory() as f64) { return; } }