From f13949e5a99dbacc0f2e5b52b5205affbb950008 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Sun, 28 Dec 2025 21:20:46 -0800 Subject: [PATCH] [model-gateway] perf: optimize observability logging for minimal CPU/memory overhead (#16039) --- sgl-model-gateway/src/observability/events.rs | 47 +++++++---- .../src/observability/logging.rs | 80 ++++++++++++------- .../src/observability/otel_trace.rs | 52 ++++-------- .../src/routers/http/pd_router.rs | 5 +- sgl-model-gateway/src/routers/http/router.rs | 6 +- 5 files changed, 104 insertions(+), 86 deletions(-) diff --git a/sgl-model-gateway/src/observability/events.rs b/sgl-model-gateway/src/observability/events.rs index 0001cf80b..c6595a744 100644 --- a/sgl-model-gateway/src/observability/events.rs +++ b/sgl-model-gateway/src/observability/events.rs @@ -1,31 +1,30 @@ //! Request events for observability and monitoring. //! -//! Events use conditional log levels: -//! - DEBUG when OTEL is disabled (keeps logs quiet) -//! - INFO when OTEL is enabled (passes through EnvFilter to OTEL layer) +//! Events use DEBUG level when OTEL is disabled, INFO when enabled. use tracing::{debug, event, Level}; use super::otel_trace::is_otel_enabled; /// Module path used by CustomOtelFilter to identify events for OTEL export. -pub fn get_module_path() -> &'static str { - module_path!() +#[inline] +pub const fn get_module_path() -> &'static str { + "sgl_model_gateway::observability::events" } -/// Trait for emitting observability events. pub trait Event { fn emit(&self); } /// Event emitted when a prefill-decode request pair is sent. -#[derive(Debug)] -pub struct RequestPDSentEvent { - pub prefill_url: String, - pub decode_url: String, +#[derive(Debug, Clone, Copy)] +pub struct RequestPDSentEvent<'a> { + pub prefill_url: &'a str, + pub decode_url: &'a str, } -impl Event for RequestPDSentEvent { +impl Event for RequestPDSentEvent<'_> { + #[inline] fn emit(&self) { if is_otel_enabled() { event!( @@ -45,12 +44,13 @@ impl Event for RequestPDSentEvent { } /// Event emitted when a request is sent to a worker. -#[derive(Debug)] -pub struct RequestSentEvent { - pub url: String, +#[derive(Debug, Clone, Copy)] +pub struct RequestSentEvent<'a> { + pub url: &'a str, } -impl Event for RequestSentEvent { +impl Event for RequestSentEvent<'_> { + #[inline] fn emit(&self) { if is_otel_enabled() { event!(Level::INFO, url = %self.url, "Sending request"); @@ -61,10 +61,11 @@ impl Event for RequestSentEvent { } /// Event emitted when concurrent requests are received. -#[derive(Debug)] +#[derive(Debug, Clone, Copy)] pub struct RequestReceivedEvent; impl Event for RequestReceivedEvent { + #[inline] fn emit(&self) { if is_otel_enabled() { event!(Level::INFO, "Received concurrent requests"); @@ -73,3 +74,17 @@ impl Event for RequestReceivedEvent { } } } + +#[cfg(test)] +mod tests { + use std::mem::size_of; + + use super::*; + + #[test] + fn test_event_sizes() { + assert_eq!(size_of::(), 0); + assert_eq!(size_of::(), 16); + assert_eq!(size_of::(), 32); + } +} diff --git a/sgl-model-gateway/src/observability/logging.rs b/sgl-model-gateway/src/observability/logging.rs index 4c8afb401..03861c767 100644 --- a/sgl-model-gateway/src/observability/logging.rs +++ b/sgl-model-gateway/src/observability/logging.rs @@ -1,3 +1,5 @@ +//! Logging infrastructure with non-blocking file I/O. + use std::path::PathBuf; use tracing::Level; @@ -13,6 +15,9 @@ use tracing_subscriber::{ use super::otel_trace::get_otel_layer; use crate::config::TraceConfig; +const TIME_FORMAT: &str = "%Y-%m-%d %H:%M:%S"; +const DEFAULT_LOG_TARGET: &str = "sgl_model_gateway"; + #[derive(Debug, Clone)] pub struct LoggingConfig { pub level: Level, @@ -24,6 +29,7 @@ pub struct LoggingConfig { } impl Default for LoggingConfig { + #[inline] fn default() -> Self { Self { level: Level::INFO, @@ -31,56 +37,76 @@ impl Default for LoggingConfig { log_dir: None, colorize: true, log_file_name: "sgl-model-gateway".to_string(), - log_targets: Some(vec!["sgl_model_gateway".to_string()]), + log_targets: Some(vec![DEFAULT_LOG_TARGET.to_string()]), } } } +/// Guard that keeps the file appender thread alive. #[allow(dead_code)] pub struct LogGuard { _file_guard: Option, } -pub fn init_logging(config: LoggingConfig, otel_layer_config: Option) -> LogGuard { - let _ = LogTracer::init(); - - let level_filter = match config.level { +#[inline] +const fn level_to_str(level: Level) -> &'static str { + match level { Level::TRACE => "trace", Level::DEBUG => "debug", Level::INFO => "info", Level::WARN => "warn", Level::ERROR => "error", - }; + } +} + +#[inline] +fn build_filter_string(targets: &[String], level_filter: &str) -> String { + // Exact capacity: sum of target lengths + "=" and level per target + commas between + let capacity = targets.iter().map(String::len).sum::() + + targets.len() * (1 + level_filter.len()) + + targets.len().saturating_sub(1); + let mut filter_string = String::with_capacity(capacity); + + for (i, target) in targets.iter().enumerate() { + if i > 0 { + filter_string.push(','); + } + filter_string.push_str(target); + filter_string.push('='); + filter_string.push_str(level_filter); + } + + filter_string +} + +pub fn init_logging(config: LoggingConfig, otel_layer_config: Option) -> LogGuard { + let _ = LogTracer::init(); + + let level_filter = level_to_str(config.level); let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| { - let filter_string = if let Some(targets) = &config.log_targets { - targets - .iter() - .enumerate() - .map(|(i, target)| { - if i > 0 { - format!(",{}={}", target, level_filter) - } else { - format!("{}={}", target, level_filter) - } - }) - .collect::() - } else { - format!("sgl_model_gateway={}", level_filter) + let filter_string = match &config.log_targets { + Some(targets) if !targets.is_empty() => build_filter_string(targets, level_filter), + _ => { + let mut s = + String::with_capacity(DEFAULT_LOG_TARGET.len() + 1 + level_filter.len()); + s.push_str(DEFAULT_LOG_TARGET); + s.push('='); + s.push_str(level_filter); + s + } }; EnvFilter::new(filter_string) }); - let mut layers = Vec::new(); - - let time_format = "%Y-%m-%d %H:%M:%S".to_string(); + let mut layers = Vec::with_capacity(3); let stdout_layer = tracing_subscriber::fmt::layer() .with_ansi(config.colorize) .with_file(true) .with_line_number(true) - .with_timer(ChronoUtc::new(time_format.clone())); + .with_timer(ChronoUtc::new(TIME_FORMAT.to_string())); let stdout_layer = if config.json_format { stdout_layer.json().flatten_event(true).boxed() @@ -93,7 +119,6 @@ pub fn init_logging(config: LoggingConfig, otel_layer_config: Option = OnceLock::new(); static PROVIDER: OnceLock = OnceLock::new(); - -/// Targets allowed for OTEL export. Using a static slice avoids allocations. -/// Note: "sgl_model_gateway::otel-trace" is a custom target used for manual spans, -/// not the actual module path. static ALLOWED_TARGETS: OnceLock<[&'static str; 3]> = OnceLock::new(); +#[inline] fn get_allowed_targets() -> &'static [&'static str; 3] { ALLOWED_TARGETS.get_or_init(|| { [ - "sgl_model_gateway::otel-trace", // Custom target for manual spans + "sgl_model_gateway::otel-trace", "sgl_model_gateway::observability::otel_trace", events_module_path(), ] @@ -49,12 +46,12 @@ fn get_allowed_targets() -> &'static [&'static str; 3] { } /// Filter that only allows specific module targets to be exported to OTEL. -/// This reduces noise and cost by only exporting relevant spans. -#[derive(Clone)] +#[derive(Clone, Copy, Default)] pub struct CustomOtelFilter; impl CustomOtelFilter { - pub fn new() -> Self { + #[inline] + pub const fn new() -> Self { Self } @@ -70,10 +67,12 @@ impl Filter for CustomOtelFilter where S: Subscriber, { + #[inline] fn enabled(&self, meta: &Metadata<'_>, _cx: &Context<'_, S>) -> bool { Self::is_allowed(meta.target()) } + #[inline] fn callsite_enabled(&self, meta: &'static Metadata<'static>) -> tracing::subscriber::Interest { if Self::is_allowed(meta.target()) { tracing::subscriber::Interest::always() @@ -83,17 +82,6 @@ where } } -impl Default for CustomOtelFilter { - fn default() -> Self { - Self::new() - } -} - -/// Initialize OpenTelemetry tracing with OTLP exporter. -/// -/// # Arguments -/// * `enable` - Whether to enable OTEL tracing -/// * `otlp_endpoint` - OTLP collector endpoint (defaults to "localhost:4317") pub fn otel_tracing_init(enable: bool, otlp_endpoint: Option<&str>) -> Result<()> { if !enable { ENABLED.store(false, Ordering::Relaxed); @@ -156,9 +144,7 @@ pub fn otel_tracing_init(enable: bool, otlp_endpoint: Option<&str>) -> Result<() Ok(()) } -/// Get the OpenTelemetry tracing layer to add to the subscriber. -/// -/// Must be called after `otel_tracing_init` with `enable=true`. +/// Get the OpenTelemetry tracing layer. Must be called after `otel_tracing_init`. pub fn get_otel_layer() -> Result + Send + Sync + 'static>> where S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a> + Send + Sync, @@ -179,15 +165,11 @@ where Ok(Box::new(layer)) } -/// Returns whether OpenTelemetry tracing is enabled. #[inline] pub fn is_otel_enabled() -> bool { ENABLED.load(Ordering::Relaxed) } -/// Flush all pending spans to the OTLP collector. -/// -/// This is useful before shutdown or when you need to ensure spans are exported. pub async fn flush_spans_async() -> Result<()> { if !is_otel_enabled() { return Ok(()); @@ -205,7 +187,6 @@ pub async fn flush_spans_async() -> Result<()> { Ok(()) } -/// Shutdown OpenTelemetry tracing and flush remaining spans. pub fn shutdown_otel() { if ENABLED.load(Ordering::Relaxed) { global::shutdown_tracer_provider(); @@ -215,9 +196,7 @@ pub fn shutdown_otel() { } /// Inject W3C trace context headers into an HTTP request. -/// -/// This propagates the current span context to downstream services. -/// Does nothing if OTEL is not enabled. +#[inline] pub fn inject_trace_context_http(headers: &mut HeaderMap) { if !is_otel_enabled() { return; @@ -228,6 +207,7 @@ pub fn inject_trace_context_http(headers: &mut HeaderMap) { struct HeaderInjector<'a>(&'a mut HeaderMap); impl opentelemetry::propagation::Injector for HeaderInjector<'_> { + #[inline] fn set(&mut self, key: &str, value: String) { if let Ok(header_name) = HeaderName::from_bytes(key.as_bytes()) { if let Ok(header_value) = HeaderValue::from_str(&value) { @@ -243,9 +223,7 @@ pub fn inject_trace_context_http(headers: &mut HeaderMap) { } /// Inject W3C trace context into gRPC metadata. -/// -/// This propagates the current span context to downstream gRPC services. -/// Does nothing if OTEL is not enabled. +#[inline] pub fn inject_trace_context_grpc(metadata: &mut MetadataMap) { if !is_otel_enabled() { return; @@ -256,9 +234,9 @@ pub fn inject_trace_context_grpc(metadata: &mut MetadataMap) { struct MetadataInjector<'a>(&'a mut MetadataMap); impl opentelemetry::propagation::Injector for MetadataInjector<'_> { + #[inline] fn set(&mut self, key: &str, value: String) { - // gRPC metadata keys must be lowercase ASCII - if let Ok(metadata_key) = MetadataKey::from_bytes(key.to_lowercase().as_bytes()) { + if let Ok(metadata_key) = MetadataKey::from_bytes(key.as_bytes()) { if let Ok(metadata_value) = MetadataValue::try_from(&value) { self.0.insert(metadata_key, metadata_value); } diff --git a/sgl-model-gateway/src/routers/http/pd_router.rs b/sgl-model-gateway/src/routers/http/pd_router.rs index 308f4eeee..4408c44e6 100644 --- a/sgl-model-gateway/src/routers/http/pd_router.rs +++ b/sgl-model-gateway/src/routers/http/pd_router.rs @@ -565,9 +565,10 @@ impl PDRouter { ); // Send both requests concurrently and wait for both + // Note: Using borrowed references avoids heap allocation events::RequestPDSentEvent { - prefill_url: prefill.url().to_string(), - decode_url: decode.url().to_string(), + prefill_url: prefill.url(), + decode_url: decode.url(), } .emit(); diff --git a/sgl-model-gateway/src/routers/http/router.rs b/sgl-model-gateway/src/routers/http/router.rs index a8956e2a5..e530da882 100644 --- a/sgl-model-gateway/src/routers/http/router.rs +++ b/sgl-model-gateway/src/routers/http/router.rs @@ -303,10 +303,8 @@ impl Router { let load_guard = (policy.name() == "cache_aware").then(|| WorkerLoadGuard::new(worker.clone())); - events::RequestSentEvent { - url: worker.url().to_string(), - } - .emit(); + // Note: Using borrowed reference avoids heap allocation + events::RequestSentEvent { url: worker.url() }.emit(); let mut headers_with_trace = headers.cloned().unwrap_or_default(); inject_trace_context_http(&mut headers_with_trace); let headers = Some(&headers_with_trace);