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

[model-gateway] Remove legacy RouterMetrics and Rename SmgMetrics to Metrics and smg_labels to metrics_labels (#15160)

Browse Source
This commit is contained in:
Simo Lin
2025-12-15 04:57:42 -08:00
committed by GitHub
parent b098b1ae24
commit 3518b33178
25 changed files with 332 additions and 1114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sgl-model-gateway/src/core/circuit_breaker.rs
+7 -22
View File
@@ -8,7 +8,7 @@ use std::{
use tracing::info;
use crate::observability::metrics::{RouterMetrics, SmgMetrics};
use crate::observability::metrics::Metrics;
/// Circuit breaker configuration
#[derive(Debug, Clone)]
@@ -96,10 +96,7 @@ impl CircuitBreaker {
/// Create a new circuit breaker with custom configuration and metric label
pub fn with_config_and_label(config: CircuitBreakerConfig, metric_label: String) -> Self {
let init_state = CircuitState::Closed;
// New metrics
SmgMetrics::set_worker_cb_state(&metric_label, init_state.to_int());
// Legacy metrics
RouterMetrics::set_cb_state(&metric_label, init_state.to_int());
Metrics::set_worker_cb_state(&metric_label, init_state.to_int());
Self {
state: Arc::new(RwLock::new(init_state)),
consecutive_failures: Arc::new(AtomicU32::new(0)),
@@ -156,10 +153,7 @@ impl CircuitBreaker {
}
let outcome_str = if success { "success" } else { "failure" };
// New metrics
SmgMetrics::record_worker_cb_outcome(&self.metric_label, outcome_str);
// Legacy metrics
RouterMetrics::record_cb_outcome(&self.metric_label, outcome_str);
Metrics::record_worker_cb_outcome(&self.metric_label, outcome_str);
self.publish_gauge_metrics();
}
@@ -238,12 +232,8 @@ impl CircuitBreaker {
let from = old_state.as_str();
let to = new_state.as_str();
info!("Circuit breaker state transition: {} -> {}", from, to);
// New metrics
SmgMetrics::record_worker_cb_transition(&self.metric_label, from, to);
SmgMetrics::set_worker_cb_state(&self.metric_label, new_state.to_int());
// Legacy metrics
RouterMetrics::record_cb_state_transition(&self.metric_label, from, to);
RouterMetrics::set_cb_state(&self.metric_label, new_state.to_int());
Metrics::record_worker_cb_transition(&self.metric_label, from, to);
Metrics::set_worker_cb_state(&self.metric_label, new_state.to_int());
self.publish_gauge_metrics();
}
}
@@ -323,14 +313,9 @@ impl CircuitBreaker {
}
}
// TODO maybe publish whenever the variable is changed
fn publish_gauge_metrics(&self) {
// New metrics
SmgMetrics::set_worker_cb_consecutive_failures(&self.metric_label, self.failure_count());
SmgMetrics::set_worker_cb_consecutive_successes(&self.metric_label, self.success_count());
// Legacy metrics
RouterMetrics::set_cb_consecutive_failures(&self.metric_label, self.failure_count());
RouterMetrics::set_cb_consecutive_successes(&self.metric_label, self.success_count());
Metrics::set_worker_cb_consecutive_failures(&self.metric_label, self.failure_count());
Metrics::set_worker_cb_consecutive_successes(&self.metric_label, self.success_count());
}
}
sgl-model-gateway/src/core/job_queue.rs
+6 -20
View File
@@ -21,7 +21,6 @@ use crate::{
WorkerRemovalRequest,
},
mcp::McpConfig,
observability::metrics::RouterMetrics,
protocols::worker_spec::{JobStatus, WorkerConfigRequest, WorkerUpdateRequest},
workflow::{WorkflowContext, WorkflowEngine, WorkflowId, WorkflowInstanceId, WorkflowStatus},
};
@@ -232,7 +231,6 @@ impl JobQueue {
pub async fn submit(&self, job: Job) -> Result<(), String> {
// Check if context is still alive before accepting jobs
if self.context.upgrade().is_none() {
RouterMetrics::record_job_shutdown_rejected();
return Err("Job queue shutting down: AppContext dropped".to_string());
}
@@ -249,7 +247,6 @@ impl JobQueue {
match self.tx.send(job).await {
Ok(_) => {
let (queue_depth, available_permits) = self.get_load_info();
RouterMetrics::set_job_queue_depth(queue_depth);
debug!(
"Job submitted: type={}, worker={}, queue_depth={}, available_slots={}",
job_type, worker_url, queue_depth, available_permits
@@ -257,7 +254,6 @@ impl JobQueue {
Ok(())
}
Err(_) => {
RouterMetrics::record_job_queue_full();
self.status_map.remove(&worker_url);
let (queue_depth, _) = self.get_load_info();
Err(format!(
@@ -806,40 +802,30 @@ impl JobQueue {
}
}
/// Record job completion metrics and update status
/// Update job status on completion
fn record_job_completion(
job_type: &'static str,
worker_url: &str,
duration: Duration,
_duration: Duration,
result: &Result<String, String>,
status_map: &Arc<DashMap<String, JobStatus>>,
) {
RouterMetrics::record_job_duration(job_type, duration);
match result {
Ok(message) => {
RouterMetrics::record_job_success(job_type);
status_map.remove(worker_url);
debug!(
"Completed job: type={}, worker={}, duration={:.3}s, result={}",
job_type,
worker_url,
duration.as_secs_f64(),
message
"Completed job: type={}, worker={}, result={}",
job_type, worker_url, message
);
}
Err(error) => {
RouterMetrics::record_job_failure(job_type);
status_map.insert(
worker_url.to_string(),
JobStatus::failed(job_type, worker_url, error.clone()),
);
warn!(
"Failed job: type={}, worker={}, duration={:.3}s, error={}",
job_type,
worker_url,
duration.as_secs_f64(),
error
"Failed job: type={}, worker={}, error={}",
job_type, worker_url, error
);
}
}
sgl-model-gateway/src/core/retry.rs
+2 -2
View File
@@ -111,8 +111,8 @@ impl RetryExecutor {
/// resp
/// },
/// |res, _| matches!(res.status(), StatusCode::REQUEST_TIMEOUT | StatusCode::TOO_MANY_REQUESTS | StatusCode::INTERNAL_SERVER_ERROR | StatusCode::BAD_GATEWAY | StatusCode::SERVICE_UNAVAILABLE | StatusCode::GATEWAY_TIMEOUT),
/// |delay, attempt| RouterMetrics::record_retry_backoff_duration(delay, attempt),
/// || RouterMetrics::record_retries_exhausted("/route"),
/// |delay, _attempt| { /* record backoff metrics */ },
/// || { /* record retries exhausted */ },
/// ).await;
/// ```
pub async fn execute_response_with_retry<Op, Fut, ShouldRetry, OnBackoff, OnExhausted>(
sgl-model-gateway/src/core/steps/mcp_registration.rs
+2 -2
View File
@@ -7,7 +7,7 @@ use tracing::{debug, error, info, warn};
use crate::{
app_context::AppContext,
mcp::{config::McpServerConfig, manager::McpManager},
observability::metrics::SmgMetrics,
observability::metrics::Metrics,
workflow::*,
};
@@ -153,7 +153,7 @@ impl StepExecutor for RegisterMcpServerStep {
mcp_manager.register_static_server(config_request.name.clone(), mcp_client);
// Update active MCP servers metric
SmgMetrics::set_mcp_servers_active(mcp_manager.list_servers().len());
Metrics::set_mcp_servers_active(mcp_manager.list_servers().len());
info!("Registered MCP server: {}", config_request.name);
sgl-model-gateway/src/core/steps/worker/local/remove_from_worker_registry.rs
+2 -5
View File
@@ -7,7 +7,7 @@ use tracing::{debug, warn};
use crate::{
app_context::AppContext,
observability::metrics::{RouterMetrics, SmgMetrics},
observability::metrics::Metrics,
workflow::{StepExecutor, StepResult, WorkflowContext, WorkflowError, WorkflowResult},
};
@@ -63,9 +63,6 @@ impl StepExecutor for RemoveFromWorkerRegistryStep {
debug!("Removed {} worker(s) from registry", removed_count);
}
// Update active workers metric (legacy)
RouterMetrics::set_active_workers(app_context.worker_registry.len());
// Update Layer 3 worker pool size metrics for unique configurations
for (worker_type, connection_mode, model_id) in unique_configs {
// Get labels before moving values into get_workers_filtered
@@ -83,7 +80,7 @@ impl StepExecutor for RemoveFromWorkerRegistryStep {
)
.len();
SmgMetrics::set_worker_pool_size(
Metrics::set_worker_pool_size(
worker_type_label,
connection_mode_label,
&model_id,
sgl-model-gateway/src/core/steps/worker/shared/register.rs
+2 -5
View File
@@ -8,7 +8,7 @@ use tracing::debug;
use crate::{
app_context::AppContext,
core::Worker,
observability::metrics::{RouterMetrics, SmgMetrics},
observability::metrics::Metrics,
workflow::{StepExecutor, StepResult, WorkflowContext, WorkflowResult},
};
@@ -37,9 +37,6 @@ impl StepExecutor for RegisterWorkersStep {
worker_ids.push(worker_id);
}
// Update active workers metric (legacy)
RouterMetrics::set_active_workers(app_context.worker_registry.len());
// Collect unique worker configurations to avoid redundant metric updates
let unique_configs: HashSet<_> = workers
.iter()
@@ -70,7 +67,7 @@ impl StepExecutor for RegisterWorkersStep {
)
.len();
SmgMetrics::set_worker_pool_size(
Metrics::set_worker_pool_size(
worker_type_label,
connection_mode_label,
&model_id,
sgl-model-gateway/src/core/worker.rs
+9 -11
View File
@@ -17,7 +17,7 @@ use super::{
};
use crate::{
core::{BasicWorkerBuilder, DPAwareWorkerBuilder},
observability::metrics::{smg_labels, RouterMetrics, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
protocols::worker_spec::WorkerInfo,
routers::grpc::client::GrpcClient,
};
@@ -314,8 +314,8 @@ impl ConnectionMode {
/// Get the metric label for this connection mode
pub fn as_metric_label(&self) -> &'static str {
match self {
ConnectionMode::Http => smg_labels::CONNECTION_HTTP,
ConnectionMode::Grpc { .. } => smg_labels::CONNECTION_GRPC,
ConnectionMode::Http => metrics_labels::CONNECTION_HTTP,
ConnectionMode::Grpc { .. } => metrics_labels::CONNECTION_GRPC,
}
}
}
@@ -404,9 +404,9 @@ impl WorkerType {
/// Get the metric label for this worker type
pub fn as_metric_label(&self) -> &'static str {
match self {
WorkerType::Regular => smg_labels::WORKER_REGULAR,
WorkerType::Prefill { .. } => smg_labels::WORKER_PREFILL,
WorkerType::Decode => smg_labels::WORKER_DECODE,
WorkerType::Regular => metrics_labels::WORKER_REGULAR,
WorkerType::Prefill { .. } => metrics_labels::WORKER_PREFILL,
WorkerType::Decode => metrics_labels::WORKER_DECODE,
}
}
}
@@ -557,8 +557,7 @@ impl BasicWorker {
fn update_running_requests_metrics(&self) {
let load = self.load();
RouterMetrics::set_running_requests(self.url(), load);
SmgMetrics::set_worker_requests_active(self.url(), load);
Metrics::set_worker_requests_active(self.url(), load);
}
}
@@ -586,7 +585,6 @@ impl Worker for BasicWorker {
fn set_healthy(&self, healthy: bool) {
self.healthy.store(healthy, Ordering::Release);
RouterMetrics::set_worker_health(self.url(), healthy);
}
async fn check_health_async(&self) -> WorkerResult<()> {
@@ -603,7 +601,7 @@ impl Worker for BasicWorker {
let successes = self.consecutive_successes.fetch_add(1, Ordering::AcqRel) + 1;
// Record health check success metric
SmgMetrics::record_worker_health_check(worker_type_str, smg_labels::CB_SUCCESS);
Metrics::record_worker_health_check(worker_type_str, metrics_labels::CB_SUCCESS);
if !self.is_healthy()
&& successes >= self.metadata.health_config.success_threshold as usize
@@ -617,7 +615,7 @@ impl Worker for BasicWorker {
let failures = self.consecutive_failures.fetch_add(1, Ordering::AcqRel) + 1;
// Record health check failure metric
SmgMetrics::record_worker_health_check(worker_type_str, smg_labels::CB_FAILURE);
Metrics::record_worker_health_check(worker_type_str, metrics_labels::CB_FAILURE);
if self.is_healthy()
&& failures >= self.metadata.health_config.failure_threshold as usize
sgl-model-gateway/src/core/worker_registry.rs
+1 -5
View File
@@ -7,10 +7,7 @@ use std::sync::{Arc, RwLock};
use dashmap::DashMap;
use uuid::Uuid;
use crate::{
core::{ConnectionMode, RuntimeType, Worker, WorkerType},
observability::metrics::RouterMetrics,
};
use crate::core::{ConnectionMode, RuntimeType, Worker, WorkerType};
/// Unique identifier for a worker
#[derive(Debug, Clone, Hash, Eq, PartialEq)]
@@ -142,7 +139,6 @@ impl WorkerRegistry {
}
worker.set_healthy(false);
RouterMetrics::remove_worker_metrics(worker.url());
Some(worker)
} else {
sgl-model-gateway/src/middleware.rs
+18 -31
View File
@@ -26,7 +26,7 @@ use tracing::{debug, error, field::Empty, info, info_span, warn, Span};
pub use crate::core::token_bucket::TokenBucket;
use crate::{
observability::metrics::{smg_labels, RouterMetrics, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
routers::error::extract_error_code_from_response,
server::AppState,
wasm::{
@@ -308,8 +308,6 @@ impl<B> OnRequest<B> for RequestLogger {
span.record("request_id", request_id.0.as_str());
}
RouterMetrics::record_http_request();
// Log the request start
info!(
target: "sgl_model_gateway::request",
@@ -339,12 +337,8 @@ impl<B> OnResponse<B> for ResponseLogger {
let error_code = extract_error_code_from_response(response);
// TODO support `route` information
RouterMetrics::record_http_status_code(status_code, error_code);
RouterMetrics::record_request_duration(latency);
// New SMG metrics (Layer 1: HTTP)
SmgMetrics::record_http_response(status_code, error_code);
// Layer 1: HTTP metrics
Metrics::record_http_response(status_code, error_code);
// Record these in the span for structured logging/observability tools
span.record("status_code", status_code);
@@ -520,7 +514,7 @@ pub async fn concurrency_limit_middleware(
// Try to acquire token immediately
if token_bucket.try_acquire(1.0).await.is_ok() {
debug!("Acquired token immediately");
SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_ALLOWED);
Metrics::record_http_rate_limit(metrics_labels::RATE_LIMIT_ALLOWED);
let response = next.run(request).await;
// Wrap the response body with TokenGuardBody to return token when stream ends
@@ -545,22 +539,19 @@ pub async fn concurrency_limit_middleware(
// Try to send to queue
match queue_tx.try_send(queued) {
Ok(_) => {
// On successful enqueue, update embeddings queue gauge if applicable
// On successful enqueue, update embeddings queue counter if applicable
if is_embeddings {
let new_val = EMBEDDINGS_QUEUE_SIZE.fetch_add(1, Ordering::Relaxed) + 1;
RouterMetrics::set_embeddings_queue_size(new_val as usize);
EMBEDDINGS_QUEUE_SIZE.fetch_add(1, Ordering::Relaxed);
}
// Wait for token from queue processor
match permit_rx.await {
Ok(Ok(())) => {
debug!("Acquired token from queue");
SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_ALLOWED);
Metrics::record_http_rate_limit(metrics_labels::RATE_LIMIT_ALLOWED);
// Dequeue for embeddings
if is_embeddings {
let new_val =
EMBEDDINGS_QUEUE_SIZE.fetch_sub(1, Ordering::Relaxed) - 1;
RouterMetrics::set_embeddings_queue_size(new_val as usize);
EMBEDDINGS_QUEUE_SIZE.fetch_sub(1, Ordering::Relaxed);
}
let response = next.run(request).await;
@@ -572,23 +563,19 @@ pub async fn concurrency_limit_middleware(
}
Ok(Err(status)) => {
warn!("Queue returned error status: {}", status);
SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED);
Metrics::record_http_rate_limit(metrics_labels::RATE_LIMIT_REJECTED);
// Dequeue for embeddings on error
if is_embeddings {
let new_val =
EMBEDDINGS_QUEUE_SIZE.fetch_sub(1, Ordering::Relaxed) - 1;
RouterMetrics::set_embeddings_queue_size(new_val as usize);
EMBEDDINGS_QUEUE_SIZE.fetch_sub(1, Ordering::Relaxed);
}
status.into_response()
}
Err(_) => {
error!("Queue response channel closed");
SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED);
Metrics::record_http_rate_limit(metrics_labels::RATE_LIMIT_REJECTED);
// Dequeue for embeddings on channel error
if is_embeddings {
let new_val =
EMBEDDINGS_QUEUE_SIZE.fetch_sub(1, Ordering::Relaxed) - 1;
RouterMetrics::set_embeddings_queue_size(new_val as usize);
EMBEDDINGS_QUEUE_SIZE.fetch_sub(1, Ordering::Relaxed);
}
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
@@ -596,13 +583,13 @@ pub async fn concurrency_limit_middleware(
}
Err(_) => {
warn!("Request queue is full, returning 429");
SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED);
Metrics::record_http_rate_limit(metrics_labels::RATE_LIMIT_REJECTED);
StatusCode::TOO_MANY_REQUESTS.into_response()
}
}
} else {
warn!("No tokens available and queuing is disabled, returning 429");
SmgMetrics::record_http_rate_limit(smg_labels::RATE_LIMIT_REJECTED);
Metrics::record_http_rate_limit(metrics_labels::RATE_LIMIT_REJECTED);
StatusCode::TOO_MANY_REQUESTS.into_response()
}
}
@@ -663,22 +650,22 @@ where
Box::pin(async move {
// Increment inside async block - ensures no leak if future is dropped before polling
let active = ACTIVE_HTTP_CONNECTIONS.fetch_add(1, Ordering::Relaxed) + 1;
SmgMetrics::set_http_connections_active(active as usize);
Metrics::set_http_connections_active(active as usize);
// Capture result before decrementing to ensure decrement happens on error too
let result = inner.call(req).await;
// Always decrement, regardless of success or failure
let active = ACTIVE_HTTP_CONNECTIONS.fetch_sub(1, Ordering::Relaxed) - 1;
SmgMetrics::set_http_connections_active(active as usize);
Metrics::set_http_connections_active(active as usize);
let response = result?;
let duration = start.elapsed();
let status_class = status_to_class(response.status().as_u16());
SmgMetrics::record_http_request(&method, &path, status_class);
SmgMetrics::record_http_duration(&method, &path, duration);
Metrics::record_http_request(&method, &path, status_class);
Metrics::record_http_duration(&method, &path, duration);
Ok(response)
})
sgl-model-gateway/src/observability/metrics.rs
+6 -644
View File
@@ -24,197 +24,6 @@ impl Default for PrometheusConfig {
}
pub fn init_metrics() {
describe_counter!(
"sgl_router_requests_total",
"Total number of requests by route and method"
);
describe_histogram!(
"sgl_router_request_duration_seconds",
"Request duration in seconds"
);
describe_counter!(
"sgl_router_request_errors_total",
"Total number of request errors by route and error type"
);
describe_counter!(
"sgl_router_attempt_http_responses_total",
"Total number of upstream engine HTTP responses by status code"
);
describe_counter!(
"sgl_router_retries_total",
"Total number of request retries by route"
);
describe_histogram!(
"sgl_router_retry_backoff_duration_seconds",
"Backoff duration in seconds by attempt index"
);
describe_counter!(
"sgl_router_retries_exhausted_total",
"Total number of requests that exhausted retries by route"
);
describe_gauge!(
"sgl_router_cb_state",
"Circuit breaker state per worker (0=closed, 1=open, 2=half_open)"
);
describe_counter!(
"sgl_router_cb_state_transitions_total",
"Total number of circuit breaker state transitions by worker"
);
describe_counter!(
"sgl_router_cb_outcomes_total",
"Total number of circuit breaker outcomes by worker and outcome type (success/failure)"
);
describe_gauge!(
"sgl_router_cb_consecutive_failures",
"Current consecutive failure count per worker circuit breaker"
);
describe_gauge!(
"sgl_router_cb_consecutive_successes",
"Current consecutive success count per worker circuit breaker"
);
describe_counter!(
"sgl_router_discovery_watcher_errors_total",
"Total number of Kubernetes watcher errors"
);
describe_counter!(
"sgl_router_discovery_watcher_restarts_total",
"Total number of Kubernetes watcher restarts"
);
describe_gauge!(
"sgl_router_active_workers",
"Number of currently active workers"
);
describe_gauge!(
"sgl_router_worker_health",
"Worker health status (1=healthy, 0=unhealthy)"
);
describe_counter!(
"sgl_router_processed_requests_total",
"Total requests processed by each worker"
);
describe_gauge!(
"sgl_router_job_queue_depth",
"Current number of pending jobs in the queue"
);
describe_histogram!(
"sgl_router_job_duration_seconds",
"Job processing duration in seconds by job type"
);
describe_counter!(
"sgl_router_job_success_total",
"Total successful job completions by job type"
);
describe_counter!(
"sgl_router_job_failure_total",
"Total failed job completions by job type"
);
describe_counter!(
"sgl_router_job_queue_full_total",
"Total number of jobs rejected due to queue full"
);
describe_counter!(
"sgl_router_job_shutdown_rejected_total",
"Total number of jobs rejected due to shutdown"
);
describe_counter!(
"sgl_router_policy_decisions_total",
"Total routing policy decisions by policy and worker"
);
describe_counter!("sgl_router_cache_hits_total", "Total cache hits");
describe_counter!("sgl_router_cache_misses_total", "Total cache misses");
describe_gauge!(
"sgl_router_tree_size",
"Current tree size for cache-aware routing"
);
describe_counter!(
"sgl_router_load_balancing_events_total",
"Total load balancing trigger events"
);
describe_gauge!("sgl_router_max_load", "Maximum worker load");
describe_gauge!("sgl_router_min_load", "Minimum worker load");
describe_counter!("sgl_router_pd_requests_total", "Total PD requests by route");
describe_counter!(
"sgl_router_pd_prefill_requests_total",
"Total prefill requests per worker"
);
describe_counter!(
"sgl_router_pd_decode_requests_total",
"Total decode requests per worker"
);
describe_counter!(
"sgl_router_pd_errors_total",
"Total PD errors by error type"
);
describe_counter!(
"sgl_router_pd_prefill_errors_total",
"Total prefill server errors"
);
describe_counter!(
"sgl_router_pd_decode_errors_total",
"Total decode server errors"
);
describe_counter!(
"sgl_router_pd_stream_errors_total",
"Total streaming errors per worker"
);
describe_histogram!(
"sgl_router_pd_request_duration_seconds",
"PD request duration by route"
);
describe_counter!(
"sgl_router_discovery_updates_total",
"Total service discovery update events"
);
describe_gauge!(
"sgl_router_discovery_workers_added",
"Number of workers added in last discovery update"
);
describe_gauge!(
"sgl_router_discovery_workers_removed",
"Number of workers removed in last discovery update"
);
describe_histogram!(
"sgl_router_generate_duration_seconds",
"Generate request duration"
);
describe_counter!("sgl_router_embeddings_total", "Total embedding requests");
describe_histogram!(
"sgl_router_embeddings_duration_seconds",
"Embedding request duration"
);
describe_counter!(
"sgl_router_embeddings_errors_total",
"Embedding request errors"
);
describe_gauge!("sgl_router_embeddings_queue_size", "Embedding queue size");
describe_gauge!(
"sgl_router_running_requests",
"Number of running requests per worker"
);
describe_counter!(
"sgl_router_http_requests_total",
"Total number of HTTP requests"
);
describe_counter!(
"sgl_router_http_responses_total",
"Total number of HTTP responses by status code and error code"
);
// ========================================================================
// SMG Metrics (new layered architecture)
// ========================================================================
// Layer 1: HTTP metrics
describe_counter!(
"smg_http_requests_total",
@@ -414,339 +223,8 @@ pub fn start_prometheus(config: PrometheusConfig) {
.expect("failed to install Prometheus metrics exporter");
}
pub struct RouterMetrics;
impl RouterMetrics {
pub fn record_request(route: &'static str) {
counter!("sgl_router_requests_total",
"route" => route
)
.increment(1);
}
pub fn record_request_duration(duration: Duration) {
histogram!("sgl_router_request_duration_seconds").record(duration.as_secs_f64());
}
pub fn record_request_error(route: &'static str, error_type: &'static str) {
counter!("sgl_router_request_errors_total",
"route" => route,
"error_type" => error_type
)
.increment(1);
}
// TODO unify metric names
pub fn record_attempt_http_response(route: &'static str, status_code: u16, error_code: &str) {
counter!("sgl_router_attempt_http_responses_total",
"route" => route,
"status_code" => status_code.to_string(),
"error_code" => error_code.to_string()
)
.increment(1);
}
pub fn record_retry(route: &'static str) {
counter!("sgl_router_retries_total",
"route" => route
)
.increment(1);
}
pub fn record_retry_backoff_duration(duration: Duration, attempt: u32) {
histogram!("sgl_router_retry_backoff_duration_seconds",
"attempt" => attempt.to_string()
)
.record(duration.as_secs_f64());
}
pub fn record_retries_exhausted(route: &'static str) {
counter!("sgl_router_retries_exhausted_total",
"route" => route
)
.increment(1);
}
pub fn set_worker_health(worker_url: &str, healthy: bool) {
gauge!("sgl_router_worker_health",
"worker" => worker_url.to_string()
)
.set(if healthy { 1.0 } else { 0.0 });
}
pub fn set_active_workers(count: usize) {
gauge!("sgl_router_active_workers").set(count as f64);
}
pub fn record_processed_request(worker_url: &str) {
counter!("sgl_router_processed_requests_total",
"worker" => worker_url.to_string()
)
.increment(1);
}
pub fn record_policy_decision(policy: &'static str, worker: &str) {
counter!("sgl_router_policy_decisions_total",
"policy" => policy,
"worker" => worker.to_string()
)
.increment(1);
}
pub fn record_cache_hit() {
counter!("sgl_router_cache_hits_total").increment(1);
}
pub fn record_cache_miss() {
counter!("sgl_router_cache_misses_total").increment(1);
}
pub fn set_tree_size(worker: &str, size: usize) {
gauge!("sgl_router_tree_size",
"worker" => worker.to_string()
)
.set(size as f64);
}
pub fn record_load_balancing_event() {
counter!("sgl_router_load_balancing_events_total").increment(1);
}
pub fn set_load_range(max_load: usize, min_load: usize) {
gauge!("sgl_router_max_load").set(max_load as f64);
gauge!("sgl_router_min_load").set(min_load as f64);
}
pub fn record_pd_request(route: &'static str) {
counter!("sgl_router_pd_requests_total",
"route" => route
)
.increment(1);
}
pub fn record_pd_request_duration(route: &'static str, duration: Duration) {
histogram!("sgl_router_pd_request_duration_seconds",
"route" => route
)
.record(duration.as_secs_f64());
}
pub fn record_pd_prefill_request(worker: &str) {
counter!("sgl_router_pd_prefill_requests_total",
"worker" => worker.to_string()
)
.increment(1);
}
pub fn record_pd_decode_request(worker: &str) {
counter!("sgl_router_pd_decode_requests_total",
"worker" => worker.to_string()
)
.increment(1);
}
pub fn record_pd_error(error_type: &'static str) {
counter!("sgl_router_pd_errors_total",
"error_type" => error_type
)
.increment(1);
}
pub fn record_pd_prefill_error(worker: &str) {
counter!("sgl_router_pd_prefill_errors_total",
"worker" => worker.to_string()
)
.increment(1);
}
pub fn record_pd_decode_error(worker: &str) {
counter!("sgl_router_pd_decode_errors_total",
"worker" => worker.to_string()
)
.increment(1);
}
pub fn record_pd_stream_error(worker: &str) {
counter!("sgl_router_pd_stream_errors_total",
"worker" => worker.to_string()
)
.increment(1);
}
pub fn record_discovery_update(added: usize, removed: usize) {
counter!("sgl_router_discovery_updates_total").increment(1);
gauge!("sgl_router_discovery_workers_added").set(added as f64);
gauge!("sgl_router_discovery_workers_removed").set(removed as f64);
}
pub fn record_generate_duration(duration: Duration) {
histogram!("sgl_router_generate_duration_seconds").record(duration.as_secs_f64());
}
pub fn record_embeddings_request() {
counter!("sgl_router_embeddings_total").increment(1);
}
pub fn record_embeddings_duration(duration: Duration) {
histogram!("sgl_router_embeddings_duration_seconds").record(duration.as_secs_f64());
}
pub fn record_embeddings_error(error_type: &str) {
counter!(
"sgl_router_embeddings_errors_total",
"error_type" => error_type.to_string()
)
.increment(1);
}
pub fn set_embeddings_queue_size(size: usize) {
gauge!("sgl_router_embeddings_queue_size").set(size as f64);
}
pub fn record_classify_request() {
counter!("sgl_router_classify_total").increment(1);
}
pub fn record_classify_duration(duration: Duration) {
histogram!("sgl_router_classify_duration_seconds").record(duration.as_secs_f64());
}
pub fn record_classify_error(error_type: &str) {
counter!(
"sgl_router_classify_errors_total",
"error_type" => error_type.to_string()
)
.increment(1);
}
pub fn set_classify_queue_size(size: usize) {
gauge!("sgl_router_classify_queue_size").set(size as f64);
}
pub fn set_running_requests(worker: &str, count: usize) {
gauge!("sgl_router_running_requests",
"worker" => worker.to_string()
)
.set(count as f64);
}
pub fn set_cb_state(worker: &str, state_code: u8) {
gauge!("sgl_router_cb_state",
"worker" => worker.to_string()
)
.set(state_code as f64);
}
pub fn record_cb_state_transition(worker: &str, from: &'static str, to: &'static str) {
counter!("sgl_router_cb_state_transitions_total",
"worker" => worker.to_string(),
"from" => from,
"to" => to
)
.increment(1);
}
pub fn record_cb_outcome(worker: &str, outcome: &'static str) {
counter!("sgl_router_cb_outcomes_total",
"worker" => worker.to_string(),
"outcome" => outcome
)
.increment(1);
}
pub fn set_cb_consecutive_failures(worker: &str, count: u32) {
gauge!("sgl_router_cb_consecutive_failures",
"worker" => worker.to_string()
)
.set(count as f64);
}
pub fn set_cb_consecutive_successes(worker: &str, count: u32) {
gauge!("sgl_router_cb_consecutive_successes",
"worker" => worker.to_string()
)
.set(count as f64);
}
pub fn record_discovery_watcher_error() {
counter!("sgl_router_discovery_watcher_errors_total").increment(1);
}
pub fn record_discovery_watcher_restart() {
counter!("sgl_router_discovery_watcher_restarts_total").increment(1);
}
// TODO delete the metrics (instead of setting them to zero)
pub fn remove_worker_metrics(worker_url: &str) {
gauge!("sgl_router_cb_consecutive_failures","worker" => worker_url.to_string()).set(0.0);
gauge!("sgl_router_cb_consecutive_successes","worker" => worker_url.to_string()).set(0.0);
gauge!("sgl_router_running_requests","worker" => worker_url.to_string()).set(0.0);
gauge!("sgl_router_tree_size","worker" => worker_url.to_string()).set(0.0);
// Zero for these metrics have special valid meaning, thus we set to -1 temporarily
// (and will remove them completely after https://github.com/metrics-rs/metrics/issues/653)
gauge!("sgl_router_cb_state","worker" => worker_url.to_string()).set(-1.0);
gauge!("sgl_router_worker_health","worker" => worker_url.to_string()).set(-1.0);
}
pub fn set_job_queue_depth(depth: usize) {
gauge!("sgl_router_job_queue_depth").set(depth as f64);
}
pub fn record_job_duration(job_type: &'static str, duration: Duration) {
histogram!("sgl_router_job_duration_seconds",
"job_type" => job_type
)
.record(duration.as_secs_f64());
}
pub fn record_job_success(job_type: &'static str) {
counter!("sgl_router_job_success_total",
"job_type" => job_type
)
.increment(1);
}
pub fn record_job_failure(job_type: &'static str) {
counter!("sgl_router_job_failure_total",
"job_type" => job_type
)
.increment(1);
}
pub fn record_job_queue_full() {
counter!("sgl_router_job_queue_full_total").increment(1);
}
pub fn record_job_shutdown_rejected() {
counter!("sgl_router_job_shutdown_rejected_total").increment(1);
}
// This is different from the following:
// * sgl_router_requests_total: bump when a request is handled and response is to be returned, thus very different from this.
// * sgl_router_processed_requests_total: bump when routing decision is made.
// Here we want a metric to directly reflect user's experience ("I am sending a request")
// when viewing the router as a blackbox, and is bumped immediately when the request arrives.
// TODO: add route name
pub fn record_http_request() {
counter!("sgl_router_http_requests_total").increment(1);
}
pub fn record_http_status_code(status_code: u16, error_code: &str) {
counter!("sgl_router_http_responses_total",
"status_code" => status_code.to_string(),
"error_code" => error_code.to_string()
)
.increment(1);
}
}
// ============================================================================
// SMG Metrics - New layered architecture
// ============================================================================
/// Label constants for consistent metric labeling
pub mod smg_labels {
pub mod metrics_labels {
// Router types
pub const ROUTER_OPENAI: &str = "openai";
pub const ROUTER_HTTP: &str = "http";
@@ -808,12 +286,6 @@ pub mod smg_labels {
pub const REGISTRATION_SUCCESS: &str = "success";
pub const REGISTRATION_FAILED: &str = "failed";
pub const REGISTRATION_DUPLICATE: &str = "duplicate";
// Deregistration reasons
pub const DEREGISTRATION_HEALTH_CHECK_FAILED: &str = "health_check_failed";
pub const DEREGISTRATION_TIMEOUT: &str = "timeout";
pub const DEREGISTRATION_MANUAL: &str = "manual";
pub const DEREGISTRATION_SHUTDOWN: &str = "shutdown";
pub const DEREGISTRATION_POD_DELETED: &str = "pod_deleted";
// Rate limit results
@@ -835,19 +307,10 @@ pub mod smg_labels {
pub const ERROR_BACKEND: &str = "backend_error";
pub const ERROR_VALIDATION: &str = "validation_error";
pub const ERROR_INTERNAL: &str = "internal_error";
// Pipeline stages (gRPC router)
pub const STAGE_PREPARATION: &str = "preparation";
pub const STAGE_WORKER_SELECTION: &str = "worker_selection";
pub const STAGE_CLIENT_ACQUISITION: &str = "client_acquisition";
pub const STAGE_REQUEST_BUILDING: &str = "request_building";
pub const STAGE_DISPATCH_METADATA: &str = "dispatch_metadata";
pub const STAGE_REQUEST_EXECUTION: &str = "request_execution";
pub const STAGE_RESPONSE_PROCESSING: &str = "response_processing";
}
/// SMG Metrics helper struct for the new layered metrics architecture
pub struct SmgMetrics;
pub struct Metrics;
/// Parameters for recording streaming metrics.
pub struct StreamingMetricsParams<'a> {
@@ -869,7 +332,7 @@ pub struct StreamingMetricsParams<'a> {
pub output_tokens: u64,
}
impl SmgMetrics {
impl Metrics {
/// Record an HTTP request
pub fn record_http_request(method: &str, path: &str, status_class: &str) {
counter!(
@@ -1151,7 +614,7 @@ impl SmgMetrics {
"backend_type" => backend_type,
"model" => model.clone(),
"endpoint" => endpoint,
"token_type" => smg_labels::TOKEN_INPUT
"token_type" => metrics_labels::TOKEN_INPUT
)
.increment(input);
}
@@ -1163,7 +626,7 @@ impl SmgMetrics {
"backend_type" => backend_type,
"model" => model,
"endpoint" => endpoint,
"token_type" => smg_labels::TOKEN_OUTPUT
"token_type" => metrics_labels::TOKEN_OUTPUT
)
.increment(output_tokens);
}
@@ -1628,7 +1091,7 @@ mod tests {
let _matching_metrics = [
"request_duration_seconds",
"response_duration_seconds",
"sgl_router_request_duration_seconds",
"smg_request_duration_seconds",
];
let _non_matching_metrics = ["duration_total", "duration_seconds_total", "other_metric"];
@@ -1680,37 +1143,6 @@ mod tests {
}
}
#[test]
fn test_metrics_static_methods() {
RouterMetrics::record_request("/generate");
RouterMetrics::record_request_duration(Duration::from_millis(100));
RouterMetrics::record_request_error("/generate", "timeout");
RouterMetrics::record_retry("/generate");
RouterMetrics::set_worker_health("http://worker1", true);
RouterMetrics::record_processed_request("http://worker1");
RouterMetrics::record_policy_decision("random", "http://worker1");
RouterMetrics::record_cache_hit();
RouterMetrics::record_cache_miss();
RouterMetrics::set_tree_size("http://worker1", 1000);
RouterMetrics::record_load_balancing_event();
RouterMetrics::set_load_range(20, 5);
RouterMetrics::record_pd_request("/v1/chat/completions");
RouterMetrics::record_pd_request_duration("/v1/chat/completions", Duration::from_secs(1));
RouterMetrics::record_pd_prefill_request("http://prefill1");
RouterMetrics::record_pd_decode_request("http://decode1");
RouterMetrics::record_pd_error("invalid_request");
RouterMetrics::record_pd_prefill_error("http://prefill1");
RouterMetrics::record_pd_decode_error("http://decode1");
RouterMetrics::record_pd_stream_error("http://decode1");
RouterMetrics::record_discovery_update(3, 1);
RouterMetrics::record_generate_duration(Duration::from_secs(2));
RouterMetrics::set_running_requests("http://worker1", 15);
}
#[test]
fn test_port_already_in_use() {
let port = 29123;
@@ -1739,74 +1171,4 @@ mod tests {
assert_eq!(socket_addr.to_string(), "127.0.0.1:29000");
}
#[test]
fn test_concurrent_metric_updates() {
use std::{
sync::{
atomic::{AtomicBool, Ordering},
Arc,
},
thread,
};
let done = Arc::new(AtomicBool::new(false));
let mut handles = vec![];
for i in 0..3 {
let done_clone = done.clone();
let handle = thread::spawn(move || {
let worker = format!("http://worker{}", i);
while !done_clone.load(Ordering::Relaxed) {
RouterMetrics::record_processed_request(&worker);
thread::sleep(Duration::from_millis(1));
}
});
handles.push(handle);
}
thread::sleep(Duration::from_millis(10));
done.store(true, Ordering::Relaxed);
for handle in handles {
handle.join().unwrap();
}
}
#[test]
fn test_empty_string_metrics() {
RouterMetrics::record_request("");
RouterMetrics::set_worker_health("", true);
RouterMetrics::record_policy_decision("", "");
}
#[test]
fn test_very_long_metric_labels() {
let long_label = "a".repeat(1000);
RouterMetrics::record_request("/very_long_test_route");
RouterMetrics::set_worker_health(&long_label, false);
}
#[test]
fn test_special_characters_in_labels() {
let special_labels = [
"test/with/slashes",
"test-with-dashes",
"test_with_underscores",
"test.with.dots",
"test:with:colons",
];
for label in special_labels {
RouterMetrics::record_request(label);
RouterMetrics::set_worker_health(label, true);
}
}
#[test]
fn test_extreme_metric_values() {
RouterMetrics::record_request_duration(Duration::from_nanos(1));
RouterMetrics::record_request_duration(Duration::from_secs(86400));
}
}
sgl-model-gateway/src/policies/cache_aware.rs
+1 -15
View File
@@ -73,7 +73,7 @@ use rand::Rng;
use tracing::debug;
use super::{get_healthy_worker_indices, tree::Tree, CacheAwareConfig, LoadBalancingPolicy};
use crate::{core::Worker, observability::metrics::RouterMetrics};
use crate::core::Worker;
/// Cache-aware routing policy
///
@@ -135,11 +135,6 @@ impl CacheAwarePolicy {
let tree = tree_ref.value();
tree.evict_tenant_by_size(max_tree_size);
// Update tree size metrics per worker (tenant)
for entry in tree.tenant_char_count.iter() {
RouterMetrics::set_tree_size(entry.key(), *entry.value());
}
debug!(
"Cache eviction completed for model {}, max_size: {}",
model_id, max_tree_size
@@ -274,9 +269,6 @@ impl CacheAwarePolicy {
max_load, min_load, worker_loads
);
RouterMetrics::record_load_balancing_event();
RouterMetrics::set_load_range(max_load, min_load);
// Use shortest queue when imbalanced
let min_load_idx = healthy_indices
.iter()
@@ -302,8 +294,6 @@ impl CacheAwarePolicy {
// Increment processed counter
workers[min_load_idx].increment_processed();
RouterMetrics::record_processed_request(workers[min_load_idx].url());
RouterMetrics::record_policy_decision(self.name(), workers[min_load_idx].url());
Some(min_load_idx)
}
@@ -369,10 +359,8 @@ impl LoadBalancingPolicy for CacheAwarePolicy {
};
let selected_url = if match_rate > self.config.cache_threshold {
RouterMetrics::record_cache_hit();
matched_worker.to_string()
} else {
RouterMetrics::record_cache_miss();
let min_load_idx = *healthy_indices
.iter()
.min_by_key(|&&idx| workers[idx].load())?;
@@ -388,8 +376,6 @@ impl LoadBalancingPolicy for CacheAwarePolicy {
// Increment processed counter
workers[selected_idx].increment_processed();
RouterMetrics::record_processed_request(&selected_url);
RouterMetrics::record_policy_decision(self.name(), &selected_url);
return Some(selected_idx);
}
sgl-model-gateway/src/policies/power_of_two.rs
+1 -3
View File
@@ -9,7 +9,7 @@ use rand::Rng;
use tracing::debug;
use super::{get_healthy_worker_indices, LoadBalancingPolicy};
use crate::{core::Worker, observability::metrics::RouterMetrics};
use crate::core::Worker;
/// Power-of-two choices policy
///
@@ -100,8 +100,6 @@ impl LoadBalancingPolicy for PowerOfTwoPolicy {
// Increment processed counter
workers[selected_idx].increment_processed();
RouterMetrics::record_processed_request(workers[selected_idx].url());
RouterMetrics::record_policy_decision(self.name(), workers[selected_idx].url());
Some(selected_idx)
}
sgl-model-gateway/src/policies/random.rs
+1 -4
View File
@@ -5,7 +5,7 @@ use std::sync::Arc;
use rand::Rng;
use super::{get_healthy_worker_indices, LoadBalancingPolicy};
use crate::{core::Worker, observability::metrics::RouterMetrics};
use crate::core::Worker;
/// Random selection policy
///
@@ -33,10 +33,7 @@ impl LoadBalancingPolicy for RandomPolicy {
let mut rng = rand::rng();
let random_idx = rng.random_range(0..healthy_indices.len());
let worker = workers[healthy_indices[random_idx]].url();
RouterMetrics::record_processed_request(worker);
RouterMetrics::record_policy_decision(self.name(), worker);
Some(healthy_indices[random_idx])
}
sgl-model-gateway/src/policies/round_robin.rs
+1 -4
View File
@@ -6,7 +6,7 @@ use std::sync::{
};
use super::{get_healthy_worker_indices, LoadBalancingPolicy};
use crate::{core::Worker, observability::metrics::RouterMetrics};
use crate::core::Worker;
/// Round-robin selection policy
///
@@ -39,10 +39,7 @@ impl LoadBalancingPolicy for RoundRobinPolicy {
// Get and increment counter atomically
let count = self.counter.fetch_add(1, Ordering::Relaxed);
let selected_idx = count % healthy_indices.len();
let worker = workers[healthy_indices[selected_idx]].url();
RouterMetrics::record_processed_request(worker);
RouterMetrics::record_policy_decision(self.name(), worker);
Some(healthy_indices[selected_idx])
}
sgl-model-gateway/src/routers/grpc/common/stages/worker_selection.rs
+10 -10
View File
@@ -9,7 +9,7 @@ use tracing::{error, warn};
use super::PipelineStage;
use crate::{
core::{ConnectionMode, Worker, WorkerRegistry, WorkerType},
observability::metrics::{smg_labels, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
policies::PolicyRegistry,
routers::{
error,
@@ -150,9 +150,9 @@ impl WorkerSelectionStage {
let selected = available[idx].clone();
// Record worker selection metric
SmgMetrics::record_worker_selection(
smg_labels::WORKER_REGULAR,
smg_labels::CONNECTION_GRPC,
Metrics::record_worker_selection(
metrics_labels::WORKER_REGULAR,
metrics_labels::CONNECTION_GRPC,
model_id.unwrap_or("default"),
policy.name(),
);
@@ -210,15 +210,15 @@ impl WorkerSelectionStage {
let policy_name = policy.name();
// Record worker selection metrics for both prefill and decode
SmgMetrics::record_worker_selection(
smg_labels::WORKER_PREFILL,
smg_labels::CONNECTION_GRPC,
Metrics::record_worker_selection(
metrics_labels::WORKER_PREFILL,
metrics_labels::CONNECTION_GRPC,
model,
policy_name,
);
SmgMetrics::record_worker_selection(
smg_labels::WORKER_DECODE,
smg_labels::CONNECTION_GRPC,
Metrics::record_worker_selection(
metrics_labels::WORKER_DECODE,
metrics_labels::CONNECTION_GRPC,
model,
policy_name,
);
sgl-model-gateway/src/routers/grpc/harmony/responses.rs
+10 -10
View File
@@ -45,7 +45,7 @@ use uuid::Uuid;
use crate::{
data_connector::{ConversationItemStorage, ConversationStorage, ResponseId, ResponseStorage},
mcp::{self, McpManager},
observability::metrics::{smg_labels, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
protocols::{
common::{Function, ToolCall, ToolChoice, ToolChoiceValue, Usage},
responses::{
@@ -326,7 +326,7 @@ async fn execute_with_mcp_loop(
iteration_count += 1;
// Record tool loop iteration metric
SmgMetrics::record_mcp_tool_iteration(&current_request.model);
Metrics::record_mcp_tool_iteration(&current_request.model);
// Safety check: prevent infinite loops
if iteration_count > MAX_TOOL_ITERATIONS {
@@ -770,7 +770,7 @@ async fn execute_mcp_tool_loop_streaming(
iteration_count += 1;
// Record tool loop iteration metric
SmgMetrics::record_mcp_tool_iteration(&current_request.model);
Metrics::record_mcp_tool_iteration(&current_request.model);
// Safety check: prevent infinite loops
if iteration_count > MAX_TOOL_ITERATIONS {
@@ -1253,18 +1253,18 @@ async fn execute_mcp_tools(
);
// Record MCP tool metrics
SmgMetrics::record_mcp_tool_duration(
Metrics::record_mcp_tool_duration(
model_id,
&tool_call.function.name,
tool_duration,
);
SmgMetrics::record_mcp_tool_call(
Metrics::record_mcp_tool_call(
model_id,
&tool_call.function.name,
if is_error {
smg_labels::RESULT_ERROR
metrics_labels::RESULT_ERROR
} else {
smg_labels::RESULT_SUCCESS
metrics_labels::RESULT_SUCCESS
},
);
@@ -1301,15 +1301,15 @@ async fn execute_mcp_tools(
);
// Record MCP tool metrics
SmgMetrics::record_mcp_tool_duration(
Metrics::record_mcp_tool_duration(
model_id,
&tool_call.function.name,
tool_duration,
);
SmgMetrics::record_mcp_tool_call(
Metrics::record_mcp_tool_call(
model_id,
&tool_call.function.name,
smg_labels::RESULT_ERROR,
metrics_labels::RESULT_ERROR,
);
// Return error result to model (let it handle gracefully)
sgl-model-gateway/src/routers/grpc/harmony/streaming.rs
+9 -9
View File
@@ -20,7 +20,7 @@ use super::{
};
use crate::{
grpc_client::sglang_proto::generate_complete::MatchedStop::{MatchedStopStr, MatchedTokenId},
observability::metrics::{smg_labels, SmgMetrics, StreamingMetricsParams},
observability::metrics::{metrics_labels, Metrics, StreamingMetricsParams},
protocols::{
chat::{
ChatCompletionRequest, ChatCompletionStreamResponse, ChatMessageDelta, ChatStreamChoice,
@@ -315,11 +315,11 @@ impl HarmonyStreamingProcessor {
grpc_stream.mark_completed();
// Record streaming metrics
SmgMetrics::record_streaming_metrics(StreamingMetricsParams {
router_type: smg_labels::ROUTER_GRPC,
backend_type: smg_labels::BACKEND_HARMONY,
Metrics::record_streaming_metrics(StreamingMetricsParams {
router_type: metrics_labels::ROUTER_GRPC,
backend_type: metrics_labels::BACKEND_HARMONY,
model_id: &original_request.model,
endpoint: smg_labels::ENDPOINT_CHAT,
endpoint: metrics_labels::ENDPOINT_CHAT,
ttft: first_token_time.map(|t| t.duration_since(start_time)),
generation_duration: start_time.elapsed(),
input_tokens: Some(total_prompt as u64),
@@ -474,11 +474,11 @@ impl HarmonyStreamingProcessor {
}
// Record streaming metrics
SmgMetrics::record_streaming_metrics(StreamingMetricsParams {
router_type: smg_labels::ROUTER_GRPC,
backend_type: smg_labels::BACKEND_HARMONY,
Metrics::record_streaming_metrics(StreamingMetricsParams {
router_type: metrics_labels::ROUTER_GRPC,
backend_type: metrics_labels::BACKEND_HARMONY,
model_id: &original_request.model,
endpoint: smg_labels::ENDPOINT_CHAT,
endpoint: metrics_labels::ENDPOINT_CHAT,
ttft: first_token_time.map(|t| t.duration_since(start_time)),
generation_duration: start_time.elapsed(),
input_tokens: Some(total_prompt as u64),
sgl-model-gateway/src/routers/grpc/pipeline.rs
+59 -59
View File
@@ -17,7 +17,7 @@ use super::{
};
use crate::{
core::WorkerRegistry,
observability::metrics::{smg_labels, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
policies::PolicyRegistry,
protocols::{
chat::{ChatCompletionRequest, ChatCompletionResponse},
@@ -65,7 +65,7 @@ impl RequestPipeline {
reasoning_parser_factory,
configured_tool_parser,
configured_reasoning_parser,
smg_labels::BACKEND_REGULAR,
metrics_labels::BACKEND_REGULAR,
));
let stages: Vec<Box<dyn PipelineStage>> = vec![
@@ -84,7 +84,7 @@ impl RequestPipeline {
Self {
stages: Arc::new(stages),
backend_type: smg_labels::BACKEND_REGULAR,
backend_type: metrics_labels::BACKEND_REGULAR,
}
}
@@ -114,7 +114,7 @@ impl RequestPipeline {
Self {
stages: Arc::new(stages),
backend_type: smg_labels::BACKEND_REGULAR,
backend_type: metrics_labels::BACKEND_REGULAR,
}
}
@@ -144,7 +144,7 @@ impl RequestPipeline {
Self {
stages: Arc::new(stages),
backend_type: smg_labels::BACKEND_PD,
backend_type: metrics_labels::BACKEND_PD,
}
}
@@ -172,7 +172,7 @@ impl RequestPipeline {
reasoning_parser_factory,
configured_tool_parser,
configured_reasoning_parser,
smg_labels::BACKEND_PD,
metrics_labels::BACKEND_PD,
));
let stages: Vec<Box<dyn PipelineStage>> = vec![
@@ -191,7 +191,7 @@ impl RequestPipeline {
Self {
stages: Arc::new(stages),
backend_type: smg_labels::BACKEND_PD,
backend_type: metrics_labels::BACKEND_PD,
}
}
@@ -209,12 +209,12 @@ impl RequestPipeline {
let streaming = request.stream;
// Record request start
SmgMetrics::record_router_request(
smg_labels::ROUTER_GRPC,
Metrics::record_router_request(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
&request_for_metrics.model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
streaming,
);
@@ -224,24 +224,24 @@ impl RequestPipeline {
match stage.execute(&mut ctx).await {
Ok(Some(response)) => {
// Stage completed with streaming response - record success and return
SmgMetrics::record_router_duration(
smg_labels::ROUTER_GRPC,
Metrics::record_router_duration(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
&request_for_metrics.model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
start.elapsed(),
);
return response;
}
Ok(None) => continue,
Err(response) => {
SmgMetrics::record_router_error(
smg_labels::ROUTER_GRPC,
Metrics::record_router_error(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
&request_for_metrics.model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
error_type_from_status(response.status()),
);
error!(
@@ -256,12 +256,12 @@ impl RequestPipeline {
match ctx.state.response.final_response {
Some(FinalResponse::Chat(response)) => {
SmgMetrics::record_router_duration(
smg_labels::ROUTER_GRPC,
Metrics::record_router_duration(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
&request_for_metrics.model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
start.elapsed(),
);
axum::Json(response).into_response()
@@ -271,13 +271,13 @@ impl RequestPipeline {
function = "execute_chat",
"Wrong response type: expected Chat, got Generate"
);
SmgMetrics::record_router_error(
smg_labels::ROUTER_GRPC,
Metrics::record_router_error(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
&request_for_metrics.model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_INTERNAL,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_INTERNAL,
);
error::internal_error("wrong_response_type", "Internal error: wrong response type")
}
@@ -286,13 +286,13 @@ impl RequestPipeline {
function = "execute_chat",
"No response produced by pipeline"
);
SmgMetrics::record_router_error(
smg_labels::ROUTER_GRPC,
Metrics::record_router_error(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
&request_for_metrics.model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_INTERNAL,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_INTERNAL,
);
error::internal_error("no_response_produced", "No response produced")
}
@@ -314,12 +314,12 @@ impl RequestPipeline {
let streaming = request.stream;
// Record request start
SmgMetrics::record_router_request(
smg_labels::ROUTER_GRPC,
Metrics::record_router_request(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
model_for_metrics.as_deref().unwrap_or("unknown"),
smg_labels::ENDPOINT_GENERATE,
metrics_labels::ENDPOINT_GENERATE,
streaming,
);
@@ -328,24 +328,24 @@ impl RequestPipeline {
for stage in self.stages.iter() {
match stage.execute(&mut ctx).await {
Ok(Some(response)) => {
SmgMetrics::record_router_duration(
smg_labels::ROUTER_GRPC,
Metrics::record_router_duration(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
model_for_metrics.as_deref().unwrap_or("unknown"),
smg_labels::ENDPOINT_GENERATE,
metrics_labels::ENDPOINT_GENERATE,
start.elapsed(),
);
return response;
}
Ok(None) => continue,
Err(response) => {
SmgMetrics::record_router_error(
smg_labels::ROUTER_GRPC,
Metrics::record_router_error(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
model_for_metrics.as_deref().unwrap_or("unknown"),
smg_labels::ENDPOINT_GENERATE,
metrics_labels::ENDPOINT_GENERATE,
error_type_from_status(response.status()),
);
error!(
@@ -360,12 +360,12 @@ impl RequestPipeline {
match ctx.state.response.final_response {
Some(FinalResponse::Generate(response)) => {
SmgMetrics::record_router_duration(
smg_labels::ROUTER_GRPC,
Metrics::record_router_duration(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
model_for_metrics.as_deref().unwrap_or("unknown"),
smg_labels::ENDPOINT_GENERATE,
metrics_labels::ENDPOINT_GENERATE,
start.elapsed(),
);
axum::Json(response).into_response()
@@ -375,13 +375,13 @@ impl RequestPipeline {
function = "execute_generate",
"Wrong response type: expected Generate, got Chat"
);
SmgMetrics::record_router_error(
smg_labels::ROUTER_GRPC,
Metrics::record_router_error(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
model_for_metrics.as_deref().unwrap_or("unknown"),
smg_labels::ENDPOINT_GENERATE,
smg_labels::ERROR_INTERNAL,
metrics_labels::ENDPOINT_GENERATE,
metrics_labels::ERROR_INTERNAL,
);
error::internal_error("wrong_response_type", "Internal error: wrong response type")
}
@@ -390,13 +390,13 @@ impl RequestPipeline {
function = "execute_generate",
"No response produced by pipeline"
);
SmgMetrics::record_router_error(
smg_labels::ROUTER_GRPC,
Metrics::record_router_error(
metrics_labels::ROUTER_GRPC,
self.backend_type,
smg_labels::CONNECTION_GRPC,
metrics_labels::CONNECTION_GRPC,
model_for_metrics.as_deref().unwrap_or("unknown"),
smg_labels::ENDPOINT_GENERATE,
smg_labels::ERROR_INTERNAL,
metrics_labels::ENDPOINT_GENERATE,
metrics_labels::ERROR_INTERNAL,
);
error::internal_error("no_response_produced", "No response produced")
}
sgl-model-gateway/src/routers/grpc/regular/responses/tool_loop.rs
+11 -11
View File
@@ -22,7 +22,7 @@ use uuid::Uuid;
use super::conversions;
use crate::{
mcp::{self, McpManager},
observability::metrics::{smg_labels, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
protocols::{
chat::{
ChatChoice, ChatCompletionMessage, ChatCompletionRequest, ChatCompletionResponse,
@@ -285,7 +285,7 @@ pub(super) async fn execute_tool_loop(
state.iteration += 1;
// Record tool loop iteration metric
SmgMetrics::record_mcp_tool_iteration(&current_request.model);
Metrics::record_mcp_tool_iteration(&current_request.model);
debug!(
"Tool loop iteration {}: found {} tool call(s)",
@@ -408,18 +408,18 @@ pub(super) async fn execute_tool_loop(
let tool_duration = tool_start.elapsed();
// Record MCP tool metrics
SmgMetrics::record_mcp_tool_duration(
Metrics::record_mcp_tool_duration(
&current_request.model,
&tool_name,
tool_duration,
);
SmgMetrics::record_mcp_tool_call(
Metrics::record_mcp_tool_call(
&current_request.model,
&tool_name,
if success {
smg_labels::RESULT_SUCCESS
metrics_labels::RESULT_SUCCESS
} else {
smg_labels::RESULT_ERROR
metrics_labels::RESULT_ERROR
},
);
@@ -665,7 +665,7 @@ async fn execute_tool_loop_streaming_internal(
state.iteration += 1;
// Record tool loop iteration metric
SmgMetrics::record_mcp_tool_iteration(&model);
Metrics::record_mcp_tool_iteration(&model);
if state.iteration > MAX_ITERATIONS {
return Err(format!(
@@ -928,14 +928,14 @@ async fn execute_tool_loop_streaming_internal(
let tool_duration = tool_start.elapsed();
// Record MCP tool metrics
SmgMetrics::record_mcp_tool_duration(&model, &tool_name, tool_duration);
SmgMetrics::record_mcp_tool_call(
Metrics::record_mcp_tool_duration(&model, &tool_name, tool_duration);
Metrics::record_mcp_tool_call(
&model,
&tool_name,
if success {
smg_labels::RESULT_SUCCESS
metrics_labels::RESULT_SUCCESS
} else {
smg_labels::RESULT_ERROR
metrics_labels::RESULT_ERROR
},
);
sgl-model-gateway/src/routers/grpc/regular/streaming.rs
+7 -7
View File
@@ -14,7 +14,7 @@ use tracing::{debug, error, warn};
use crate::{
grpc_client::sglang_proto::generate_complete::MatchedStop::{MatchedStopStr, MatchedTokenId},
observability::metrics::{smg_labels, SmgMetrics, StreamingMetricsParams},
observability::metrics::{metrics_labels, Metrics, StreamingMetricsParams},
protocols::{
chat::{ChatCompletionRequest, ChatCompletionStreamResponse},
common::{
@@ -573,11 +573,11 @@ impl StreamingProcessor {
// Record streaming metrics
let total_prompt: u32 = prompt_tokens.values().sum();
let total_completion: u32 = completion_tokens.values().sum();
SmgMetrics::record_streaming_metrics(StreamingMetricsParams {
router_type: smg_labels::ROUTER_GRPC,
Metrics::record_streaming_metrics(StreamingMetricsParams {
router_type: metrics_labels::ROUTER_GRPC,
backend_type: self.backend_type,
model_id: model,
endpoint: smg_labels::ENDPOINT_CHAT,
endpoint: metrics_labels::ENDPOINT_CHAT,
ttft: first_token_time.map(|t| t.duration_since(start_time)),
generation_duration: start_time.elapsed(),
input_tokens: Some(total_prompt as u64),
@@ -1015,11 +1015,11 @@ impl StreamingProcessor {
total_completion: u32,
ctx: &GenerateStreamContext,
) {
SmgMetrics::record_streaming_metrics(StreamingMetricsParams {
router_type: smg_labels::ROUTER_GRPC,
Metrics::record_streaming_metrics(StreamingMetricsParams {
router_type: metrics_labels::ROUTER_GRPC,
backend_type: ctx.backend_type,
model_id: &ctx.model,
endpoint: smg_labels::ENDPOINT_GENERATE,
endpoint: metrics_labels::ENDPOINT_GENERATE,
ttft: first_token_time.map(|t| t.duration_since(start_time)),
generation_duration: start_time.elapsed(),
input_tokens: None, // generate endpoint doesn't expose prompt tokens in streaming
sgl-model-gateway/src/routers/grpc/utils.rs
+11 -11
View File
@@ -16,7 +16,7 @@ use super::{
use crate::{
core::Worker,
grpc_client::sglang_proto::{InputLogProbs, OutputLogProbs},
observability::metrics::smg_labels,
observability::metrics::metrics_labels,
protocols::{
chat::{ChatCompletionRequest, ChatMessage},
common::{
@@ -966,11 +966,11 @@ pub fn parse_finish_reason(reason_str: &str, completion_tokens: i32) -> Generate
/// Map route path to endpoint label for metrics
pub fn route_to_endpoint(route: &str) -> &'static str {
match route {
"/v1/chat/completions" => smg_labels::ENDPOINT_CHAT,
"/generate" => smg_labels::ENDPOINT_GENERATE,
"/v1/completions" => smg_labels::ENDPOINT_COMPLETIONS,
"/v1/rerank" => smg_labels::ENDPOINT_RERANK,
"/v1/responses" => smg_labels::ENDPOINT_RESPONSES,
"/v1/chat/completions" => metrics_labels::ENDPOINT_CHAT,
"/generate" => metrics_labels::ENDPOINT_GENERATE,
"/v1/completions" => metrics_labels::ENDPOINT_COMPLETIONS,
"/v1/rerank" => metrics_labels::ENDPOINT_RERANK,
"/v1/responses" => metrics_labels::ENDPOINT_RESPONSES,
_ => "other",
}
}
@@ -978,11 +978,11 @@ pub fn route_to_endpoint(route: &str) -> &'static str {
/// Map HTTP status code to error type label for metrics
pub fn error_type_from_status(status: StatusCode) -> &'static str {
match status.as_u16() {
400 => smg_labels::ERROR_VALIDATION,
404 => smg_labels::ERROR_NO_WORKERS,
408 | 504 => smg_labels::ERROR_TIMEOUT,
500..=599 => smg_labels::ERROR_BACKEND,
_ => smg_labels::ERROR_INTERNAL,
400 => metrics_labels::ERROR_VALIDATION,
404 => metrics_labels::ERROR_NO_WORKERS,
408 | 504 => metrics_labels::ERROR_TIMEOUT,
500..=599 => metrics_labels::ERROR_BACKEND,
_ => metrics_labels::ERROR_INTERNAL,
}
}
sgl-model-gateway/src/routers/http/pd_router.rs
+31 -48
View File
@@ -22,7 +22,7 @@ use crate::{
},
observability::{
events::{self, Event},
metrics::{smg_labels, RouterMetrics, SmgMetrics},
metrics::{metrics_labels, Metrics},
otel_trace::inject_trace_context_http,
},
policies::{LoadBalancingPolicy, PolicyRegistry},
@@ -165,7 +165,6 @@ impl PDRouter {
fn handle_server_selection_error(error: String) -> Response {
error!("Failed to select PD pair error={}", error);
RouterMetrics::record_pd_error("server_selection");
error::service_unavailable(
"server_selection_failed",
format!("No available servers: {}", error),
@@ -283,10 +282,10 @@ impl PDRouter {
let endpoint = route_to_endpoint(route);
// Record request start (Layer 2)
SmgMetrics::record_router_request(
smg_labels::ROUTER_HTTP,
smg_labels::BACKEND_PD,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_request(
metrics_labels::ROUTER_HTTP,
metrics_labels::BACKEND_PD,
metrics_labels::CONNECTION_HTTP,
model,
endpoint,
context.is_stream,
@@ -308,7 +307,6 @@ impl PDRouter {
{
Ok(pair) => pair,
Err(e) => {
RouterMetrics::record_pd_error("server_selection");
return Self::handle_server_selection_error(e);
}
};
@@ -353,14 +351,14 @@ impl PDRouter {
// Record worker errors for server errors (5xx)
if status.is_server_error() {
let error_type = error_type_from_status(status);
SmgMetrics::record_worker_error(
smg_labels::WORKER_PREFILL,
smg_labels::CONNECTION_HTTP,
Metrics::record_worker_error(
metrics_labels::WORKER_PREFILL,
metrics_labels::CONNECTION_HTTP,
error_type,
);
SmgMetrics::record_worker_error(
smg_labels::WORKER_DECODE,
smg_labels::CONNECTION_HTTP,
Metrics::record_worker_error(
metrics_labels::WORKER_DECODE,
metrics_labels::CONNECTION_HTTP,
error_type,
);
}
@@ -371,17 +369,14 @@ impl PDRouter {
},
|res, _attempt| is_retryable_status(res.status()),
|delay, attempt| {
RouterMetrics::record_retry(route);
RouterMetrics::record_retry_backoff_duration(delay, attempt);
// Layer 3 worker metrics (PD mode uses both prefill and decode workers)
SmgMetrics::record_worker_retry(smg_labels::WORKER_PREFILL, endpoint);
SmgMetrics::record_worker_retry(smg_labels::WORKER_DECODE, endpoint);
SmgMetrics::record_worker_retry_backoff(attempt, delay);
Metrics::record_worker_retry(metrics_labels::WORKER_PREFILL, endpoint);
Metrics::record_worker_retry(metrics_labels::WORKER_DECODE, endpoint);
Metrics::record_worker_retry_backoff(attempt, delay);
},
|| {
RouterMetrics::record_retries_exhausted(route);
SmgMetrics::record_worker_retries_exhausted(smg_labels::WORKER_PREFILL, endpoint);
SmgMetrics::record_worker_retries_exhausted(smg_labels::WORKER_DECODE, endpoint);
Metrics::record_worker_retries_exhausted(metrics_labels::WORKER_PREFILL, endpoint);
Metrics::record_worker_retries_exhausted(metrics_labels::WORKER_DECODE, endpoint);
},
)
.await;
@@ -389,19 +384,19 @@ impl PDRouter {
// Record Layer 2 metrics
let duration = start_time.elapsed();
if response.status().is_success() {
SmgMetrics::record_router_duration(
smg_labels::ROUTER_HTTP,
smg_labels::BACKEND_PD,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_duration(
metrics_labels::ROUTER_HTTP,
metrics_labels::BACKEND_PD,
metrics_labels::CONNECTION_HTTP,
model,
endpoint,
duration,
);
} else if !is_retryable_status(response.status()) {
SmgMetrics::record_router_error(
smg_labels::ROUTER_HTTP,
smg_labels::BACKEND_PD,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_HTTP,
metrics_labels::BACKEND_PD,
metrics_labels::CONNECTION_HTTP,
model,
endpoint,
error_type_from_status(response.status()),
@@ -533,7 +528,7 @@ impl PDRouter {
context: PDRequestContext<'_>,
prefill: &dyn Worker,
decode: &dyn Worker,
start_time: Instant,
_start_time: Instant,
) -> Response {
// For non-streaming: use guard for automatic load management
// For streaming: load will be managed in create_streaming_response
@@ -577,12 +572,6 @@ impl PDRouter {
events::RequestReceivedEvent {}.emit();
let duration = start_time.elapsed();
RouterMetrics::record_pd_request_duration(context.route, duration);
RouterMetrics::record_pd_request(context.route);
RouterMetrics::record_pd_prefill_request(prefill.url());
RouterMetrics::record_pd_decode_request(decode.url());
// Process decode response
match decode_result {
Ok(res) => {
@@ -591,7 +580,6 @@ impl PDRouter {
debug!("Decode response status: {}", status);
if !status.is_success() {
RouterMetrics::record_pd_decode_error(decode.url());
error!(
"Decode server returned error status decode_url={} status={}",
decode.url(),
@@ -691,7 +679,6 @@ impl PDRouter {
error = %e,
"Decode request failed"
);
RouterMetrics::record_pd_decode_error(decode.url());
error::bad_gateway("decode_server_error", format!("Decode server error: {}", e))
}
}
@@ -754,15 +741,15 @@ impl PDRouter {
// Record worker selection metrics (Layer 3)
let model = model_id.unwrap_or("default");
SmgMetrics::record_worker_selection(
smg_labels::WORKER_PREFILL,
smg_labels::CONNECTION_HTTP,
Metrics::record_worker_selection(
metrics_labels::WORKER_PREFILL,
metrics_labels::CONNECTION_HTTP,
model,
prefill_policy.name(),
);
SmgMetrics::record_worker_selection(
smg_labels::WORKER_DECODE,
smg_labels::CONNECTION_HTTP,
Metrics::record_worker_selection(
metrics_labels::WORKER_DECODE,
metrics_labels::CONNECTION_HTTP,
model,
decode_policy.name(),
);
@@ -862,7 +849,6 @@ impl PDRouter {
Err(e) => {
if let Some(ref url) = decode_url {
error!("Stream error from decode server {}: {}", url, e);
RouterMetrics::record_pd_stream_error(url);
}
let _ = tx.send(Err(format!("Stream error: {}", e)));
break;
@@ -957,7 +943,6 @@ impl PDRouter {
let prefill_response = match prefill_result {
Ok(response) => response,
Err(e) => {
RouterMetrics::record_pd_prefill_error(prefill_url);
error!(
"Prefill server failed (CRITICAL) prefill_url={} error={}. Decode will timeout without prefill KV cache.",
prefill_url,
@@ -980,8 +965,6 @@ impl PDRouter {
// Check if prefill succeeded
if !prefill_status.is_success() {
RouterMetrics::record_pd_prefill_error(prefill_url);
// Get error body from prefill
let error_msg = prefill_response
.text()
sgl-model-gateway/src/routers/http/router.rs
+28 -75
View File
@@ -24,7 +24,7 @@ use crate::{
},
observability::{
events::{self, Event},
metrics::{smg_labels, RouterMetrics, SmgMetrics},
metrics::{metrics_labels, Metrics},
otel_trace::inject_trace_context_http,
},
policies::PolicyRegistry,
@@ -159,9 +159,9 @@ impl Router {
let idx = policy.select_worker(&available, text)?;
// Record worker selection metric (Layer 3)
SmgMetrics::record_worker_selection(
smg_labels::WORKER_REGULAR,
smg_labels::CONNECTION_HTTP,
Metrics::record_worker_selection(
metrics_labels::WORKER_REGULAR,
metrics_labels::CONNECTION_HTTP,
model_id.unwrap_or("default"),
policy.name(),
);
@@ -183,10 +183,10 @@ impl Router {
let endpoint = route_to_endpoint(route);
// Record request start (Layer 2)
SmgMetrics::record_router_request(
smg_labels::ROUTER_HTTP,
smg_labels::BACKEND_REGULAR,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_request(
metrics_labels::ROUTER_HTTP,
metrics_labels::BACKEND_REGULAR,
metrics_labels::CONNECTION_HTTP,
model,
endpoint,
is_stream,
@@ -196,55 +196,39 @@ impl Router {
&self.retry_config,
// operation per attempt
|_: u32| async {
let res = self
.route_typed_request_once(headers, typed_req, route, model_id, is_stream, &text)
.await;
// Need to be outside `route_typed_request_once` because that function has multiple return paths
RouterMetrics::record_attempt_http_response(
route,
res.status().as_u16(),
extract_error_code_from_response(&res),
);
res
self.route_typed_request_once(headers, typed_req, route, model_id, is_stream, &text)
.await
},
// should_retry predicate
|res, _attempt| is_retryable_status(res.status()),
// on_backoff hook
|delay, attempt| {
RouterMetrics::record_retry(route);
RouterMetrics::record_retry_backoff_duration(delay, attempt);
// Layer 3 worker metrics
SmgMetrics::record_worker_retry(smg_labels::WORKER_REGULAR, endpoint);
SmgMetrics::record_worker_retry_backoff(attempt, delay);
Metrics::record_worker_retry(metrics_labels::WORKER_REGULAR, endpoint);
Metrics::record_worker_retry_backoff(attempt, delay);
},
// on_exhausted hook
|| {
RouterMetrics::record_retries_exhausted(route);
SmgMetrics::record_worker_retries_exhausted(smg_labels::WORKER_REGULAR, endpoint);
Metrics::record_worker_retries_exhausted(metrics_labels::WORKER_REGULAR, endpoint);
},
)
.await;
if response.status().is_success() {
let duration = start.elapsed();
RouterMetrics::record_request(route);
RouterMetrics::record_generate_duration(duration);
SmgMetrics::record_router_duration(
smg_labels::ROUTER_HTTP,
smg_labels::BACKEND_REGULAR,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_duration(
metrics_labels::ROUTER_HTTP,
metrics_labels::BACKEND_REGULAR,
metrics_labels::CONNECTION_HTTP,
model,
endpoint,
duration,
);
} else if !is_retryable_status(response.status()) {
RouterMetrics::record_request_error(route, "non_retryable_error");
SmgMetrics::record_router_error(
smg_labels::ROUTER_HTTP,
smg_labels::BACKEND_REGULAR,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_HTTP,
metrics_labels::BACKEND_REGULAR,
metrics_labels::CONNECTION_HTTP,
model,
endpoint,
error_type_from_status(response.status()),
@@ -266,7 +250,6 @@ impl Router {
let worker = match self.select_worker_for_model(model_id, Some(text)) {
Some(w) => w,
None => {
RouterMetrics::record_request_error(route, "no_available_workers");
return error::service_unavailable(
"no_available_workers",
"No available workers (all circuits open or unhealthy)",
@@ -310,9 +293,9 @@ impl Router {
// Record worker errors for server errors (5xx)
if status.is_server_error() {
SmgMetrics::record_worker_error(
smg_labels::WORKER_REGULAR,
smg_labels::CONNECTION_HTTP,
Metrics::record_worker_error(
metrics_labels::WORKER_REGULAR,
metrics_labels::CONNECTION_HTTP,
error_type_from_status(status),
);
}
@@ -687,8 +670,6 @@ fn convert_reqwest_error(e: reqwest::Error) -> Response {
use async_trait::async_trait;
use crate::routers::error::extract_error_code_from_response;
#[async_trait]
impl RouterTrait for Router {
fn as_any(&self) -> &dyn std::any::Any {
@@ -772,22 +753,8 @@ impl RouterTrait for Router {
body: &EmbeddingRequest,
model_id: Option<&str>,
) -> Response {
// Record embeddings-specific metrics in addition to general request metrics
let start = Instant::now();
let res = self
.route_typed_request(headers, body, "/v1/embeddings", model_id)
.await;
// Embedding specific metrics
if res.status().is_success() {
RouterMetrics::record_embeddings_request();
RouterMetrics::record_embeddings_duration(start.elapsed());
} else {
let error_type = format!("http_{}", res.status().as_u16());
RouterMetrics::record_embeddings_error(&error_type);
}
res
self.route_typed_request(headers, body, "/v1/embeddings", model_id)
.await
}
async fn route_classify(
@@ -796,22 +763,8 @@ impl RouterTrait for Router {
body: &ClassifyRequest,
model_id: Option<&str>,
) -> Response {
// Record classification-specific metrics in addition to general request metrics
let start = Instant::now();
let res = self
.route_typed_request(headers, body, "/v1/classify", model_id)
.await;
// Classification specific metrics
if res.status().is_success() {
RouterMetrics::record_classify_request();
RouterMetrics::record_classify_duration(start.elapsed());
} else {
let error_type = format!("http_{}", res.status().as_u16());
RouterMetrics::record_classify_error(&error_type);
}
res
self.route_typed_request(headers, body, "/v1/classify", model_id)
.await
}
async fn route_rerank(
sgl-model-gateway/src/routers/openai/router.rs
+80 -80
View File
@@ -36,7 +36,7 @@ use crate::{
app_context::AppContext,
core::{model_type::Endpoint, ModelCard, ProviderType, RuntimeType, Worker, WorkerRegistry},
data_connector::{ConversationId, ListParams, ResponseId, SortOrder},
observability::metrics::{smg_labels, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
protocols::{
chat::ChatCompletionRequest,
responses::{
@@ -583,12 +583,12 @@ impl crate::routers::RouterTrait for OpenAIRouter {
let streaming = body.stream;
// Record request start
SmgMetrics::record_router_request(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_request(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
streaming,
);
@@ -600,13 +600,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
{
Ok(w) => w,
Err(response) => {
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_NO_WORKERS,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_NO_WORKERS,
);
return response;
}
@@ -615,13 +615,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
let mut payload = match to_value(body) {
Ok(v) => v,
Err(e) => {
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_VALIDATION,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_VALIDATION,
);
return error_responses::bad_request(format!("Failed to serialize request: {}", e));
}
@@ -629,13 +629,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
let provider = self.get_provider_arc_for_worker(worker.as_ref(), model_id);
if let Err(e) = provider.transform_request(&mut payload, Endpoint::Chat) {
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_VALIDATION,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_VALIDATION,
);
return error_responses::bad_request(format!("Provider transform error: {}", e));
}
@@ -672,13 +672,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
Ok(r) => r,
Err(e) => {
worker.circuit_breaker().record_failure();
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_BACKEND,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_BACKEND,
);
return (
StatusCode::SERVICE_UNAVAILABLE,
@@ -696,12 +696,12 @@ impl crate::routers::RouterTrait for OpenAIRouter {
match resp.bytes().await {
Ok(body) => {
worker.circuit_breaker().record_success();
SmgMetrics::record_router_duration(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_duration(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
start.elapsed(),
);
let mut response = Response::new(Body::from(body));
@@ -713,13 +713,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
}
Err(e) => {
worker.circuit_breaker().record_failure();
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
smg_labels::ERROR_BACKEND,
metrics_labels::ENDPOINT_CHAT,
metrics_labels::ERROR_BACKEND,
);
(
StatusCode::INTERNAL_SERVER_ERROR,
@@ -730,12 +730,12 @@ impl crate::routers::RouterTrait for OpenAIRouter {
}
} else {
// For streaming, record duration at start since we can't track completion
SmgMetrics::record_router_duration(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_duration(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_CHAT,
metrics_labels::ENDPOINT_CHAT,
start.elapsed(),
);
let stream = resp.bytes_stream();
@@ -776,12 +776,12 @@ impl crate::routers::RouterTrait for OpenAIRouter {
let streaming = body.stream.unwrap_or(false);
// Record request start
SmgMetrics::record_router_request(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_request(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_RESPONSES,
metrics_labels::ENDPOINT_RESPONSES,
streaming,
);
@@ -793,13 +793,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
{
Ok(w) => w,
Err(response) => {
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_RESPONSES,
smg_labels::ERROR_NO_WORKERS,
metrics_labels::ENDPOINT_RESPONSES,
metrics_labels::ERROR_NO_WORKERS,
);
return response;
}
@@ -853,13 +853,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
.get_conversation(&conv_id)
.await
{
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_RESPONSES,
smg_labels::ERROR_VALIDATION,
metrics_labels::ENDPOINT_RESPONSES,
metrics_labels::ERROR_VALIDATION,
);
return error_responses::not_found("conversation", &conv_id.0);
}
@@ -970,13 +970,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
let mut payload = match to_value(&request_body) {
Ok(v) => v,
Err(e) => {
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_RESPONSES,
smg_labels::ERROR_VALIDATION,
metrics_labels::ENDPOINT_RESPONSES,
metrics_labels::ERROR_VALIDATION,
);
return error_responses::bad_request(format!("Failed to serialize request: {}", e));
}
@@ -984,13 +984,13 @@ impl crate::routers::RouterTrait for OpenAIRouter {
let provider = self.get_provider_arc_for_worker(worker.as_ref(), model_id);
if let Err(e) = provider.transform_request(&mut payload, Endpoint::Responses) {
SmgMetrics::record_router_error(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_error(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_RESPONSES,
smg_labels::ERROR_VALIDATION,
metrics_labels::ENDPOINT_RESPONSES,
metrics_labels::ERROR_VALIDATION,
);
return error_responses::bad_request(format!("Provider transform error: {}", e));
}
@@ -1021,12 +1021,12 @@ impl crate::routers::RouterTrait for OpenAIRouter {
// Record duration only for successful requests (errors tracked inside handlers)
if response.status().is_success() {
SmgMetrics::record_router_duration(
smg_labels::ROUTER_OPENAI,
smg_labels::BACKEND_EXTERNAL,
smg_labels::CONNECTION_HTTP,
Metrics::record_router_duration(
metrics_labels::ROUTER_OPENAI,
metrics_labels::BACKEND_EXTERNAL,
metrics_labels::CONNECTION_HTTP,
model,
smg_labels::ENDPOINT_RESPONSES,
metrics_labels::ENDPOINT_RESPONSES,
start.elapsed(),
);
}
sgl-model-gateway/src/service_discovery.rs
+17 -21
View File
@@ -21,7 +21,7 @@ use tracing::{debug, error, info, warn};
use crate::{
app_context::AppContext,
core::Job,
observability::metrics::{smg_labels, RouterMetrics, SmgMetrics},
observability::metrics::{metrics_labels, Metrics},
protocols::worker_spec::WorkerConfigRequest,
};
@@ -302,7 +302,6 @@ pub async fn start_service_discovery(
}
Err(err) => {
error!("Error in Kubernetes watcher: {}", err);
RouterMetrics::record_discovery_watcher_error();
warn!(
"Retrying in {} seconds with exponential backoff",
retry_delay.as_secs()
@@ -317,7 +316,6 @@ pub async fn start_service_discovery(
"Kubernetes watcher exited, restarting in {} seconds",
config_arc.check_interval.as_secs()
);
RouterMetrics::record_discovery_watcher_restart();
time::sleep(config_arc.check_interval).await;
}
});
@@ -412,17 +410,16 @@ async fn handle_pod_event(
match job_queue.submit(job).await {
Ok(_) => {
debug!("Worker addition job submitted for: {}", worker_url);
RouterMetrics::record_discovery_update(1, 0);
// Layer 4: Record successful registration from K8s discovery
SmgMetrics::record_discovery_registration(
smg_labels::DISCOVERY_KUBERNETES,
smg_labels::REGISTRATION_SUCCESS,
Metrics::record_discovery_registration(
metrics_labels::DISCOVERY_KUBERNETES,
metrics_labels::REGISTRATION_SUCCESS,
);
// Update workers discovered gauge (using count from initial lock)
SmgMetrics::set_discovery_workers_discovered(
smg_labels::DISCOVERY_KUBERNETES,
Metrics::set_discovery_workers_discovered(
metrics_labels::DISCOVERY_KUBERNETES,
tracked_count,
);
}
@@ -433,9 +430,9 @@ async fn handle_pod_event(
);
// Layer 4: Record failed registration
SmgMetrics::record_discovery_registration(
smg_labels::DISCOVERY_KUBERNETES,
smg_labels::REGISTRATION_FAILED,
Metrics::record_discovery_registration(
metrics_labels::DISCOVERY_KUBERNETES,
metrics_labels::REGISTRATION_FAILED,
);
if let Ok(mut tracker) = tracked_pods.lock() {
@@ -451,9 +448,9 @@ async fn handle_pod_event(
}
} else {
// Pod already tracked - this is a duplicate event
SmgMetrics::record_discovery_registration(
smg_labels::DISCOVERY_KUBERNETES,
smg_labels::REGISTRATION_DUPLICATE,
Metrics::record_discovery_registration(
metrics_labels::DISCOVERY_KUBERNETES,
metrics_labels::REGISTRATION_DUPLICATE,
);
}
}
@@ -498,17 +495,16 @@ async fn handle_pod_deletion(
);
} else {
debug!("Submitted worker removal job for {}", worker_url);
RouterMetrics::record_discovery_update(0, 1);
// Layer 4: Record deregistration from K8s pod deletion
SmgMetrics::record_discovery_deregistration(
smg_labels::DISCOVERY_KUBERNETES,
smg_labels::DEREGISTRATION_POD_DELETED,
Metrics::record_discovery_deregistration(
metrics_labels::DISCOVERY_KUBERNETES,
metrics_labels::DEREGISTRATION_POD_DELETED,
);
// Update workers discovered gauge (using count from initial lock)
SmgMetrics::set_discovery_workers_discovered(
smg_labels::DISCOVERY_KUBERNETES,
Metrics::set_discovery_workers_discovered(
metrics_labels::DISCOVERY_KUBERNETES,
remaining_count,
);
}