Incoporating a round of feedback

Signed-off-by: Zhongxuan Wang <[email protected]>

Author: zhongxuanwang-nv

components/src/dynamo/vllm/handlers.py

@@ -75,11 +75,11 @@ def build_sampling_params(
 
 
 def _request_contains_timing_metrics(request: Dict[str, Any]) -> bool:
-    """Check if timing_metrics is requested in extra_fields."""
-    extra_fields: Optional[List[str]] = request.get("extra_fields")
-    if extra_fields is None:
+    """Check if timing_metrics is requested in observability_fields."""
+    observability_fields: Optional[List[str]] = request.get("observability_fields")
+    if observability_fields is None:
         return False
-    return "timing_metrics" in extra_fields
+    return "timing_metrics" in observability_fields
 
 
 class BaseWorkerHandler(ABC):
@@ -259,10 +259,10 @@ async def generate_tokens(
                     out = {"token_ids": output.token_ids[num_output_tokens_so_far:]}
                     if output.finish_reason:
                         out["finish_reason"] = output.finish_reason
-                        out[
-                            "completion_usage"
-                        ] = BaseWorkerHandler._build_completion_usage(
-                            request_output=res,
+                        out["completion_usage"] = (
+                            BaseWorkerHandler._build_completion_usage(
+                                request_output=res,
+                            )
                         )
                     if output.stop_reason:
                         out["stop_reason"] = output.stop_reason
@@ -309,9 +309,18 @@ async def generate(self, request, context):
         include_timing = _request_contains_timing_metrics(request)
 
         # Initialize timing metrics using request_received_seconds from frontend (passed via PreprocessedRequest)
-        # NOTE: If frontend, prefill workers, and decode workers are running on different machines,
-        # there may be slight clock drifts between them. As a result, timing values recorded on
-        # different machines may not be perfectly synchronized and could show minor inconsistencies.
+        #
+        # TIMING METRICS:
+        # - Reliable durations: Use same-machine timestamps (e.g., decode_end - decode_start).
+        #   We use time.perf_counter() for intra-worker duration calculations to ensure monotonic,
+        #   high-resolution timing that's immune to system clock adjustments.
+        # - Cross-machine calculations (e.g., prefill_start - request_received) assume perfect NTP
+        #   synchronization and should be used with UTMOST CAUTION due to clock drift. Even with NTP,
+        #   clocks can drift by milliseconds each day, leading to negative durations or misleading latency values.
+        #   These cross-machine metrics are useful for rough end-to-end analysis but should not be
+        #   relied upon for precise performance measurements.
+        # - TODO: Measure actual overhead (network, queueing, etc.) - expected to be low but needs
+        #   benchmarking
         timing_metrics: Dict[str, float] = {}
         if include_timing:
             # Use request_received_seconds from the request (set by frontend) if available
@@ -371,6 +380,7 @@ async def generate(self, request, context):
                 # Record decode start time
                 if include_timing:
                     decode_start_seconds = time.time()
+                    decode_start_perf_counter = time.perf_counter()
                     # If this is aggregated mode (no prefill_result), prefill_start == decode_start
                     if prefill_result is None:
                         timing_metrics["prefill_start_seconds"] = decode_start_seconds
@@ -396,7 +406,9 @@ async def generate(self, request, context):
                     # On finish, record decode_end_seconds and inject timing_metrics
                     # Note: request_finish_seconds is set in the Rust HTTP layer when the response actually leaves the server
                     if tok.get("finish_reason") is not None and include_timing:
-                        timing_metrics["decode_end_seconds"] = time.time()
+                        timing_metrics["decode_end_seconds"] = decode_start_seconds + (
+                            time.perf_counter() - decode_start_perf_counter
+                        )
 
                         # Inject timing_metrics into disaggregated_params
                         if (
@@ -442,9 +454,7 @@ async def generate(self, request, context):
         include_timing = _request_contains_timing_metrics(request)
 
         # Initialize timing metrics using request_received_seconds from frontend (passed via PreprocessedRequest)
-        # NOTE: If frontend, prefill workers, and decode workers are running on different machines,
-        # there may be slight clock drifts between them. As a result, timing values recorded on
-        # different machines may not be perfectly synchronized and could show minor inconsistencies.
+        # See DecodeWorkerHandler.generate() for timing metrics documentation
         timing_metrics: Dict[str, float] = {}
         if include_timing:
             # Use request_received_seconds from the request (set by frontend) if available
@@ -453,7 +463,9 @@ async def generate(self, request, context):
                 timing_metrics["request_received_seconds"] = frontend_received
 
             # Record prefill_start as when we start processing in the prefill worker
-            timing_metrics["prefill_start_seconds"] = time.time()
+            prefill_start_seconds = time.time()
+            prefill_start_perf_counter = time.perf_counter()
+            timing_metrics["prefill_start_seconds"] = prefill_start_seconds
 
         # Extract and decode multimodal data if present
         multi_modal_data = await self._extract_multimodal_data(request)
@@ -511,12 +523,15 @@ async def generate(self, request, context):
                     disaggregated_params: Optional[Dict[str, Any]] = {}
 
                     if res.kv_transfer_params:
-                        disaggregated_params[
-                            "kv_transfer_params"
-                        ] = res.kv_transfer_params
+                        disaggregated_params["kv_transfer_params"] = (
+                            res.kv_transfer_params
+                        )
 
                     if include_timing and timing_metrics:
-                        timing_metrics["prefill_end_seconds"] = time.time()
+                        timing_metrics["prefill_end_seconds"] = (
+                            prefill_start_seconds
+                            + (time.perf_counter() - prefill_start_perf_counter)
+                        )
                         disaggregated_params["timing_metrics"] = timing_metrics
 
                     output: Dict[str, Any] = {

components/src/dynamo/vllm/multimodal_handlers/worker_handler.py

@@ -227,9 +227,9 @@ async def generate(self, request: vLLMMultimodalRequest, context):
                 # Update the prompt token id in the decode request to the one
                 # in response, which has image templated filled in. So that
                 # the decode worker will fetch correct amount of KV blocks.
-                decode_request.engine_prompt[
-                    "prompt_token_ids"
-                ] = prefill_response.prompt_token_ids
+                decode_request.engine_prompt["prompt_token_ids"] = (
+                    prefill_response.prompt_token_ids
+                )
                 logger.debug(
                     f"Prefill response kv_transfer_params: {prefill_response.kv_transfer_params}"
                 )

components/src/dynamo/vllm/multimodal_utils/chat_processor.py

@@ -178,7 +178,9 @@ async def stream_response(
         if request.stream:
             # Handle streaming response
             num_output_text_so_far = 0
-            async for raw_response in self.openai_serving.chat_completion_stream_generator(
+            async for (
+                raw_response
+            ) in self.openai_serving.chat_completion_stream_generator(
                 request,
                 result_generator,
                 request_id,
@@ -212,7 +214,9 @@ async def stream_response(
             # Collect all chunks into a single response
             full_response = None
             num_output_text_so_far = 0
-            async for raw_response in self.openai_serving.chat_completion_stream_generator(
+            async for (
+                raw_response
+            ) in self.openai_serving.chat_completion_stream_generator(
                 request,
                 result_generator,
                 request_id,

…amo/vllm/tests/test_vllm_extra_fields.py …/tests/test_vllm_observability_fields.pycomponents/src/dynamo/vllm/tests/test_vllm_extra_fields.py renamed to components/src/dynamo/vllm/tests/test_vllm_observability_fields.py components/src/dynamo/vllm/tests/test_vllm_extra_fields.py renamed to components/src/dynamo/vllm/tests/test_vllm_observability_fields.py

@@ -1,7 +1,7 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Unit tests for extra_fields handling in vLLM handlers."""
+"""Unit tests for observability_fields handling in vLLM handlers."""
 
 import asyncio
 import warnings
@@ -33,18 +33,18 @@
 class TestShouldIncludeTimingMetrics:
     """Tests for _request_contains_timing_metrics helper function."""
 
-    def test_returns_true_with_multiple_extra_fields(self):
+    def test_returns_true_with_multiple_observability_fields(self):
         """Timing metrics should be included when explicitly requested."""
-        request = {"extra_fields": ["worker_id", "timing_metrics", "other_field"]}
+        request = {"observability_fields": ["worker_id", "timing_metrics", "other_field"]}
         assert _request_contains_timing_metrics(request) is True
 
-    def test_returns_false_when_extra_fields_is_none(self):
-        """Timing metrics should not be included when extra_fields is None."""
-        request = {"extra_fields": None}
+    def test_returns_false_when_observability_fields_is_none(self):
+        """Timing metrics should not be included when observability_fields is None."""
+        request = {"observability_fields": None}
         assert _request_contains_timing_metrics(request) is False
 
-    def test_returns_false_when_extra_fields_missing(self):
-        """Timing metrics should not be included when extra_fields key is absent."""
+    def test_returns_false_when_observability_fields_missing(self):
+        """Timing metrics should not be included when observability_fields key is absent."""
         request: dict[str, list[str]] = {}
         assert _request_contains_timing_metrics(request) is False
 
@@ -145,7 +145,7 @@ async def mock_generate(*args, **kwargs):
             "token_ids": [1, 2, 3],
             "sampling_options": {},
             "stop_conditions": {},
-            "extra_fields": ["timing_metrics"],
+            "observability_fields": ["timing_metrics"],
             "request_received_seconds": 1000.0,
             "prefill_result": {
                 "disaggregated_params": {
@@ -192,7 +192,7 @@ async def mock_generate(*args, **kwargs):
             "token_ids": [1, 2, 3],
             "sampling_options": {},
             "stop_conditions": {},
-            "extra_fields": ["timing_metrics"],
+            "observability_fields": ["timing_metrics"],
             "request_received_seconds": 1000.0,
         }
 

examples/backends/vllm/launch/disagg_router.sh

@@ -9,13 +9,15 @@ export PYTHONHASHSEED=0
 
 # Common configuration
 MODEL="Qwen/Qwen3-0.6B"
-BLOCK_SIZE=64
+BLOCK_SIZE=16
+NUM_GPU_BLOCKS=20000
 
 # Start frontend with KV routing
 # The frontend will automatically detect prefill workers and activate an internal prefill router
 # dynamo.frontend accepts either --http-port flag or DYN_HTTP_PORT env var (defaults to 8000)
 python -m dynamo.frontend \
     --router-mode kv \
+    --enforce-disagg \
     --router-reset-states &
 
 # two decode workers
@@ -24,13 +26,15 @@ CUDA_VISIBLE_DEVICES=0 python3 -m dynamo.vllm \
     --model $MODEL \
     --block-size $BLOCK_SIZE \
     --enforce-eager \
+    --num-gpu-blocks-override $NUM_GPU_BLOCKS \
     --kv-events-config '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20080","enable_kv_cache_events":true}'&
 
 VLLM_NIXL_SIDE_CHANNEL_PORT=20097 \
 CUDA_VISIBLE_DEVICES=1 python3 -m dynamo.vllm \
     --model $MODEL \
     --block-size $BLOCK_SIZE \
     --enforce-eager \
+    --num-gpu-blocks-override $NUM_GPU_BLOCKS \
     --kv-events-config '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20081","enable_kv_cache_events":true}' &
 
 # two prefill workers
@@ -42,6 +46,7 @@ CUDA_VISIBLE_DEVICES=2 python3 -m dynamo.vllm \
     --block-size $BLOCK_SIZE \
     --enforce-eager \
     --is-prefill-worker \
+    --num-gpu-blocks-override $NUM_GPU_BLOCKS \
     --kv-events-config '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20082","enable_kv_cache_events":true}'&
 
 VLLM_NIXL_SIDE_CHANNEL_PORT=20099 \
@@ -50,4 +55,5 @@ CUDA_VISIBLE_DEVICES=3 python3 -m dynamo.vllm \
     --block-size $BLOCK_SIZE \
     --enforce-eager \
     --is-prefill-worker \
+    --num-gpu-blocks-override $NUM_GPU_BLOCKS \
     --kv-events-config '{"publisher":"zmq","topic":"kv-events","endpoint":"tcp://*:20083","enable_kv_cache_events":true}'

lib/llm/src/http/service/openai.rs

@@ -58,7 +58,7 @@ pub const ANNOTATION_REQUEST_ID: &str = "request_id";
 
 /// Injects `request_completed_seconds` into the nvext timing_metrics field.
 /// This captures the exact moment when the response is about to leave the server.
-/// Only injects if timing_metrics already exists (i.e., the user requested it via extra_fields).
+/// Only injects if timing_metrics already exists (i.e., the user requested it via observability_fields).
 fn inject_request_completed_seconds(nvext: &mut Option<serde_json::Value>) {
     let ts = SystemTime::now()
         .duration_since(UNIX_EPOCH)

lib/llm/src/kv_router.rs

@@ -690,7 +690,7 @@ impl AsyncEngine<SingleIn<PreprocessedRequest>, ManyOut<Annotated<LLMEngineOutpu
                         // Always inject worker_id in first item's disaggregated_params
                         // This is needed for:
                         // 1. PrefillRouter to know which prefill worker was chosen
-                        // 2. Client response when extra_fields contains "worker_id"
+                        // 2. Client response when observability_fields contains "worker_id"
                         if first_item {
                             first_item = false;
 

lib/llm/src/preprocessor.rs

@@ -237,10 +237,10 @@ impl OpenAIPreprocessor {
         builder.annotations(request.annotations().unwrap_or_default());
         builder.mdc_sum(Some(self.mdcsum.clone()));
         builder.estimated_prefix_hit_num_blocks(None);
-        // Extract backend_instance_id, extra_fields, and request_received_seconds from nvext if present
+        // Extract backend_instance_id, observability_fields, and request_received_seconds from nvext if present
         if let Some(nvext) = request.nvext() {
             builder.backend_instance_id(nvext.backend_instance_id);
-            builder.extra_fields(nvext.extra_fields.clone());
+            builder.observability_fields(nvext.observability_fields.clone());
             builder.request_received_seconds(nvext.request_received_seconds);
         }
 

lib/llm/src/protocols/common/preprocessor.rs

@@ -97,10 +97,10 @@ pub struct PreprocessedRequest {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub extra_args: Option<serde_json::Value>,
 
-    /// Extra fields requested to be included in the response's nvext
+    /// Observability fields requested to be included in the response's nvext
     #[builder(default)]
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub extra_fields: Option<Vec<String>>,
+    pub observability_fields: Option<Vec<String>>,
 
     /// Timestamp when the request was received by the frontend (seconds since epoch)
     /// Used for timing metrics to track end-to-end latency

lib/llm/src/protocols/openai/chat_completions/delta.rs

@@ -50,7 +50,7 @@ impl NvCreateChatCompletionRequest {
                 .unwrap_or(false),
             enable_logprobs: self.inner.logprobs.unwrap_or(false)
                 || self.inner.top_logprobs.unwrap_or(0) > 0,
-            extra_fields: self.nvext.as_ref().and_then(|nv| nv.extra_fields.clone()),
+            observability_fields: self.nvext.as_ref().and_then(|nv| nv.observability_fields.clone()),
             runtime_config: ModelRuntimeConfig::default(),
         };
 
@@ -66,7 +66,7 @@ pub struct DeltaGeneratorOptions {
     /// Determines whether log probabilities should be included in the response.
     pub enable_logprobs: bool,
     /// Extra fields to include in response nvext (e.g., "worker_id", "timing_metrics")
-    pub extra_fields: Option<Vec<String>>,
+    pub observability_fields: Option<Vec<String>>,
 
     pub runtime_config: ModelRuntimeConfig,
 }
@@ -292,10 +292,10 @@ impl DeltaGenerator {
         self.options.enable_usage
     }
 
-    /// Check if an extra field is requested
-    fn is_extra_field_requested(&self, field: &str) -> bool {
+    /// Check if an observability field is requested
+    fn is_observability_field_requested(&self, field: &str) -> bool {
         self.options
-            .extra_fields
+            .observability_fields
             .as_ref()
             .map(|fields| fields.iter().any(|f| f == field))
             .unwrap_or(false)
@@ -375,19 +375,19 @@ impl crate::protocols::openai::DeltaGeneratorExt<NvCreateChatCompletionStreamRes
         let mut stream_response = self.create_choice(index, delta.text, finish_reason, logprobs);
 
         // Extract worker_id and timing_metrics from disaggregated_params and inject into nvext
-        // Only include fields that were explicitly requested via extra_fields
+        // Only include fields that were explicitly requested via observability_fields
         if let Some(ref disaggregated_params) = delta.disaggregated_params {
             let mut nvext_obj = serde_json::Map::new();
 
             // Extract worker_id if present and requested
-            if self.is_extra_field_requested("worker_id")
+            if self.is_observability_field_requested("worker_id")
                 && let Some(worker_id_json) = disaggregated_params.get("worker_id")
             {
                 nvext_obj.insert("worker_id".to_string(), worker_id_json.clone());
             }
 
             // Extract timing_metrics if present and requested
-            if self.is_extra_field_requested("timing_metrics")
+            if self.is_observability_field_requested("timing_metrics")
                 && let Some(timing_metrics_json) = disaggregated_params.get("timing_metrics")
             {
                 nvext_obj.insert("timing_metrics".to_string(), timing_metrics_json.clone());
@@ -483,7 +483,7 @@ mod tests {
         use crate::protocols::openai::DeltaGeneratorExt;
 
         let options = DeltaGeneratorOptions {
-            extra_fields: Some(vec!["worker_id".to_string(), "timing_metrics".to_string()]),
+            observability_fields: Some(vec!["worker_id".to_string(), "timing_metrics".to_string()]),
             ..Default::default()
         };
         let mut generator = DeltaGenerator::new(