From 3518b3317834088f664f4cd015977b21013e81cd Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Mon, 15 Dec 2025 04:57:42 -0800 Subject: [PATCH] [model-gateway] Remove legacy RouterMetrics and Rename SmgMetrics to Metrics and smg_labels to metrics_labels (#15160) --- sgl-model-gateway/src/core/circuit_breaker.rs | 29 +- sgl-model-gateway/src/core/job_queue.rs | 26 +- sgl-model-gateway/src/core/retry.rs | 4 +- .../src/core/steps/mcp_registration.rs | 4 +- .../local/remove_from_worker_registry.rs | 7 +- .../src/core/steps/worker/shared/register.rs | 7 +- sgl-model-gateway/src/core/worker.rs | 20 +- sgl-model-gateway/src/core/worker_registry.rs | 6 +- sgl-model-gateway/src/middleware.rs | 49 +- .../src/observability/metrics.rs | 650 +----------------- sgl-model-gateway/src/policies/cache_aware.rs | 16 +- .../src/policies/power_of_two.rs | 4 +- sgl-model-gateway/src/policies/random.rs | 5 +- sgl-model-gateway/src/policies/round_robin.rs | 5 +- .../grpc/common/stages/worker_selection.rs | 20 +- .../src/routers/grpc/harmony/responses.rs | 20 +- .../src/routers/grpc/harmony/streaming.rs | 18 +- .../src/routers/grpc/pipeline.rs | 118 ++-- .../grpc/regular/responses/tool_loop.rs | 22 +- .../src/routers/grpc/regular/streaming.rs | 14 +- sgl-model-gateway/src/routers/grpc/utils.rs | 22 +- .../src/routers/http/pd_router.rs | 79 +-- sgl-model-gateway/src/routers/http/router.rs | 103 +-- .../src/routers/openai/router.rs | 160 ++--- sgl-model-gateway/src/service_discovery.rs | 38 +- 25 files changed, 332 insertions(+), 1114 deletions(-) diff --git a/sgl-model-gateway/src/core/circuit_breaker.rs b/sgl-model-gateway/src/core/circuit_breaker.rs index b9d0cbe52..c1f140682 100644 --- a/sgl-model-gateway/src/core/circuit_breaker.rs +++ b/sgl-model-gateway/src/core/circuit_breaker.rs @@ -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()); } } diff --git a/sgl-model-gateway/src/core/job_queue.rs b/sgl-model-gateway/src/core/job_queue.rs index 13607f135..cd5324ee4 100644 --- a/sgl-model-gateway/src/core/job_queue.rs +++ b/sgl-model-gateway/src/core/job_queue.rs @@ -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, status_map: &Arc>, ) { - 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 ); } } diff --git a/sgl-model-gateway/src/core/retry.rs b/sgl-model-gateway/src/core/retry.rs index d2ced03a1..7864dcf85 100644 --- a/sgl-model-gateway/src/core/retry.rs +++ b/sgl-model-gateway/src/core/retry.rs @@ -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( diff --git a/sgl-model-gateway/src/core/steps/mcp_registration.rs b/sgl-model-gateway/src/core/steps/mcp_registration.rs index 55fac7a7e..dd49cd6e5 100644 --- a/sgl-model-gateway/src/core/steps/mcp_registration.rs +++ b/sgl-model-gateway/src/core/steps/mcp_registration.rs @@ -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); diff --git a/sgl-model-gateway/src/core/steps/worker/local/remove_from_worker_registry.rs b/sgl-model-gateway/src/core/steps/worker/local/remove_from_worker_registry.rs index 7da3fefce..f507e5542 100644 --- a/sgl-model-gateway/src/core/steps/worker/local/remove_from_worker_registry.rs +++ b/sgl-model-gateway/src/core/steps/worker/local/remove_from_worker_registry.rs @@ -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, diff --git a/sgl-model-gateway/src/core/steps/worker/shared/register.rs b/sgl-model-gateway/src/core/steps/worker/shared/register.rs index ae2e15b77..e0584e927 100644 --- a/sgl-model-gateway/src/core/steps/worker/shared/register.rs +++ b/sgl-model-gateway/src/core/steps/worker/shared/register.rs @@ -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, diff --git a/sgl-model-gateway/src/core/worker.rs b/sgl-model-gateway/src/core/worker.rs index 8541c8cfd..16522231a 100644 --- a/sgl-model-gateway/src/core/worker.rs +++ b/sgl-model-gateway/src/core/worker.rs @@ -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 diff --git a/sgl-model-gateway/src/core/worker_registry.rs b/sgl-model-gateway/src/core/worker_registry.rs index 8fc0c00b3..4222119d2 100644 --- a/sgl-model-gateway/src/core/worker_registry.rs +++ b/sgl-model-gateway/src/core/worker_registry.rs @@ -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 { diff --git a/sgl-model-gateway/src/middleware.rs b/sgl-model-gateway/src/middleware.rs index 89bc08b29..ab8f2ea96 100644 --- a/sgl-model-gateway/src/middleware.rs +++ b/sgl-model-gateway/src/middleware.rs @@ -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 OnRequest 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 OnResponse 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) }) diff --git a/sgl-model-gateway/src/observability/metrics.rs b/sgl-model-gateway/src/observability/metrics.rs index 98bcaf4a5..1033d3bd5 100644 --- a/sgl-model-gateway/src/observability/metrics.rs +++ b/sgl-model-gateway/src/observability/metrics.rs @@ -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)); - } } diff --git a/sgl-model-gateway/src/policies/cache_aware.rs b/sgl-model-gateway/src/policies/cache_aware.rs index 1a1a8b598..6a6545367 100644 --- a/sgl-model-gateway/src/policies/cache_aware.rs +++ b/sgl-model-gateway/src/policies/cache_aware.rs @@ -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); } diff --git a/sgl-model-gateway/src/policies/power_of_two.rs b/sgl-model-gateway/src/policies/power_of_two.rs index 9f7856eb0..802a51194 100644 --- a/sgl-model-gateway/src/policies/power_of_two.rs +++ b/sgl-model-gateway/src/policies/power_of_two.rs @@ -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) } diff --git a/sgl-model-gateway/src/policies/random.rs b/sgl-model-gateway/src/policies/random.rs index e316f95c7..12f0ac1dd 100644 --- a/sgl-model-gateway/src/policies/random.rs +++ b/sgl-model-gateway/src/policies/random.rs @@ -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]) } diff --git a/sgl-model-gateway/src/policies/round_robin.rs b/sgl-model-gateway/src/policies/round_robin.rs index 1c50884f5..739f18ca6 100644 --- a/sgl-model-gateway/src/policies/round_robin.rs +++ b/sgl-model-gateway/src/policies/round_robin.rs @@ -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]) } diff --git a/sgl-model-gateway/src/routers/grpc/common/stages/worker_selection.rs b/sgl-model-gateway/src/routers/grpc/common/stages/worker_selection.rs index 7cbf6976b..17785f3bf 100644 --- a/sgl-model-gateway/src/routers/grpc/common/stages/worker_selection.rs +++ b/sgl-model-gateway/src/routers/grpc/common/stages/worker_selection.rs @@ -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, ); diff --git a/sgl-model-gateway/src/routers/grpc/harmony/responses.rs b/sgl-model-gateway/src/routers/grpc/harmony/responses.rs index 0453aef75..da39bed03 100644 --- a/sgl-model-gateway/src/routers/grpc/harmony/responses.rs +++ b/sgl-model-gateway/src/routers/grpc/harmony/responses.rs @@ -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(¤t_request.model); + Metrics::record_mcp_tool_iteration(¤t_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(¤t_request.model); + Metrics::record_mcp_tool_iteration(¤t_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) diff --git a/sgl-model-gateway/src/routers/grpc/harmony/streaming.rs b/sgl-model-gateway/src/routers/grpc/harmony/streaming.rs index 174ae9d14..14a53ec06 100644 --- a/sgl-model-gateway/src/routers/grpc/harmony/streaming.rs +++ b/sgl-model-gateway/src/routers/grpc/harmony/streaming.rs @@ -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), diff --git a/sgl-model-gateway/src/routers/grpc/pipeline.rs b/sgl-model-gateway/src/routers/grpc/pipeline.rs index 6a218ffeb..cf09fb8e3 100644 --- a/sgl-model-gateway/src/routers/grpc/pipeline.rs +++ b/sgl-model-gateway/src/routers/grpc/pipeline.rs @@ -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> = 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> = 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") } diff --git a/sgl-model-gateway/src/routers/grpc/regular/responses/tool_loop.rs b/sgl-model-gateway/src/routers/grpc/regular/responses/tool_loop.rs index 71e97f5a6..d88b9f0d9 100644 --- a/sgl-model-gateway/src/routers/grpc/regular/responses/tool_loop.rs +++ b/sgl-model-gateway/src/routers/grpc/regular/responses/tool_loop.rs @@ -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(¤t_request.model); + Metrics::record_mcp_tool_iteration(¤t_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( ¤t_request.model, &tool_name, tool_duration, ); - SmgMetrics::record_mcp_tool_call( + Metrics::record_mcp_tool_call( ¤t_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 }, ); diff --git a/sgl-model-gateway/src/routers/grpc/regular/streaming.rs b/sgl-model-gateway/src/routers/grpc/regular/streaming.rs index 74f291604..5875ea130 100644 --- a/sgl-model-gateway/src/routers/grpc/regular/streaming.rs +++ b/sgl-model-gateway/src/routers/grpc/regular/streaming.rs @@ -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 diff --git a/sgl-model-gateway/src/routers/grpc/utils.rs b/sgl-model-gateway/src/routers/grpc/utils.rs index 8e2be7994..3f0770ca0 100644 --- a/sgl-model-gateway/src/routers/grpc/utils.rs +++ b/sgl-model-gateway/src/routers/grpc/utils.rs @@ -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, } } diff --git a/sgl-model-gateway/src/routers/http/pd_router.rs b/sgl-model-gateway/src/routers/http/pd_router.rs index 2c32097f1..76032fee0 100644 --- a/sgl-model-gateway/src/routers/http/pd_router.rs +++ b/sgl-model-gateway/src/routers/http/pd_router.rs @@ -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() diff --git a/sgl-model-gateway/src/routers/http/router.rs b/sgl-model-gateway/src/routers/http/router.rs index 2a4467f90..7c207a535 100644 --- a/sgl-model-gateway/src/routers/http/router.rs +++ b/sgl-model-gateway/src/routers/http/router.rs @@ -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( diff --git a/sgl-model-gateway/src/routers/openai/router.rs b/sgl-model-gateway/src/routers/openai/router.rs index 5bfe7899a..574529e46 100644 --- a/sgl-model-gateway/src/routers/openai/router.rs +++ b/sgl-model-gateway/src/routers/openai/router.rs @@ -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(), ); } diff --git a/sgl-model-gateway/src/service_discovery.rs b/sgl-model-gateway/src/service_discovery.rs index 6024ddb88..5115cee39 100644 --- a/sgl-model-gateway/src/service_discovery.rs +++ b/sgl-model-gateway/src/service_discovery.rs @@ -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, ); }