chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

[model-gateway] perf: optimize observability logging for minimal CPU/memory overhead (#16039)

Browse Source
This commit is contained in:
Simo Lin
2025-12-28 21:20:46 -08:00
committed by GitHub
parent c236a3fdf0
commit f13949e5a9
5 changed files with 104 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
sgl-model-gateway/src/observability/events.rs
View File

@@ -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::<RequestReceivedEvent>(), 0);
assert_eq!(size_of::<RequestSentEvent>(), 16);
assert_eq!(size_of::<RequestPDSentEvent>(), 32);
}
}

80
sgl-model-gateway/src/observability/logging.rs
View File

@@ -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<WorkerGuard>,
}
pub fn init_logging(config: LoggingConfig, otel_layer_config: Option<TraceConfig>) -> 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::<usize>()
+ 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<TraceConfig>) -> 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::<String>()
} 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<TraceConfig
let mut file_guard = None;
if let Some(log_dir) = &config.log_dir {
let file_name = config.log_file_name.clone();
let log_dir = PathBuf::from(log_dir);
if !log_dir.exists() {
@@ -103,7 +128,8 @@ pub fn init_logging(config: LoggingConfig, otel_layer_config: Option<TraceConfig
}
}
let file_appender = RollingFileAppender::new(Rotation::DAILY, log_dir, file_name);
let file_appender =
RollingFileAppender::new(Rotation::DAILY, log_dir, &config.log_file_name);
let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
file_guard = Some(guard);
@@ -112,7 +138,7 @@ pub fn init_logging(config: LoggingConfig, otel_layer_config: Option<TraceConfig
.with_ansi(false)
.with_file(true)
.with_line_number(true)
.with_timer(ChronoUtc::new(time_format))
.with_timer(ChronoUtc::new(TIME_FORMAT.to_string()))
.with_writer(non_blocking);
let file_layer = if config.json_format {

52
sgl-model-gateway/src/observability/otel_trace.rs
View File

@@ -1,3 +1,5 @@
//! OpenTelemetry tracing integration.
use std::{
sync::{
atomic::{AtomicBool, Ordering},
@@ -28,20 +30,15 @@ use tracing_subscriber::{
use super::events::get_module_path as events_module_path;
static ENABLED: AtomicBool = AtomicBool::new(false);
// Global tracer and provider
static TRACER: OnceLock<SdkTracer> = OnceLock::new();
static PROVIDER: OnceLock<TracerProvider> = 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<S> Filter<S> 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<S>() -> Result<Box<dyn Layer<S> + 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);
}

5
sgl-model-gateway/src/routers/http/pd_router.rs
View File

@@ -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();

6
sgl-model-gateway/src/routers/http/router.rs
View File

@@ -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);