From e8b69e00fd4a463f9889024d7943a5d5bdcd271f Mon Sep 17 00:00:00 2001 From: ssssnow <38564764+ssssnow@users.noreply.github.com> Date: Wed, 14 Oct 2026 20:22:36 +0800 Subject: [PATCH 1/5] feat(observability): add exclusive frontend TTFT breakdown Signed-off-by: ssssnow <38564764+ssssnow@users.noreply.github.com> --- components/src/dynamo/frontend/prepost.py | 23 +- .../src/dynamo/frontend/sglang_prepost.py | 34 +- .../src/dynamo/frontend/sglang_processor.py | 90 ++- .../dynamo/frontend/tests/test_ttft_stages.py | 588 ++++++++++++++++++ components/src/dynamo/frontend/ttft.py | 13 + .../src/dynamo/frontend/vllm_processor.py | 153 +++-- .../components/frontend-configuration.mdx | 7 + .../pages/reference/observability/logging.mdx | 35 ++ .../reference/observability/metric-labels.mdx | 11 +- .../observability/metrics-catalog.mdx | 111 ++++ lib/bindings/python/rust/context.rs | 85 +++ lib/bindings/python/rust/engine.rs | 11 +- lib/bindings/python/rust/lib.rs | 1 + lib/bindings/python/rust/llm/routed_engine.rs | 47 +- lib/bindings/python/src/dynamo/_core.pyi | 19 + .../python/src/dynamo/prometheus_names.py | 2 + lib/llm/src/http/service/metrics.rs | 492 ++++++++++++++- lib/llm/src/http/service/openai.rs | 24 + lib/llm/src/kv_router/routing_host.rs | 14 +- lib/llm/src/kv_router/routing_host/builtin.rs | 3 + lib/llm/src/kv_router/routing_host/kv.rs | 15 +- .../kv_router/routing_host/request_guard.rs | 239 ++++++- lib/llm/src/kv_router/routing_host/tests.rs | 24 +- lib/llm/src/preprocessor.rs | 127 +++- lib/llm/src/protocols/common.rs | 1 + lib/llm/src/protocols/common/timing.rs | 26 + lib/llm/src/protocols/common/ttft.rs | 464 ++++++++++++++ lib/runtime/src/config/environment_names.rs | 3 + lib/runtime/src/metrics/prometheus_names.rs | 3 + 29 files changed, 2516 insertions(+), 149 deletions(-) create mode 100644 components/src/dynamo/frontend/tests/test_ttft_stages.py create mode 100644 components/src/dynamo/frontend/ttft.py create mode 100644 lib/llm/src/protocols/common/ttft.rs diff --git a/components/src/dynamo/frontend/prepost.py b/components/src/dynamo/frontend/prepost.py index d4004f18095a..2ce3a356940f 100644 --- a/components/src/dynamo/frontend/prepost.py +++ b/components/src/dynamo/frontend/prepost.py @@ -38,6 +38,7 @@ should_attempt_structural_tag, ) from .thinking import apply_default_thinking_mode_to_template_kwargs +from .ttft import ttft_stage from .utils import legacy_guided_decoding get_model_structural_tag: Any @@ -747,6 +748,7 @@ async def preprocess_chat_request( structural_tag_scope: str = "auto", structural_tag_schema: str = "auto", structural_tag_excludes_reasoning: bool = False, + context: Any | None = None, ) -> PreprocessResult: validated_request = _validate_chat_completion_request(request) assistant_guided_decoding = _build_assistant_guided_decoding(validated_request) @@ -803,17 +805,18 @@ async def preprocess_chat_request( # Render first: whether generation starts inside a reasoning block depends # on the rendered prompt (e.g. a template that ends the generation prompt # with an open think tag), and the tool-call grammar must match it. - _, engine_prompt = await renderer.render_messages_async(messages, chat_params) + with ttft_stage(context, "template_tokenize"): + _, engine_prompt = await renderer.render_messages_async(messages, chat_params) - if "prompt_token_ids" in engine_prompt: - tokens = list(engine_prompt["prompt_token_ids"]) - else: - async_tokenizer = _get_async_tokenizer(tokenizer) - encoded = await async_tokenizer( - engine_prompt["prompt"], - add_special_tokens=request_for_sampling.add_special_tokens, - ) - tokens = list(encoded.input_ids) + if "prompt_token_ids" in engine_prompt: + tokens = list(engine_prompt["prompt_token_ids"]) + else: + async_tokenizer = _get_async_tokenizer(tokenizer) + encoded = await async_tokenizer( + engine_prompt["prompt"], + add_special_tokens=request_for_sampling.add_special_tokens, + ) + tokens = list(encoded.input_ids) tool_guided_decoding = build_tool_call_guided_decoding( request_for_sampling, diff --git a/components/src/dynamo/frontend/sglang_prepost.py b/components/src/dynamo/frontend/sglang_prepost.py index 2c295da24704..8529912a7533 100644 --- a/components/src/dynamo/frontend/sglang_prepost.py +++ b/components/src/dynamo/frontend/sglang_prepost.py @@ -39,6 +39,7 @@ should_attempt_structural_tag, ) from .thinking import apply_default_thinking_mode_to_template_kwargs +from .ttft import ttft_stage from .utils import PreprocessError, legacy_guided_decoding, random_call_id logger = logging.getLogger(__name__) @@ -509,6 +510,7 @@ def _render_deepseek_v4_prompt_token_ids( messages: list[dict[str, Any]], tokenizer, template_tools: list[dict[str, Any]] | None, + context: Any | None = None, ) -> list[int]: """Render DeepSeek-V4 prompt token ids via SGLang's ``encoding_dsv4``. @@ -568,12 +570,13 @@ def _render_deepseek_v4_prompt_token_ids( if reasoning_effort not in ("max", "high", None): reasoning_effort = None - prompt = encode_messages( - encoding_messages, - thinking_mode=thinking_mode, - reasoning_effort=reasoning_effort, - ) - return _normalize_prompt_token_ids(tokenizer.encode(prompt)) + with ttft_stage(context, "template_tokenize"): + prompt = encode_messages( + encoding_messages, + thinking_mode=thinking_mode, + reasoning_effort=reasoning_effort, + ) + return _normalize_prompt_token_ids(tokenizer.encode(prompt)) @lru_cache(maxsize=64) @@ -874,6 +877,7 @@ def preprocess_chat_request( structural_tag_mode: str = "off", structural_tag_scope: str = "auto", structural_tag_schema: str = "auto", + context: Any | None = None, ) -> SglangPreprocessResult: """Preprocess a chat request using SGLang tokenizer and parser APIs. @@ -945,6 +949,7 @@ def preprocess_chat_request( messages=messages, tokenizer=tokenizer, template_tools=template_tools, + context=context, ) else: # Build template kwargs -- single call for rendering + tokenization @@ -968,14 +973,15 @@ def preprocess_chat_request( template_messages = _normalize_messages_for_template(messages, tokenizer) - try: - rendered = tokenizer.apply_chat_template( - template_messages, **template_kwargs - ) - except (TemplateError, TypeError) as exc: - # Jinja filters such as tojson can raise TypeError for invalid inputs. - raise PreprocessError(str(exc)) from exc - prompt_token_ids = _normalize_prompt_token_ids(rendered) + with ttft_stage(context, "template_tokenize"): + try: + rendered = tokenizer.apply_chat_template( + template_messages, **template_kwargs + ) + except (TemplateError, TypeError) as exc: + # Jinja filters such as tojson can raise TypeError for invalid inputs. + raise PreprocessError(str(exc)) from exc + prompt_token_ids = _normalize_prompt_token_ids(rendered) response_format_guided_decoding = build_response_format_guided_decoding(request) tool_call_guided_decoding = build_tool_call_guided_decoding( diff --git a/components/src/dynamo/frontend/sglang_processor.py b/components/src/dynamo/frontend/sglang_processor.py index 5a25d3e64140..8666bf35ff5c 100644 --- a/components/src/dynamo/frontend/sglang_processor.py +++ b/components/src/dynamo/frontend/sglang_processor.py @@ -14,6 +14,7 @@ from collections.abc import AsyncGenerator from concurrent.futures import ProcessPoolExecutor from concurrent.futures import wait as _futures_wait +from contextlib import contextmanager from dataclasses import dataclass from typing import Any, cast @@ -42,6 +43,7 @@ ) from .structural_tag_policy import runtime_structural_tag_options from .thinking import runtime_default_thinking_mode +from .ttft import ttft_stage from .utils import ( PreprocessError, as_error_envelope, @@ -316,6 +318,25 @@ class SglangPreprocessWorkerResult: # a reasoning parser in that case so the pool path matches the inline # path byte-for-byte. effective_reasoning_parser_name: str | None = None + # Same-host monotonic timestamps cross the process boundary without + # serializing the native request context. + template_tokenize_intervals_ns: list[tuple[int, int]] | None = None + + +class _WorkerPreprocessTiming: + """Collect template boundaries in the preprocessing process pool.""" + + def __init__(self) -> None: + self.intervals_ns: list[tuple[int, int]] = [] + + @contextmanager + def start_ttft_stage(self, stage: str): + assert stage == "template_tokenize" + start_ns = time.monotonic_ns() + try: + yield + finally: + self.intervals_ns.append((start_ns, time.monotonic_ns())) def _init_worker( @@ -353,8 +374,10 @@ def _preprocess_worker( request: dict[str, Any], model_name: str, eos_token_ids: list[int] | None, + record_ttft_timing: bool = False, ) -> SglangPreprocessWorkerResult: """Preprocess a request in a worker process and return a picklable result.""" + timing = _WorkerPreprocessTiming() if record_ttft_timing else None pre = preprocess_chat_request( request, tokenizer=_w_tokenizer, @@ -366,6 +389,7 @@ def _preprocess_worker( structural_tag_mode=_w_structural_tag_mode, structural_tag_scope=_w_structural_tag_scope, structural_tag_schema=_w_structural_tag_schema, + context=timing, ) n = request.get("n", 1) @@ -397,6 +421,7 @@ def _preprocess_worker( force_reasoning=pre.force_reasoning, effective_reasoning_parser_name=effective_reasoning_parser_name, named_zero_arg_tool=pre.named_zero_arg_tool, + template_tokenize_intervals_ns=timing.intervals_ns if timing else None, ) @@ -618,18 +643,20 @@ async def _generator_inner( if self.debug_perf: t0 = time.monotonic() - pre = preprocess_chat_request( - request, - tokenizer=self.tokenizer, - tool_call_parser_name=self.tool_call_parser_name, - reasoning_parser_name=self.reasoning_parser_name, - exclude_tools_when_tool_choice_none=self.exclude_tools_when_tool_choice_none, - template_force_reasoning=self.template_force_reasoning, - default_thinking_mode=self.default_thinking_mode, - structural_tag_mode=self.structural_tag_mode, - structural_tag_scope=self.structural_tag_scope, - structural_tag_schema=self.structural_tag_schema, - ) + with ttft_stage(context, "python_preprocess"): + pre = preprocess_chat_request( + request, + tokenizer=self.tokenizer, + tool_call_parser_name=self.tool_call_parser_name, + reasoning_parser_name=self.reasoning_parser_name, + exclude_tools_when_tool_choice_none=self.exclude_tools_when_tool_choice_none, + template_force_reasoning=self.template_force_reasoning, + default_thinking_mode=self.default_thinking_mode, + structural_tag_mode=self.structural_tag_mode, + structural_tag_scope=self.structural_tag_scope, + structural_tag_schema=self.structural_tag_schema, + context=context, + ) if self.debug_perf: t1 = time.monotonic() @@ -702,17 +729,36 @@ async def _generator_inner_pool( # --- Phase 1: Preprocess (semaphore held) --- assert self._worker_semaphore is not None assert self.preprocess_pool is not None + record_stage = getattr(context, "record_ttft_stage_ago", None) try: - async with self._worker_semaphore: - future = self.preprocess_pool.submit( - _preprocess_worker, - request, - request["model"], - self.eos_token_ids, - ) - preproc_result: SglangPreprocessWorkerResult = ( - await asyncio.wrap_future(future) - ) + with ttft_stage(context, "python_preprocess"): + with ttft_stage(context, "preprocess_queue"): + await self._worker_semaphore.acquire() + try: + future = self.preprocess_pool.submit( + _preprocess_worker, + request, + request["model"], + self.eos_token_ids, + record_stage is not None, + ) + preproc_result: SglangPreprocessWorkerResult = ( + await asyncio.wrap_future(future) + ) + finally: + self._worker_semaphore.release() + if record_stage is not None: + for start_ns, end_ns in ( + preproc_result.template_tokenize_intervals_ns or () + ): + # monotonic_ns is shared across processes on this host. + # Re-anchor immediately before crossing into Rust. + now_ns = time.monotonic_ns() + record_stage( + "template_tokenize", + (now_ns - start_ns) / 1_000_000_000, + (now_ns - end_ns) / 1_000_000_000, + ) except InvalidArgument: raise except PreprocessError as exc: diff --git a/components/src/dynamo/frontend/tests/test_ttft_stages.py b/components/src/dynamo/frontend/tests/test_ttft_stages.py new file mode 100644 index 000000000000..481fc8c65966 --- /dev/null +++ b/components/src/dynamo/frontend/tests/test_ttft_stages.py @@ -0,0 +1,588 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Frontend TTFT boundaries with lightweight backend and native-binding stubs.""" + +import asyncio +import importlib +import pickle +from concurrent.futures import Future +from contextlib import contextmanager +from pathlib import Path +from types import SimpleNamespace + +import pytest +from test_sglang_processor_metrics_unit import _load_processor_module +from test_sglang_processor_metrics_unit import module_stubs as module_stubs + +pytestmark = [ + pytest.mark.unit, + pytest.mark.gpu_0, + pytest.mark.pre_merge, + pytest.mark.core, +] + + +class RecordingContext: + def __init__(self): + self.events = [] + self.intervals = [] + + @contextmanager + def start_ttft_stage(self, stage): + self.events.append(("start", stage)) + try: + yield + finally: + self.events.append(("end", stage)) + + def record_ttft_stage_ago(self, stage, start, end): + self.intervals.append((stage, start, end)) + + +class Tokenizer: + chat_template = "{{ messages }}" + bos_token_id = 1 + + def apply_chat_template(self, messages, **kwargs): + return [1, 2, 3] + + def encode(self, text): + return [4, 5] + + +def chat_request(): + return {"model": "test", "messages": [{"role": "user", "content": "hello"}]} + + +@pytest.fixture +def frontend_stubs(request): + install, remove = request.getfixturevalue("module_stubs") + # Load only the pure Python utilities used here; common.utils.__init__ + # otherwise imports unrelated runtime-backed endpoint helpers. + install( + "dynamo.common.utils", + __path__=[str(Path(__file__).parents[2] / "common" / "utils")], + ) + install("dynamo._internal", ModelDeploymentCard=object) + install("dynamo.frontend.frontend_args", FrontendConfig=object) + install( + "dynamo.llm", + ModelCardInstanceId=object, + PythonAsyncEngine=object, + RoutedEngine=object, + ) + install( + "dynamo.llm.exceptions", + HttpError=type("HttpError", (Exception,), {}), + InvalidArgument=type("InvalidArgument", (Exception,), {}), + Unknown=type("Unknown", (Exception,), {}), + ) + # Restore all frontend modules imported against these stubs after each test. + for name in ( + "utils", + "sglang_prepost", + "sglang_processor", + "prepost", + "vllm_processor", + ): + remove("dynamo.frontend." + name) + for name in ("dynamo.common.utils.guided_json", "dynamo.vllm.errors"): + remove(name) + return install, remove + + +@pytest.fixture +def sglang_modules(frontend_stubs, monkeypatch): + processor = _load_processor_module(frontend_stubs) + _, remove = frontend_stubs + importlib.import_module("dynamo.llm").HttpError = importlib.import_module( + "dynamo.llm.exceptions" + ).HttpError + remove("dynamo.frontend.sglang_prepost") + prepost = importlib.import_module("dynamo.frontend.sglang_prepost") + monkeypatch.setattr( + processor, "preprocess_chat_request", prepost.preprocess_chat_request + ) + monkeypatch.setattr( + processor, "SglangStreamingPostProcessor", lambda **kwargs: object() + ) + + async def stream(*args, **kwargs): + yield {"ok": True} + + monkeypatch.setattr(processor.SglangProcessor, "_generate_and_stream", stream) + return processor, prepost + + +def make_sglang_processor(module, **kwargs): + return module.SglangProcessor( + tokenizer=Tokenizer(), + routed_engine=None, + tool_call_parser_name=None, + reasoning_parser_name=None, + eos_token_ids=None, + **kwargs, + ) + + +def test_sglang_inline_preprocess_and_legacy_context(sglang_modules): + module, _ = sglang_modules + processor = make_sglang_processor(module) + context = RecordingContext() + + async def collect(context): + return [ + item async for item in processor.generator(chat_request(), context=context) + ] + + assert asyncio.run(collect(context)) == [{"ok": True}] + assert context.events == [ + ("start", "python_preprocess"), + ("start", "template_tokenize"), + ("end", "template_tokenize"), + ("end", "python_preprocess"), + ] + assert asyncio.run(collect(None)) == [{"ok": True}] + assert asyncio.run(collect(object())) == [{"ok": True}] + + +def test_sglang_template_error_closes_spans(sglang_modules): + from jinja2.exceptions import TemplateError + + module, _ = sglang_modules + context = RecordingContext() + + class BrokenTokenizer(Tokenizer): + def apply_chat_template(self, *args, **kwargs): + raise TemplateError("bad template") + + processor = make_sglang_processor(module) + processor.tokenizer = BrokenTokenizer() + + async def collect(): + return [ + item async for item in processor.generator(chat_request(), context=context) + ] + + with pytest.raises(module.InvalidArgument, match="bad template"): + asyncio.run(collect()) + assert context.events[-2:] == [ + ("end", "template_tokenize"), + ("end", "python_preprocess"), + ] + + +def test_sglang_deepseek_custom_encoder_is_timed( + sglang_modules, frontend_stubs, monkeypatch +): + _, prepost = sglang_modules + install, _ = frontend_stubs + context = RecordingContext() + + def encode_messages(*args, **kwargs): + assert context.events[-1] == ("start", "template_tokenize") + return "rendered" + + install( + "sglang.srt.entrypoints.openai.encoding_dsv4", encode_messages=encode_messages + ) + monkeypatch.setattr( + prepost, "_should_use_deepseek_v4_encoding", lambda *args, **kwargs: True + ) + result = prepost.preprocess_chat_request( + chat_request(), + tokenizer=Tokenizer(), + tool_call_parser_name=None, + reasoning_parser_name=None, + context=context, + ) + assert result.prompt_token_ids == [4, 5] + assert context.events == [ + ("start", "template_tokenize"), + ("end", "template_tokenize"), + ] + + +def test_sglang_worker_returns_picklable_monotonic_intervals( + sglang_modules, monkeypatch +): + module, _ = sglang_modules + monkeypatch.setattr(module, "_w_tokenizer", Tokenizer()) + clock = iter([10_000_000_000, 10_200_000_000]) + monkeypatch.setattr( + module, "time", SimpleNamespace(monotonic_ns=lambda: next(clock)) + ) + result = module._preprocess_worker(chat_request(), "test", None, True) + result = pickle.loads(pickle.dumps(result)) + assert result.template_tokenize_intervals_ns == [(10_000_000_000, 10_200_000_000)] + assert result.prompt_token_ids == [1, 2, 3] + assert ( + module._preprocess_worker( + chat_request(), "test", None + ).template_tokenize_intervals_ns + is None + ) + + +def test_sglang_pool_reattaches_worker_intervals(sglang_modules, monkeypatch): + module, _ = sglang_modules + context = RecordingContext() + result = module.SglangPreprocessWorkerResult( + prompt_token_ids=[1], + dynamo_preproc={"sampling_options": {"guided_decoding": None}}, + request=chat_request(), + template_tokenize_intervals_ns=[(10_000_000_000, 10_200_000_000)], + ) + + class Pool: + def submit(self, fn, *args): + assert args[-1] is True + assert context not in args + assert context.events[-1] == ("end", "preprocess_queue") + future = Future() + future.set_result(result) + return future + + monkeypatch.setattr( + module, "time", SimpleNamespace(monotonic_ns=lambda: 11_000_000_000) + ) + processor = make_sglang_processor(module, preprocess_pool=Pool()) + + async def collect(): + return [ + item async for item in processor.generator(chat_request(), context=context) + ] + + assert asyncio.run(collect()) == [{"ok": True}] + assert context.intervals == [("template_tokenize", 1.0, 0.8)] + assert context.events == [ + ("start", "python_preprocess"), + ("start", "preprocess_queue"), + ("end", "preprocess_queue"), + ("end", "python_preprocess"), + ] + assert processor._worker_semaphore._value == 2 + + +@pytest.mark.parametrize("queued", [False, True]) +def test_sglang_pool_cancellation_preserves_semaphore(sglang_modules, queued): + module, _ = sglang_modules + context = RecordingContext() + + class Pool: + def submit(self, *args): + assert not queued + return Future() + + processor = make_sglang_processor(module, preprocess_pool=Pool()) + initial_capacity = 0 if queued else 2 + processor._worker_semaphore = asyncio.Semaphore(initial_capacity) + + async def cancel(): + task = asyncio.create_task( + anext(processor.generator(chat_request(), context=context)) + ) + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(cancel()) + assert context.events[-1] == ("end", "python_preprocess") + assert processor._worker_semaphore._value == initial_capacity + + +@pytest.fixture +def vllm_modules(frontend_stubs): + install, _ = frontend_stubs + dummy = type("BackendType", (), {}) + install("vllm") + install( + "vllm.config", + CacheConfig=dummy, + LoadConfig=dummy, + ModelConfig=dummy, + VllmConfig=dummy, + ) + install( + "vllm.entrypoints.chat_utils", + make_tool_call_id=lambda: "call", + load_chat_template=lambda value: value, + ) + install( + "vllm.entrypoints.openai.chat_completion.protocol", + ChatCompletionNamedFunction=dummy, + ChatCompletionNamedToolChoiceParam=dummy, + ChatCompletionRequest=dummy, + ) + install("vllm.reasoning", ReasoningParser=dummy, ReasoningParserManager=dummy) + install("vllm.renderers", ChatParams=dummy, merge_kwargs=lambda *args: {}) + install( + "vllm.sampling_params", + SamplingParams=SimpleNamespace, + RequestOutputKind=SimpleNamespace(DELTA=0), + ) + install("vllm.tokenizers", TokenizerLike=dummy) + install("vllm.tool_parsers", ToolParser=dummy, ToolParserManager=dummy) + install("vllm.tool_parsers.utils", get_json_schema_from_tools=lambda *args: None) + install("vllm.utils.async_utils", make_async=lambda fn, **kwargs: fn) + install("vllm.tasks", GENERATION_TASKS=("generate",)) + install( + "vllm.v1.engine", + EngineCoreOutput=dummy, + EngineCoreRequest=dummy, + FinishReason=type("FinishReason", (), {"STOP": 0, "LENGTH": 1, "ABORT": 2}), + ) + install("vllm.v1.engine.input_processor", InputProcessor=dummy) + install( + "vllm.v1.engine.output_processor", + OutputProcessor=dummy, + OutputProcessorOutput=dummy, + ) + install("vllm.v1.engine.parallel_sampling", ParentRequest=dummy) + install( + "vllm.exceptions", + VLLMClientError=type("VLLMClientError", (Exception,), {}), + VLLMNotFoundError=type("VLLMNotFoundError", (Exception,), {}), + VLLMUnprocessableEntityError=type( + "VLLMUnprocessableEntityError", (Exception,), {} + ), + ) + install("msgspec.structs", replace=lambda value, **kwargs: value) + install( + "dynamo.frontend.vllm_protocol", + DeltaFunctionCall=dummy, + DeltaMessage=dummy, + DeltaToolCall=dummy, + FunctionDefinition=dummy, + ) + install( + "dynamo.common.multimodal.mm_kwargs_transfer", + MmKwargsSender=dummy, + MmKwargsShmSender=dummy, + MmKwargsNixlSender=dummy, + ) + install( + "dynamo.common.multimodal.routing_utils", + build_mm_routing_info_from_features=lambda *args, **kwargs: { + "block_mm_infos": [[]] + }, + ) + importlib.import_module("dynamo.llm").HttpError = importlib.import_module( + "dynamo.llm.exceptions" + ).HttpError + prepost = importlib.import_module("dynamo.frontend.prepost") + processor = importlib.import_module("dynamo.frontend.vllm_processor") + return processor, prepost + + +@pytest.mark.parametrize("pretokenized", [False, True]) +@pytest.mark.parametrize("fail_render", [False, True]) +def test_vllm_template_tokenizer_preserves_tokens_and_closes_on_error( + vllm_modules, monkeypatch, pretokenized, fail_render +): + _, prepost = vllm_modules + context = RecordingContext() + request = SimpleNamespace( + tool_choice="auto", + structured_outputs=None, + add_special_tokens=True, + extract_structured_outputs=lambda: None, + ) + monkeypatch.setattr( + prepost, "_validate_chat_completion_request", lambda value: request + ) + monkeypatch.setattr(prepost, "_build_assistant_guided_decoding", lambda value: None) + monkeypatch.setattr(prepost, "_has_explicit_output_constraint", lambda value: False) + monkeypatch.setattr( + prepost, + "_prepare_request", + lambda *args, **kwargs: (request, None, {}, [], None), + ) + monkeypatch.setattr( + prepost, "build_tool_call_guided_decoding", lambda *args, **kwargs: None + ) + + class Renderer: + async def render_messages_async(self, messages, params): + assert context.events[-1] == ("start", "template_tokenize") + if fail_render: + raise ValueError("render failed") + return [], {"prompt_token_ids": [1, 2]} if pretokenized else { + "prompt": "hello" + } + + async def tokenize(prompt, **kwargs): + assert not pretokenized + assert context.events[-1] == ("start", "template_tokenize") + return SimpleNamespace(input_ids=[1, 2]) + + monkeypatch.setattr(prepost, "_get_async_tokenizer", lambda tokenizer: tokenize) + coro = prepost.preprocess_chat_request( + chat_request(), + tokenizer=Tokenizer(), + renderer=Renderer(), + tool_parser_class=None, + context=context, + ) + if fail_render: + with pytest.raises(ValueError, match="render failed"): + asyncio.run(coro) + else: + assert asyncio.run(coro).prompt_token_ids == [1, 2] + assert context.events == [ + ("start", "template_tokenize"), + ("end", "template_tokenize"), + ] + + +@pytest.mark.parametrize("has_media", [False, True]) +def test_vllm_generator_forwards_timeline_and_times_input_preparation( + vllm_modules, monkeypatch, has_media +): + module, _ = vllm_modules + context = RecordingContext() + + class RequestForSampling(SimpleNamespace): + model_fields = frozenset() + + request_for_sampling = RequestForSampling( + max_completion_tokens=None, + max_tokens=1, + logprobs=False, + top_logprobs=None, + cache_salt=None, + mm_processor_kwargs=None, + ) + + async def preprocess(request, **kwargs): + assert kwargs["context"] is context + assert context.events[-1] == ("start", "python_preprocess") + return SimpleNamespace( + request_for_sampling=request_for_sampling, + tool_parser=None, + chat_template_kwargs={}, + engine_prompt={}, + prompt_token_ids=[1], + guided_decoding=None, + ) + + monkeypatch.setattr(module, "preprocess_chat_request", preprocess) + monkeypatch.setattr( + module, + "extract_mm_urls", + lambda messages: ({"image_url": ["image"]} if has_media else None, None), + ) + expected_stage = "multimodal_prepare" if has_media else "python_preprocess" + + class ReachedInputPreparation(Exception): + pass + + async def build_inputs(*args, **kwargs): + assert context.events[-1] == ("start", expected_stage) + raise ReachedInputPreparation() + + monkeypatch.setattr(module, "_build_engine_inputs", build_inputs) + processor = module.VllmProcessor( + tokenizer=Tokenizer(), + input_processor=SimpleNamespace( + renderer=None, model_config=None, generation_config_fields={} + ), + output_processor=None, + tool_parser_class=None, + reasoning_parser_class=None, + routed_engine=None, + ) + + async def collect(): + return [ + item async for item in processor.generator(chat_request(), context=context) + ] + + with pytest.raises(ReachedInputPreparation): + asyncio.run(collect()) + assert context.events == [ + ("start", "python_preprocess"), + ("end", "python_preprocess"), + ("start", expected_stage), + ("end", expected_stage), + ] + + +@pytest.mark.parametrize("fail_transfer", [False, True]) +def test_vllm_mm_routing_and_transfer_have_separate_intervals( + vllm_modules, monkeypatch, fail_transfer +): + module, _ = vllm_modules + context = RecordingContext() + + def build_metadata(*args, **kwargs): + assert context.events[-1] == ("start", "mm_routing_build") + return {"block_mm_infos": [[]]} + + monkeypatch.setattr(module, "build_mm_routing_info_from_features", build_metadata) + feature = SimpleNamespace( + modality="image", + mm_hash="hash", + mm_position=SimpleNamespace(offset=1, length=2), + ) + preprocessed = SimpleNamespace(mm_features=[feature], prompt_token_ids=[1, 2]) + + class Sender: + async def prepare(self, *args, **kwargs): + assert context.events[-1] == ("start", "multimodal_prepare") + if fail_transfer: + raise ValueError("transfer failed") + return {"transport": "shm"}, ["cleanup"] + + processor = module.VllmProcessor( + tokenizer=Tokenizer(), + input_processor=None, + output_processor=None, + tool_parser_class=None, + reasoning_parser_class=None, + routed_engine=None, + ) + processor.nixl_mm_enabled = True + processor._sender = Sender() + request = {} + routing, cleanup, transferred = asyncio.run( + processor._prepare_mm_routing( + preprocessed, + request, + context=context, + ) + ) + assert routing == {"block_mm_infos": [[]]} + assert transferred is not fail_transfer + assert cleanup == ([] if fail_transfer else ["cleanup"]) + assert request["extra_args"]["expanded_token_ids"] == [1, 2] + assert context.events == [ + ("start", "mm_routing_build"), + ("end", "mm_routing_build"), + ("start", "multimodal_prepare"), + ("end", "multimodal_prepare"), + ] + + +def test_vllm_text_and_uuid_only_paths_skip_mm_stages(vllm_modules): + module, _ = vllm_modules + processor = module.VllmProcessor( + tokenizer=Tokenizer(), + input_processor=None, + output_processor=None, + tool_parser_class=None, + reasoning_parser_class=None, + routed_engine=None, + ) + context = RecordingContext() + for request in ({}, {"multi_modal_uuids": {"image": ["cached"]}}): + assert asyncio.run( + processor._prepare_mm_routing( + SimpleNamespace(mm_features=[]), + request, + context=context, + ) + ) == (None, [], False) + assert context.events == [] diff --git a/components/src/dynamo/frontend/ttft.py b/components/src/dynamo/frontend/ttft.py new file mode 100644 index 000000000000..ee54a04211c5 --- /dev/null +++ b/components/src/dynamo/frontend/ttft.py @@ -0,0 +1,13 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Shared access to the HTTP request's process-local TTFT timeline.""" + +from contextlib import AbstractContextManager, nullcontext +from typing import Any + + +def ttft_stage(context: Any | None, stage: str) -> AbstractContextManager: + """Record a stage when the caller carries a native HTTP timing context.""" + start_stage = getattr(context, "start_ttft_stage", None) + return start_stage(stage) if start_stage is not None else nullcontext() diff --git a/components/src/dynamo/frontend/vllm_processor.py b/components/src/dynamo/frontend/vllm_processor.py index 1e2a75f20534..38795a65dcae 100644 --- a/components/src/dynamo/frontend/vllm_processor.py +++ b/components/src/dynamo/frontend/vllm_processor.py @@ -50,6 +50,7 @@ runtime_structural_tag_options, ) from .thinking import runtime_default_thinking_mode +from .ttft import ttft_stage from .utils import ( as_error_envelope, backend_invalid_argument_to_http_error, @@ -768,6 +769,7 @@ async def _prepare_mm_routing( vllm_preproc: EngineCoreRequest, dynamo_preproc: dict[str, Any], mm_processor_kwargs: dict[str, Any] | None = None, + context: Any | None = None, ) -> tuple[dict | None, list, bool]: """Extract MM routing info and prepare mm_kwargs transfer. @@ -794,70 +796,73 @@ async def _prepare_mm_routing( return None, cleanup_items, nixl_transferred if vllm_preproc.mm_features: - if mm_processor_kwargs: - # vLLM rehashes supplied UUIDs when processor kwargs are - # present. Fall back to text-prefix routing so the router and - # worker cannot publish different cache keys. - logger.debug( - "[mm-routing] Exact MM routing disabled because " - "mm_processor_kwargs is non-empty" - ) - else: - mm_routing_info = build_mm_routing_info_from_features( - vllm_preproc.mm_features, - prompt_token_ids=list(vllm_preproc.prompt_token_ids), - ) - ( - mm_hashes_list, - mm_placeholders_list, - mm_hashes_by_modality, - mm_placeholders_by_modality, - ) = _group_mm_feature_metadata(vllm_preproc.mm_features) - if "extra_args" not in dynamo_preproc: - dynamo_preproc["extra_args"] = {} - # Forward hashes only when this frontend built the matching exact - # routing sequence. Their presence is the worker's request-time - # readiness signal; transfer-only metadata is deliberately separate. - if mm_routing_info is not None: - dynamo_preproc["extra_args"]["mm_hashes"] = mm_hashes_list - dynamo_preproc["extra_args"][ - "mm_hashes_by_modality" - ] = mm_hashes_by_modality - dynamo_preproc["extra_args"]["mm_placeholders"] = mm_placeholders_list - dynamo_preproc["extra_args"][ - "mm_placeholders_by_modality" - ] = mm_placeholders_by_modality - # Forward the expanded prompt_token_ids (with image placeholders) - # so the backend can use them in the pre-rendered MultiModalInput. - dynamo_preproc["extra_args"]["expanded_token_ids"] = list( - vllm_preproc.prompt_token_ids - ) - - n_blocks = len(mm_routing_info["block_mm_infos"]) if mm_routing_info else 0 - n_mm_blocks = sum( - 1 for b in (mm_routing_info or {}).get("block_mm_infos", []) if b - ) - logger.debug( - "[mm-routing] Built mm_routing_info: %d mm_features, " - "%d hashes, %d total blocks, %d blocks with MM content, " - "block_size=%d", - len(vllm_preproc.mm_features), - len(mm_hashes_list), - n_blocks, - n_mm_blocks, - self.block_size, - ) - if logger.isEnabledFor(logging.DEBUG): - for i, f in enumerate(vllm_preproc.mm_features): + with ttft_stage(context, "mm_routing_build"): + if mm_processor_kwargs: + # vLLM rehashes supplied UUIDs when processor kwargs are + # present. Fall back to text-prefix routing so the router and + # worker cannot publish different cache keys. logger.debug( - "[mm-routing] feature[%d]: modality=%s, hash=%s..., " - "offset=%d, length=%d", - i, - _mm_feature_modality(f), - f.mm_hash[:16] if f.mm_hash else "None", - f.mm_position.offset, - f.mm_position.length, + "[mm-routing] Exact MM routing disabled because " + "mm_processor_kwargs is non-empty" + ) + else: + mm_routing_info = build_mm_routing_info_from_features( + vllm_preproc.mm_features, + prompt_token_ids=list(vllm_preproc.prompt_token_ids), ) + ( + mm_hashes_list, + mm_placeholders_list, + mm_hashes_by_modality, + mm_placeholders_by_modality, + ) = _group_mm_feature_metadata(vllm_preproc.mm_features) + if "extra_args" not in dynamo_preproc: + dynamo_preproc["extra_args"] = {} + # Forward hashes only when this frontend built the matching exact + # routing sequence. Their presence is the worker's request-time + # readiness signal; transfer-only metadata is deliberately separate. + if mm_routing_info is not None: + dynamo_preproc["extra_args"]["mm_hashes"] = mm_hashes_list + dynamo_preproc["extra_args"][ + "mm_hashes_by_modality" + ] = mm_hashes_by_modality + dynamo_preproc["extra_args"]["mm_placeholders"] = mm_placeholders_list + dynamo_preproc["extra_args"][ + "mm_placeholders_by_modality" + ] = mm_placeholders_by_modality + # Forward the expanded prompt_token_ids (with image placeholders) + # so the backend can use them in the pre-rendered MultiModalInput. + dynamo_preproc["extra_args"]["expanded_token_ids"] = list( + vllm_preproc.prompt_token_ids + ) + + n_blocks = ( + len(mm_routing_info["block_mm_infos"]) if mm_routing_info else 0 + ) + n_mm_blocks = sum( + 1 for b in (mm_routing_info or {}).get("block_mm_infos", []) if b + ) + logger.debug( + "[mm-routing] Built mm_routing_info: %d mm_features, " + "%d hashes, %d total blocks, %d blocks with MM content, " + "block_size=%d", + len(vllm_preproc.mm_features), + len(mm_hashes_list), + n_blocks, + n_mm_blocks, + self.block_size, + ) + if logger.isEnabledFor(logging.DEBUG): + for i, f in enumerate(vllm_preproc.mm_features): + logger.debug( + "[mm-routing] feature[%d]: modality=%s, hash=%s..., " + "offset=%d, length=%d", + i, + _mm_feature_modality(f), + f.mm_hash[:16] if f.mm_hash else "None", + f.mm_position.offset, + f.mm_position.length, + ) # Transfer pre-processed mm_kwargs to the backend so it can skip # the HF processor. Strategy: @@ -889,10 +894,11 @@ async def _prepare_mm_routing( ) # NVTX annotation is owned by MmKwargsSender.prepare via # the subclass's _nvtx_label/_nvtx_color class attrs. - extra_update, cleanup_items = await self._sender.prepare( - vllm_preproc.mm_features, - modality=transfer_modality, - ) + with ttft_stage(context, "multimodal_prepare"): + extra_update, cleanup_items = await self._sender.prepare( + vllm_preproc.mm_features, + modality=transfer_modality, + ) if extra_update is not None: dynamo_preproc["extra_args"].update(extra_update) nixl_transferred = True @@ -952,7 +958,10 @@ async def _generator_inner( # Images are fetched by vLLM's renderer via DynamoMediaConnector, # which wraps our ImageLoader (LRU cache + in-flight dedup). # No data URI encoding needed. - with _nvtx.annotate("mm_frontend:preprocess_chat", color="yellow"): + with ( + _nvtx.annotate("mm_frontend:preprocess_chat", color="yellow"), + ttft_stage(context, "python_preprocess"), + ): pre = await preprocess_chat_request( request, tokenizer=self.tokenizer, @@ -968,6 +977,7 @@ async def _generator_inner( structural_tag_scope=self.structural_tag_scope, structural_tag_schema=self.structural_tag_schema, structural_tag_excludes_reasoning=self.structural_tag_excludes_reasoning, + context=context, ) request_for_sampling = pre.request_for_sampling @@ -1025,7 +1035,13 @@ async def _generator_inner( # legacy nvext passthrough. thinking_token_budget = resolve_thinking_token_budget(request) - with _nvtx.annotate("mm_frontend:process_inputs", color="orange"): + with ( + _nvtx.annotate("mm_frontend:process_inputs", color="orange"), + ttft_stage( + context, + "multimodal_prepare" if mm_data or mm_uuids else "python_preprocess", + ), + ): # render_messages_async returns a raw prompt. Convert it to a typed # EngineInput before process_inputs. User UUID requests deliberately # stay token-only here: the selected worker must populate and query @@ -1118,6 +1134,7 @@ async def _generator_inner( vllm_preproc, dynamo_preproc, mm_processor_kwargs=request_for_sampling.mm_processor_kwargs, + context=context, ) # Forward multimodal URLs so the backend handler can load the media. diff --git a/docs/fern/pages/reference/components/frontend-configuration.mdx b/docs/fern/pages/reference/components/frontend-configuration.mdx index 8d6729bf3707..85adde8ca8d8 100644 --- a/docs/fern/pages/reference/components/frontend-configuration.mdx +++ b/docs/fern/pages/reference/components/frontend-configuration.mdx @@ -592,6 +592,7 @@ Most frontend latency and size histograms use log-spaced buckets, rounded to two | `dynamo_frontend_input_sequence_tokens` | `DYN_METRICS_INPUT_SEQUENCE` | `50.0` | `128000.0` | `12` | `130000` | | `dynamo_frontend_output_sequence_tokens` | `DYN_METRICS_OUTPUT_SEQUENCE` | `50.0` | `32000.0` | `10` | `32000` | | `dynamo_frontend_time_to_first_token_seconds` | `DYN_METRICS_TTFT` | `0.001` | `480.0` | `18` | `480` | +| `dynamo_frontend_ttft_stage_seconds` | `DYN_METRICS_TTFT_STAGE` | `0.000001` | `480.0` | `26` | `480` | | `dynamo_frontend_inter_token_latency_seconds` | `DYN_METRICS_ITL` | `0.001` | `80.0` | `20` | `80` | | `dynamo_frontend_embedding_latency_seconds` | `DYN_METRICS_EMBEDDING_LATENCY` | `0.001` | `10.0` | `14` | `10` | @@ -606,6 +607,12 @@ Rounding can move the top edge either way, so pick a `MAX` that is already two s `dynamo_frontend_cached_tokens` shares the input sequence buckets, so `DYN_METRICS_INPUT_SEQUENCE_*` changes both histograms. +`dynamo_frontend_ttft_stage_seconds` has 17 fixed `stage`/`phase` combinations per model. Each +combination has the same bucket boundaries. With the default 26 finite boundaries, the `_bucket`, +`_sum`, and `_count` series total 493 per model, including the `+Inf` bucket. See +[TTFT stage durations](../observability/metrics-catalog.mdx#ttft-stage-durations) for the stage +contract and additive mean queries. + Set `DYN_METRICS_ITL_BUCKETS` to an explicit, comma-separated list of ITL boundaries in seconds. The list must contain 1 to 512 finite, non-negative, strictly increasing values and be at most 16,384 bytes long. Boundaries are used as supplied, without logarithmic spacing or rounding. For example, to use explicit boundaries for an SGLang deployment: diff --git a/docs/fern/pages/reference/observability/logging.mdx b/docs/fern/pages/reference/observability/logging.mdx index ce5fdbcdac48..da9fc1a589a3 100644 --- a/docs/fern/pages/reference/observability/logging.mdx +++ b/docs/fern/pages/reference/observability/logging.mdx @@ -63,6 +63,41 @@ For example, keep the default at `info` and enable trace output for the system-s export DYN_LOG='info,dynamo_runtime::system_status_server=trace' ``` +## Per-Request First-Token Log + +For HTTP chat and text completion requests, the frontend emits one INFO event when it first observes a +nonzero output token count. The target is `dynamo_frontend::ttft` and the message is +`frontend first token`. Requests that end, fail, or are canceled before that observation do not +emit the event. Later tokens do not emit additional TTFT events. + +The event contains the canonical HTTP context `request_id`, `model`, `ttft_ms`, and 17 numeric stage +durations in milliseconds. Field names use `_ms` for an empty phase and +`__ms` otherwise, such as `template_tokenize_ms`, `multimodal_prepare_ms`, +`kv_route_prefill_ms`, `backend_first_response_decode_ms`, and +`backend_overlap_prefill_decode_ms`. All stage/phase combinations are present, including zeros for +unused stages. These frozen values match the +[TTFT stage histogram](metrics-catalog.mdx#ttft-stage-durations), and their sum equals `ttft_ms`, +subject to floating-point rounding. + +The default INFO level includes these events. To disable them while retaining metrics and other +INFO logs, set: + +```bash +export DYN_LOG='info,dynamo_frontend::ttft=off' +``` + +To retain these events with a WARN default, set: + +```bash +export DYN_LOG='warn,dynamo_frontend::ttft=info' +``` + +Set `DYN_LOGGING_CONSOLE_FORMAT=jsonl` for machine-readable records; duration fields remain JSON +numbers. Readable and JSONL formats use the same filters. TOML `log_filters` entries are applied +after `DYN_LOG`, so an entry for the same target overrides the environment directive. The file +selected by `DYN_LOGGING_CONFIG_PATH` overrides matching entries in `/opt/dynamo/etc/logging.toml`. +More specific target filters take precedence over broader prefixes. + ## OTLP Export `OTEL_EXPORT_ENABLED=true` is the master switch for runtime trace and log export. Configure a diff --git a/docs/fern/pages/reference/observability/metric-labels.mdx b/docs/fern/pages/reference/observability/metric-labels.mdx index 6f8d9731c736..10468fcd3c10 100644 --- a/docs/fern/pages/reference/observability/metric-labels.mdx +++ b/docs/fern/pages/reference/observability/metric-labels.mdx @@ -66,13 +66,20 @@ Added by an individual metric family. Each label appears only on the metrics not `[Shared]` On `dynamo_frontend_stage_requests` — the frontend pipeline stage a request currently occupies. See [Stage values](#stage-values). - Allowed values: preprocess route dispatch + On `dynamo_frontend_ttft_stage_seconds`, identifies an exclusive TTFT contribution. Its values + differ from the request gauge; see [TTFT stage durations](metrics-catalog.mdx#ttft-stage-durations). + + Request gauge values: preprocess route dispatch `[Shared]` On `dynamo_frontend_stage_requests` — the serving phase. See [Phase values](#phase-values). The empty value is used when the stage does not distinguish phases (e.g. `preprocess`). - Allowed values: prefill decode aggregated "" (empty) + On `dynamo_frontend_ttft_stage_seconds`, uses the same phase values and adds `prefill_decode` for + `backend_overlap`. See [TTFT stage durations](metrics-catalog.mdx#ttft-stage-durations) for the + allowed stage/phase pairs. + + Request gauge values: prefill decode aggregated "" (empty) diff --git a/docs/fern/pages/reference/observability/metrics-catalog.mdx b/docs/fern/pages/reference/observability/metrics-catalog.mdx index b9b6675da18f..9dbc3b661dec 100644 --- a/docs/fern/pages/reference/observability/metrics-catalog.mdx +++ b/docs/fern/pages/reference/observability/metrics-catalog.mdx @@ -119,6 +119,117 @@ The ratio is defined only when the model has an active L1 cache and observed tok Time to first token in seconds. + + Exclusive wall-clock contributions to frontend Time To First Token (TTFT), in seconds. Labeled by + `model`, `stage`, and `phase`. Every request that records total TTFT records all 17 stage/phase + combinations, including zeros for unused stages. See [TTFT stage durations](#ttft-stage-durations). + + +### TTFT stage durations + +`dynamo_frontend_ttft_stage_seconds` partitions the same interval as +`dynamo_frontend_time_to_first_token_seconds`: from the HTTP request metrics start to the first +observed nonzero output token count. Durations are clipped to that interval, concurrent work is +counted once, and gaps are assigned to `frontend_overhead`. The 17 stage/phase contributions sum to +the request's total TTFT, subject to floating-point rounding. + +| `stage` | `phase` | Measured interval | +|---------|---------|-------------------| +| `frontend_overhead` | `""` | Time outside instrumented stages, including request setup and gaps between stages | +| `python_preprocess` | `""` | Python chat preprocessing, including process-pool dispatch and IPC when used, excluding separately attributed nested stages | +| `preprocess_queue` | `""` | Waiting for the Python preprocessing concurrency semaphore | +| `template_tokenize` | `""` | Chat-template application and tokenization in the Python or native Rust preprocessor | +| `multimodal_prepare` | `""` | Frontend multimodal input preparation, including media processing and shared-memory or NIXL sender preparation | +| `mm_routing_build` | `""` | Building multimodal routing information from prepared inputs | +| `kv_route` | `prefill`, `decode`, `aggregated` | Worker selection, scheduling, and dispatch preparation for the backend phase, for KV and built-in routing policies | +| `backend_first_response` | `prefill`, `decode`, `aggregated` | Backend dispatch to the first response, including transport, backend work, and waiting | +| `backend_first_token` | `prefill`, `decode`, `aggregated` | First backend response to its first nonempty token output; zero when the first response contains tokens | +| `backend_overlap` | `prefill_decode` | Concurrent prefill and decode backend waits, counted once | +| `first_token_postprocess` | `""` | First decode or aggregated backend token to the HTTP first-token observation, including detokenization, parser buffering, and delivery | + +The empty `phase` means the stage does not distinguish serving phases. `aggregated` identifies a +single worker serving both prefill and decode. `backend_overlap` includes first-response and +first-token waits. Routing takes precedence over backend waits, so a routing interval does not also +contribute to a backend wait or overlap stage. Template/tokenizer, multimodal preparation, routing +metadata, and preprocessing queue intervals take precedence over `python_preprocess`. Routing +metadata construction takes precedence over an enclosing `multimodal_prepare` interval. Once a +decode or aggregated worker produces a token, `first_token_postprocess` takes precedence until the +HTTP first-token observation, including any remaining overlapping work. If that attempt is +aborted or retried before HTTP observes a token, its postprocessing interval closes so the +replacement attempt retains its own routing and backend contributions. + +The metric is enabled by default on the frontend's existing `/metrics` endpoint. The SGLang and +vLLM Python chat processors and native Rust preprocessor provide preprocessing timings. Chat and +text completions carry the HTTP timing context through the in-process routing host, including KV, +round-robin, random, direct, and other built-in policies. These paths record worker selection, +backend waits, and first-token postprocessing. The `kv_route` label also covers the built-in +policies. Batched text completion children share the parent HTTP timeline, so overlapping child +work is counted once. + +The timing context stays within one frontend process; a standalone router reached over the +network does not receive it. Other HTTP endpoints and uninstrumented operations retain their time +in `frontend_overhead`. A large value does not establish that the frontend consumed that time on +CPU. + +All boundaries use the frontend's local timing context. Process-pool template/tokenizer intervals +are aligned to the frontend clock using elapsed ages, with a small boundary uncertainty from the +Python-to-Rust call overhead. No worker clock synchronization or timing fields in backend RPCs are +required. Backend stages include transport, scheduling, and processing; they do not isolate GPU +prefill, decode computation, or KV transfer. A control response can end `backend_first_response` +before a token is available, which is why `backend_first_token` is separate. These values are +exclusive contributions to HTTP TTFT, not independent elapsed times for overlapping operations. +Processor boundaries differ by backend: vLLM chat rendering can fetch media inside +`template_tokenize`, while later media processing and transfer preparation contribute to +`multimodal_prepare`. + +For each model, all 17 combinations have the same sample count and request population as total +TTFT. Requests that end, fail, or are canceled before the first output token produce neither total +nor stage TTFT observations. Later tokens do not add TTFT observations. Labels never contain +request IDs. See [Per-Request First-Token Log](logging.mdx#per-request-first-token-log) for the +corresponding per-request event. + +For a stacked chart of mean stage durations in milliseconds, use: + +```promql +1000 * ( + sum by (model, stage, phase) ( + rate(dynamo_frontend_ttft_stage_seconds_sum[$__rate_interval]) + ) + / + sum by (model, stage, phase) ( + rate(dynamo_frontend_ttft_stage_seconds_count[$__rate_interval]) + ) +) +``` + +Set the legend to `{{model}} / {{stage}} / {{phase}}` and select one model per stacked panel. Apply +the same deployment selectors when comparing with mean total TTFT: + +```promql +1000 * ( + sum by (model) ( + rate(dynamo_frontend_time_to_first_token_seconds_sum[$__rate_interval]) + ) + / + sum by (model) ( + rate(dynamo_frontend_time_to_first_token_seconds_count[$__rate_interval]) + ) +) +``` + +Zero observations preserve the common denominator, so the sum of stage means matches mean total +TTFT over the same requests. Stage P95 or P99 values are not additive: different requests can +determine each percentile, and unused stages contribute zero observations. Use total TTFT +percentiles separately to assess tail latency. A scrape overlapping metric updates or differing +series scrape histories can cause small temporary differences between the sums. + +The default stage histogram has 26 finite buckets, including zero, plus `+Inf`. Set +`DYN_METRICS_TTFT_STAGE_MIN`, `DYN_METRICS_TTFT_STAGE_MAX`, and `DYN_METRICS_TTFT_STAGE_COUNT` before +frontend startup to customize the logarithmic boundaries. Bucket resolution affects percentile +estimates; `_sum` and `_count` retain the observed durations independently of the boundaries. See +[Histogram buckets](../components/frontend-configuration.mdx#histogram-buckets) for defaults, +validation, and cardinality considerations. + ### Migration Metrics for [request migration](../../kubernetes/fault-tolerance/request-migration.md), the mechanism that continues in-flight requests on a healthy worker when the original worker fails. See [Request Migration Architecture](../../developer-guide/knowledge-base/concepts/fault-tolerance/request-migration-architecture.md) for the internals. diff --git a/lib/bindings/python/rust/context.rs b/lib/bindings/python/rust/context.rs index 158a4487e008..7f0f4d2ecc99 100644 --- a/lib/bindings/python/rust/context.rs +++ b/lib/bindings/python/rust/context.rs @@ -11,8 +11,10 @@ // `set_status` operations mirror the OTel `Span` API. use dynamo_backend_common::FirstTokenNotifier; +use dynamo_llm::protocols::common::ttft::{TTFT_CONTEXT_KEY, TtftSpan, TtftStage, TtftTimings}; use dynamo_runtime::logging::DistributedTraceContext; pub use dynamo_runtime::pipeline::AsyncEngineContext; +use dynamo_runtime::pipeline::Context as PipelineContext; use dynamo_runtime::pipeline::context::Controller; use opentelemetry::global::BoxedSpan; use opentelemetry::trace::{Span as OtelSpan, Status, TraceContextExt, Tracer}; @@ -63,6 +65,7 @@ pub struct Context { trace_context: Option, first_token: Option, metadata: Arc>>, + ttft: Option>, /// Captured `engine.generate` span. `None` for Python-instantiated test /// contexts (where no parent span was plumbed in) — `current_span` / /// `start_span` return a no-op `SpanProxy` in that case. @@ -75,6 +78,40 @@ pub struct ContextMetadata { inner: Arc>>, } +fn python_ttft_stage(name: &str) -> PyResult { + TtftStage::from_python(name).ok_or_else(|| { + pyo3::exceptions::PyValueError::new_err(format!("unknown frontend TTFT stage: {name}")) + }) +} + +/// A small RAII context manager; never creates an OpenTelemetry span or log. +#[pyclass] +pub struct TtftStageGuard { + inner: Option, +} + +#[pymethods] +impl TtftStageGuard { + fn close(&mut self) { + self.inner.take(); + } + + fn __enter__(slf: PyRefMut) -> PyRefMut { + slf + } + + #[pyo3(signature = (_exc_type=None, _exc_value=None, _traceback=None))] + fn __exit__( + &mut self, + _exc_type: Option>, + _exc_value: Option>, + _traceback: Option>, + ) -> bool { + self.close(); + false + } +} + impl ContextMetadata { fn lock_map(&self) -> MutexGuard<'_, BTreeMap> { self.inner @@ -182,10 +219,25 @@ impl Context { trace_context, first_token, metadata: Arc::new(Mutex::new(metadata)), + ttft: None, span: None, } } + pub fn with_ttft_from( + mut self, + context: &PipelineContext, + ) -> Result { + self.ttft = context + .get_optional::>(TTFT_CONTEXT_KEY)? + .map(|timings| Arc::clone(timings.as_ref())); + Ok(self) + } + + pub fn ttft(&self) -> Option<&Arc> { + self.ttft.as_ref() + } + /// Attach the `engine.generate` span. Called by `PyLLMEngine::generate` /// after capturing `Span::current()` outside the spawn_blocking boundary. /// See `lib/bindings/python/rust/backend.rs` for the invariant. @@ -272,6 +324,7 @@ impl Context { trace_context: None, first_token: None, metadata: Arc::new(Mutex::new(metadata.unwrap_or_default())), + ttft: None, span: None, } } @@ -289,6 +342,7 @@ impl Context { trace_context: self.trace_context.clone(), first_token: None, metadata: Arc::new(Mutex::new(self.metadata_snapshot())), + ttft: self.ttft.clone(), span: self.span.clone(), } } @@ -297,6 +351,37 @@ impl Context { self.inner.is_stopped() } + /// Start a process-local frontend stage; no-op outside an HTTP timeline. + fn start_ttft_stage(&self, stage: &str) -> PyResult { + let stage = python_ttft_stage(stage)?; + Ok(TtftStageGuard { + inner: self.ttft.as_ref().map(|timings| timings.start_span(stage)), + }) + } + + /// Import pool-worker timings without serializing a native request context. + fn record_ttft_stage_ago( + &self, + stage: &str, + start_ago_seconds: f64, + end_ago_seconds: f64, + ) -> PyResult<()> { + let stage = python_ttft_stage(stage)?; + let start = std::time::Duration::try_from_secs_f64(start_ago_seconds) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("invalid start age"))?; + let end = std::time::Duration::try_from_secs_f64(end_ago_seconds) + .map_err(|_| pyo3::exceptions::PyValueError::new_err("invalid end age"))?; + if start < end { + return Err(pyo3::exceptions::PyValueError::new_err( + "start age precedes end age", + )); + } + if let Some(timings) = &self.ttft { + timings.record_ago(stage, start, end); + } + Ok(()) + } + fn is_killed(&self) -> bool { self.inner.is_killed() } diff --git a/lib/bindings/python/rust/engine.rs b/lib/bindings/python/rust/engine.rs index 2e3f239604a2..3ce081f5c1ef 100644 --- a/lib/bindings/python/rust/engine.rs +++ b/lib/bindings/python/rust/engine.rs @@ -308,16 +308,15 @@ where let current_trace_context = get_distributed_tracing_context(); let metadata = context.metadata().clone(); + let python_context = Context::new(ctx.clone(), current_trace_context, None, metadata) + .with_ttft_from(&context) + .map_err(Error::msg)?; let stream = invoke_generator( engine.generator.clone(), engine.event_loop.clone(), move |py| to_python_input(py, request), - engine.has_context.then_some({ - let ctx = ctx.clone(); - move |py: Python<'_>| { - Py::new(py, Context::new(ctx, current_trace_context, None, metadata)) - .map(|context| vec![("context", context.into_any())]) - } + engine.has_context.then_some(move |py: Python<'_>| { + Py::new(py, python_context).map(|context| vec![("context", context.into_any())]) }), ) .await?; diff --git a/lib/bindings/python/rust/lib.rs b/lib/bindings/python/rust/lib.rs index f355b526b791..9fe1c756cc69 100644 --- a/lib/bindings/python/rust/lib.rs +++ b/lib/bindings/python/rust/lib.rs @@ -379,6 +379,7 @@ fn register_core(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/lib/bindings/python/rust/llm/routed_engine.rs b/lib/bindings/python/rust/llm/routed_engine.rs index 47e228792ece..e0024cb96547 100644 --- a/lib/bindings/python/rust/llm/routed_engine.rs +++ b/lib/bindings/python/rust/llm/routed_engine.rs @@ -1,6 +1,8 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use std::sync::Arc; + use pyo3::prelude::*; use pythonize::{depythonize, pythonize}; use tokio_stream::StreamExt; @@ -9,6 +11,7 @@ use tracing_opentelemetry::OpenTelemetrySpanExt; use dynamo_llm::entrypoint::PrefillRoutedEngine; use dynamo_llm::protocols::common::preprocessor::PreprocessedRequest; +use dynamo_llm::protocols::common::timing::RequestTracker; use dynamo_runtime::logging::{DistributedTraceContext, otel_parent_context_from_distributed}; use dynamo_runtime::pipeline::{AsyncEngineContextProvider, SingleIn}; use dynamo_runtime::protocols::annotated::Annotated as RsAnnotated; @@ -26,6 +29,15 @@ impl RoutedEngine { } } +fn copy_ttft_tracker(parent_context: &crate::context::Context, request: &mut PreprocessedRequest) { + if let Some(ttft) = parent_context.ttft() { + let tracker = request + .tracker + .get_or_insert_with(|| Arc::new(RequestTracker::new())); + tracker.set_ttft(Arc::clone(ttft)); + } +} + #[pymethods] impl RoutedEngine { /// Send a preprocessed request through the Rust prefill-routed pipeline. @@ -36,8 +48,10 @@ impl RoutedEngine { preprocessed: PyObject, context: Option, ) -> PyResult> { - let request: PreprocessedRequest = depythonize(preprocessed.bind(py)).map_err(to_pyerr)?; + let mut request: PreprocessedRequest = + depythonize(preprocessed.bind(py)).map_err(to_pyerr)?; let request_context = if let Some(parent_context) = context.as_ref() { + copy_ttft_tracker(parent_context, &mut request); let parent_metadata = parent_context.metadata_snapshot(); let parent_context = parent_context.inner(); let child_context = SingleIn::with_id_and_metadata( @@ -140,6 +154,37 @@ mod tests { use opentelemetry::trace::{TraceContextExt, TraceId, TracerProvider as _}; use tracing_subscriber::layer::SubscriberExt; + #[test] + fn request_tracker_shares_http_timeline_across_python_boundary() { + use dynamo_llm::protocols::common::ttft::{TTFT_CONTEXT_KEY, TtftTimings}; + + let timeline = Arc::new(TtftTimings::new(std::time::Instant::now())); + let mut source = SingleIn::new(()); + source.insert(TTFT_CONTEXT_KEY, Arc::clone(&timeline)); + let python_context = + crate::context::Context::new(source.context(), None, None, Default::default()) + .with_ttft_from(&source) + .unwrap(); + let mut request: PreprocessedRequest = + serde_json::from_value(serde_json::json!({"token_ids": []})).unwrap(); + // Preserve any existing local tracker (and its other timing state). + let tracker = Arc::new(RequestTracker::new()); + request.tracker = Some(Arc::clone(&tracker)); + copy_ttft_tracker(&python_context, &mut request); + assert!(Arc::ptr_eq(&tracker, request.tracker.as_ref().unwrap())); + assert!(Arc::ptr_eq(&timeline, tracker.ttft().unwrap())); + assert!(Arc::ptr_eq( + &tracker, + request.clone().tracker.as_ref().unwrap(), + )); + // Process-local telemetry does not alter old/new worker wire schemas. + let serialized = serde_json::to_value(&request).unwrap(); + assert!(serialized.get("tracker").is_none()); + assert!(serialized.get("ttft").is_none()); + let round_trip: PreprocessedRequest = serde_json::from_value(serialized).unwrap(); + assert!(round_trip.tracker.is_none()); + } + fn make_trace_context( trace_id: &str, span_id: &str, diff --git a/lib/bindings/python/src/dynamo/_core.pyi b/lib/bindings/python/src/dynamo/_core.pyi index ba222dcd2a82..cf5164d47a7a 100644 --- a/lib/bindings/python/src/dynamo/_core.pyi +++ b/lib/bindings/python/src/dynamo/_core.pyi @@ -483,12 +483,31 @@ class FirstTokenSource: def bind(self, context: Context, dp_rank: Optional[int] = None) -> None: ... +class TtftStageGuard: + """Process-local TTFT interval. Closing is idempotent; exceptions propagate.""" + + def close(self) -> None: ... + def __enter__(self) -> TtftStageGuard: ... + def __exit__( + self, _exc_type: Any = None, _exc_value: Any = None, _traceback: Any = None + ) -> bool: ... + class Context: """ Context wrapper around AsyncEngineContext for Python bindings. Provides tracing and cancellation capabilities for request handling. """ + def start_ttft_stage(self, stage: str) -> TtftStageGuard: + """Start a frontend stage; no-op if no HTTP TTFT timeline is attached.""" + ... + + def record_ttft_stage_ago( + self, stage: str, start_ago_seconds: float, end_ago_seconds: float + ) -> None: + """Import worker-pool intervals using nonnegative monotonic ages.""" + ... + def __init__( self, id: Optional[str] = None, diff --git a/lib/bindings/python/src/dynamo/prometheus_names.py b/lib/bindings/python/src/dynamo/prometheus_names.py index 653144c48569..688770b4e53c 100644 --- a/lib/bindings/python/src/dynamo/prometheus_names.py +++ b/lib/bindings/python/src/dynamo/prometheus_names.py @@ -147,6 +147,8 @@ class frontend_service: OUTPUT_TOKENS_TOTAL = "output_tokens_total" # Time to first token in seconds TIME_TO_FIRST_TOKEN_SECONDS = "time_to_first_token_seconds" + # Exclusive contributions to HTTP time to first token, labeled by stage and phase. + TTFT_STAGE_SECONDS = "ttft_stage_seconds" # Inter-token latency in seconds INTER_TOKEN_LATENCY_SECONDS = "inter_token_latency_seconds" # End-to-end latency of an OpenAI `/v1/embeddings` request, in seconds. diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index 89f900f0f218..e89c94c05c80 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -28,6 +28,7 @@ use std::{ use crate::discovery::ModelManager; use crate::local_model::runtime_config::ModelRuntimeConfig; use crate::model_card::ModelDeploymentCard; +use crate::protocols::common::ttft::{TTFT_CONTEXT_KEY, TtftStage, TtftTimings}; use crate::protocols::{ common::metrics::{ANNOTATION_LLM_METRICS, LLMMetricAnnotation}, openai::chat_completions::NvCreateChatCompletionStreamResponse, @@ -635,6 +636,9 @@ pub struct Metrics { tokenizer_latency: HistogramVec, output_tokens_counter: IntCounterVec, time_to_first_token: HistogramVec, + ttft_stage: HistogramVec, + ttft_stage_handles: + dashmap::DashMap>, inter_token_latency: HistogramVec, /// End-to-end latency of an OpenAI `/v1/embeddings` request. Distinct from /// `request_duration` so the buckets can be sized for sub-second pooling @@ -810,6 +814,9 @@ pub struct ResponseMetricCollector { // so observations are equivalent. output_tokens_counter: prometheus::IntCounter, time_to_first_token: prometheus::Histogram, + ttft_stage: Arc<[prometheus::Histogram; TtftStage::ALL.len()]>, + ttft_timeline: Option>, + ttft_request_id: Option, inter_token_latency: Option, itl_pending_tokens: u64, input_sequence_length: prometheus::Histogram, @@ -1088,6 +1095,23 @@ impl Metrics { ) .unwrap(); + let (stage_min, stage_max, stage_count) = parse_bucket_config( + env, + env_metrics::DYN_METRICS_TTFT_STAGE, + 0.000001, + 480.0, + 26, + ); + let ttft_stage = HistogramVec::new( + HistogramOpts::new( + frontend_metric_name(frontend_service::TTFT_STAGE_SECONDS), + "Exclusive HTTP TTFT stage duration in seconds; stages sum to HTTP TTFT", + ) + .buckets(generate_log_buckets(stage_min, stage_max, stage_count)), + &["model", "stage", "phase"], + ) + .unwrap(); + let inter_token_latency_buckets = inter_token_latency_buckets(env); let inter_token_latency = HistogramVec::new( @@ -1325,6 +1349,8 @@ impl Metrics { tokenizer_latency, output_tokens_counter, time_to_first_token, + ttft_stage, + ttft_stage_handles: dashmap::DashMap::new(), inter_token_latency, embedding_latency, images_per_request, @@ -1484,6 +1510,7 @@ impl Metrics { registry.register(Box::new(self.tokenizer_latency.clone()))?; registry.register(Box::new(self.output_tokens_counter.clone()))?; registry.register(Box::new(self.time_to_first_token.clone()))?; + registry.register(Box::new(self.ttft_stage.clone()))?; registry.register(Box::new(self.inter_token_latency.clone()))?; registry.register(Box::new(self.embedding_latency.clone()))?; registry.register(Box::new(self.images_per_request.clone()))?; @@ -1963,6 +1990,20 @@ impl ResponseMetricCollector { // per-chunk / per-token hot path in `observe_response` does no label hashing. let output_tokens_counter = metrics.output_tokens_counter.with_label_values(&[&model]); let time_to_first_token = metrics.time_to_first_token.with_label_values(&[&model]); + // Resolve the finite stage label set once per model, not per request/token. + let ttft_stage = metrics + .ttft_stage_handles + .entry(model.clone()) + .or_insert_with(|| { + Arc::new(TtftStage::ALL.map(|stage| { + let [name, phase] = stage.labels(); + metrics + .ttft_stage + .with_label_values(&[model.as_str(), name, phase]) + })) + }) + .clone(); + let input_sequence_length = metrics.input_sequence_length.with_label_values(&[&model]); let cached_tokens = metrics.cached_tokens.with_label_values(&[&model]); let images_per_request = metrics.images_per_request.with_label_values(&[&model]); @@ -1976,6 +2017,9 @@ impl ResponseMetricCollector { model, output_tokens_counter, time_to_first_token, + ttft_stage, + ttft_timeline: None, + ttft_request_id: None, inter_token_latency: None, itl_pending_tokens: 0, input_sequence_length, @@ -2010,6 +2054,53 @@ impl ResponseMetricCollector { } } + /// Preserve this collector's exact start across the Python processor bridge. + /// Other paths still emit an additive sample, with uninstrumented time in + /// frontend_overhead, so the stage and total histograms have the same cohort. + pub fn attach_ttft_timeline( + &mut self, + request: &mut dynamo_runtime::pipeline::Context, + ) { + let timeline = Arc::new(TtftTimings::new(self.start_time)); + request.insert(TTFT_CONTEXT_KEY, timeline.clone()); + self.ttft_timeline = Some(timeline); + self.ttft_request_id = Some(request.id().to_owned()); + } + + /// One structured event for the same frozen durations sent to Prometheus. + /// Its dedicated target can be disabled without disabling metrics. The + /// tracing macro evaluates numeric fields only when this event is enabled. + fn log_ttft_stages(&self, total: Duration, stages: &[Duration; TtftStage::ALL.len()]) { + let Some(request_id) = self.ttft_request_id.as_deref() else { + return; + }; + let ms = |stage: TtftStage| stages[stage as usize].as_secs_f64() * 1000.0; + tracing::info!( + target: "dynamo_frontend::ttft", + request_id, + model = self.model.as_str(), + ttft_ms = total.as_secs_f64() * 1000.0, + frontend_overhead_ms = ms(TtftStage::FrontendOverhead), + python_preprocess_ms = ms(TtftStage::PythonPreprocess), + preprocess_queue_ms = ms(TtftStage::PreprocessQueue), + template_tokenize_ms = ms(TtftStage::TemplateTokenize), + multimodal_prepare_ms = ms(TtftStage::MultimodalPrepare), + mm_routing_build_ms = ms(TtftStage::MmRoutingBuild), + kv_route_prefill_ms = ms(TtftStage::RoutePrefill), + kv_route_decode_ms = ms(TtftStage::RouteDecode), + kv_route_aggregated_ms = ms(TtftStage::RouteAggregated), + backend_first_response_prefill_ms = ms(TtftStage::BackendPrefill), + backend_first_response_decode_ms = ms(TtftStage::BackendDecode), + backend_first_response_aggregated_ms = ms(TtftStage::BackendAggregated), + backend_first_token_prefill_ms = ms(TtftStage::FirstTokenPrefill), + backend_first_token_decode_ms = ms(TtftStage::FirstTokenDecode), + backend_first_token_aggregated_ms = ms(TtftStage::FirstTokenAggregated), + first_token_postprocess_ms = ms(TtftStage::ResponsePostprocess), + backend_overlap_prefill_decode_ms = ms(TtftStage::BackendOverlap), + "frontend first token" + ); + } + /// Set the worker info for per-worker TTFT/ITL metrics. /// In disaggregated mode, TTFT is attributed to prefill worker, ITL to decode worker. /// Worker types are stored at routing time to avoid expensive MDC lookup when updating metrics. @@ -2174,9 +2265,23 @@ impl ResponseMetricCollector { self.is_first_token = false; // Publish TTFT and store for span recording - let ttft = self.start_time.elapsed().as_secs_f64(); + let first_token_at = Instant::now(); + let duration = first_token_at.duration_since(self.start_time); + let ttft = duration.as_secs_f64(); self.ttft_ms = Some(ttft * 1000.0); self.time_to_first_token.observe(ttft); + let stages = if let Some(timeline) = self.ttft_timeline.take() { + timeline.finish(first_token_at) + } else { + let mut stages = [Duration::ZERO; TtftStage::ALL.len()]; + stages[TtftStage::FrontendOverhead as usize] = duration; + stages + }; + // Include zeros so all stages share the exact total-TTFT population. + for (histogram, elapsed) in self.ttft_stage.iter().zip(stages) { + histogram.observe(elapsed.as_secs_f64()); + } + self.log_ttft_stages(duration, &stages); // Update per-worker TTFT and input sequence tokens gauges - attributed to prefill worker. // Both gauges are updated atomically from the same request to correlate latency with input size. @@ -2607,6 +2712,391 @@ async fn handler_metrics(State(state): State>) -> impl mod tests { use super::*; + const EXPECTED_TTFT_STAGE_LABELS: [(&str, &str); 17] = [ + ("frontend_overhead", ""), + ("python_preprocess", ""), + ("preprocess_queue", ""), + ("template_tokenize", ""), + ("multimodal_prepare", ""), + ("mm_routing_build", ""), + ("kv_route", "prefill"), + ("kv_route", "decode"), + ("kv_route", "aggregated"), + ("backend_first_response", "prefill"), + ("backend_first_response", "decode"), + ("backend_first_response", "aggregated"), + ("backend_first_token", "prefill"), + ("backend_first_token", "decode"), + ("backend_first_token", "aggregated"), + ("backend_overlap", "prefill_decode"), + ("first_token_postprocess", ""), + ]; + + fn assert_ttft_stage_cohort(metrics: &Metrics, model: &str, expected_count: u64) { + let total = metrics.time_to_first_token.with_label_values(&[model]); + assert_eq!(total.get_sample_count(), expected_count); + let mut stage_sum = 0.0; + for (stage, phase) in EXPECTED_TTFT_STAGE_LABELS { + let histogram = metrics.ttft_stage.with_label_values(&[model, stage, phase]); + assert_eq!( + histogram.get_sample_count(), + expected_count, + "stage={stage}, phase={phase} must use the total TTFT request population", + ); + stage_sum += histogram.get_sample_sum(); + } + let total_sum = total.get_sample_sum(); + assert!( + (stage_sum - total_sum).abs() <= 1e-12 * total_sum.max(1.0), + "stage sum {stage_sum} differs from actual total TTFT histogram sum {total_sum}", + ); + } + + #[test] + fn ttft_stage_histogram_uses_frontend_prefix_and_documented_bucket_configuration() { + let env = fake_env(&[ + ("DYN_METRICS_TTFT_STAGE_MIN", "0.002"), + ("DYN_METRICS_TTFT_STAGE_MAX", "8"), + ("DYN_METRICS_TTFT_STAGE_COUNT", "5"), + ]); + let metrics = Metrics::build(Some("custom_frontend".to_string()), &env); + metrics + .ttft_stage + .with_label_values(&["model", "frontend_overhead", ""]) + .observe(0.1); + let families = metrics.ttft_stage.collect(); + assert_eq!(families[0].name(), "custom_frontend_ttft_stage_seconds"); + let buckets: Vec<_> = families[0].get_metric()[0] + .get_histogram() + .get_bucket() + .iter() + .map(|bucket| bucket.upper_bound()) + .collect(); + assert_eq!(buckets, generate_log_buckets(0.002, 8.0, 5)); + } + + #[test] + fn ttft_stage_cohorts_match_for_zero_tokens_later_tokens_and_cancelled_requests() { + let metrics = Arc::new(Metrics::new_with_prefix(None)); + let model = "ttft-cohort-model"; + let mut expected_count = 0; + + // Both instrumented SGLang and other frontend paths contribute to the + // same population; paths without a timeline put all time in overhead. + for attach_timeline in [false, true] { + let mut collector = metrics.clone().create_response_collector(model); + collector.start_time -= Duration::from_secs(1); + let mut request = dynamo_runtime::pipeline::Context::new(()); + if attach_timeline { + collector.attach_ttft_timeline(&mut request); + } + collector.observe_response(20, 0); + assert_ttft_stage_cohort(&metrics, model, expected_count); + collector.observe_response(20, 3); + expected_count += 1; + assert_ttft_stage_cohort(&metrics, model, expected_count); + + // Neither subsequent tokens nor collector drop repeat TTFT samples. + collector.observe_response(20, 0); + collector.observe_response(20, 5); + drop(collector); + assert_ttft_stage_cohort(&metrics, model, expected_count); + + let mut cancelled = metrics.clone().create_response_collector(model); + if attach_timeline { + cancelled.attach_ttft_timeline(&mut request); + } + cancelled.observe_response(20, 0); + drop(cancelled); + assert_ttft_stage_cohort(&metrics, model, expected_count); + } + } + + #[test] + fn ttft_shared_request_timeline_is_observed_on_existing_http_registry() { + let metrics = Arc::new(Metrics::new_with_prefix(None)); + let registry = prometheus::Registry::new(); + metrics.register(®istry).unwrap(); + let model = "ttft-shared-timeline-model"; + let mut collector = metrics.clone().create_response_collector(model); + let mut request = dynamo_runtime::pipeline::Context::new(()); + collector.attach_ttft_timeline(&mut request); + let shared = request.get::>(TTFT_CONTEXT_KEY).unwrap(); + assert!(Arc::ptr_eq( + shared.as_ref(), + collector.ttft_timeline.as_ref().unwrap(), + )); + + // The same handle travels through the Python bridge and routed engine; + // an open span must be clipped to this collector's first-token instant. + let route_span = shared.start_span(TtftStage::RouteDecode); + std::thread::sleep(Duration::from_millis(1)); + collector.observe_response(20, 1); + assert_ttft_stage_cohort(&metrics, model, 1); + let before = metrics + .ttft_stage + .with_label_values(&[model, "kv_route", "decode"]) + .get_sample_sum(); + assert!(before > 0.0); + drop(route_span); + collector.observe_response(20, 2); + assert_eq!( + metrics + .ttft_stage + .with_label_values(&[model, "kv_route", "decode"]) + .get_sample_sum(), + before, + "late span completion must not change the frozen HTTP TTFT", + ); + + let family = registry + .gather() + .into_iter() + .find(|family| family.name() == "dynamo_frontend_ttft_stage_seconds") + .expect("stage histogram must be exported by the existing HTTP registry"); + assert_eq!(family.get_metric().len(), EXPECTED_TTFT_STAGE_LABELS.len()); + for sample in family.get_metric() { + let labels: std::collections::BTreeMap<_, _> = sample + .get_label() + .iter() + .map(|label| (label.name(), label.value())) + .collect(); + assert_eq!(labels.len(), 3, "no request/worker IDs in stage labels"); + assert_eq!(labels["model"], model); + assert!(EXPECTED_TTFT_STAGE_LABELS.contains(&(labels["stage"], labels["phase"]))); + assert_eq!(sample.get_histogram().get_sample_count(), 1); + } + } + + #[test] + fn ttft_stage_labels_and_cached_histograms_preserve_model_separation() { + let metrics = Arc::new(Metrics::new_with_prefix(None)); + let mut first = metrics.clone().create_response_collector("ttft-model-a"); + let mut first_again = metrics.clone().create_response_collector("ttft-model-a"); + let mut second = metrics.clone().create_response_collector("ttft-model-b"); + assert!(Arc::ptr_eq(&first.ttft_stage, &first_again.ttft_stage)); + assert!(!Arc::ptr_eq(&first.ttft_stage, &second.ttft_stage)); + first.start_time -= Duration::from_secs(1); + second.start_time -= Duration::from_secs(3); + first.observe_response(10, 1); + first_again.observe_response(10, 1); + second.observe_response(10, 1); + assert_ttft_stage_cohort(&metrics, "ttft-model-a", 2); + assert_ttft_stage_cohort(&metrics, "ttft-model-b", 1); + assert!( + metrics + .ttft_stage + .with_label_values(&["ttft-model-b", "frontend_overhead", ""]) + .get_sample_sum() + >= 3.0, + ); + assert_eq!(metrics.ttft_stage_handles.len(), 2); + } + + #[derive(Default)] + struct TtftLogFields { + numbers: std::collections::BTreeMap<&'static str, f64>, + text: std::collections::BTreeMap<&'static str, String>, + } + + impl tracing::field::Visit for TtftLogFields { + fn record_f64(&mut self, field: &tracing::field::Field, value: f64) { + self.numbers.insert(field.name(), value); + } + + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.text.insert(field.name(), value.to_string()); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.text.insert(field.name(), format!("{value:?}")); + } + } + + /// Capture actual tracing events without installing a process-global logger + /// or adding a test-only logging dependency. + #[derive(Clone)] + struct TtftLogSubscriber { + info_enabled: bool, + events: Arc>>, + } + + impl tracing::Subscriber for TtftLogSubscriber { + fn enabled(&self, metadata: &tracing::Metadata<'_>) -> bool { + self.info_enabled + && metadata.target() == "dynamo_frontend::ttft" + && *metadata.level() == tracing::Level::INFO + } + + fn register_callsite( + &self, + _: &'static tracing::Metadata<'static>, + ) -> tracing::subscriber::Interest { + // Tests install independent subscribers on different threads. Do + // not cache one test's enabled/disabled setting for another test. + tracing::subscriber::Interest::sometimes() + } + + fn max_level_hint(&self) -> Option { + Some(if self.info_enabled { + tracing::level_filters::LevelFilter::INFO + } else { + tracing::level_filters::LevelFilter::OFF + }) + } + + fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { + tracing::span::Id::from_u64(1) + } + + fn record(&self, _: &tracing::span::Id, _: &tracing::span::Record<'_>) {} + + fn record_follows_from(&self, _: &tracing::span::Id, _: &tracing::span::Id) {} + + fn event(&self, event: &tracing::Event<'_>) { + if self.enabled(event.metadata()) { + let mut fields = TtftLogFields::default(); + event.record(&mut fields); + self.events + .lock() + .unwrap() + .push((*event.metadata().level(), fields)); + } + } + + fn enter(&self, _: &tracing::span::Id) {} + + fn exit(&self, _: &tracing::span::Id) {} + } + + #[test] + fn ttft_log_is_once_at_first_token_and_matches_histograms_and_request_identity() { + let events = Arc::new(std::sync::Mutex::new(Vec::new())); + let subscriber = TtftLogSubscriber { + info_enabled: true, + events: events.clone(), + }; + let metrics = Arc::new(Metrics::new_with_prefix(None)); + let model = "ttft-log-model"; + tracing::subscriber::with_default(subscriber, || { + let mut collector = metrics.clone().create_response_collector(model); + collector.start_time -= Duration::from_secs(1); + let mut request = dynamo_runtime::pipeline::Context::with_id_and_metadata( + (), + "client-request-42".to_string(), + Default::default(), + ); + collector.attach_ttft_timeline(&mut request); + let timeline = request.get::>(TTFT_CONTEXT_KEY).unwrap(); + timeline.record_ago( + TtftStage::PythonPreprocess, + Duration::from_millis(500), + Duration::from_millis(200), + ); + timeline.record_ago( + TtftStage::TemplateTokenize, + Duration::from_millis(400), + Duration::from_millis(300), + ); + let route = timeline.start_span(TtftStage::RouteDecode); + + collector.observe_response(20, 0); + assert!(events.lock().unwrap().is_empty()); + assert_ttft_stage_cohort(&metrics, model, 0); + collector.observe_response(20, 3); + assert_eq!(events.lock().unwrap().len(), 1); + assert_ttft_stage_cohort(&metrics, model, 1); + drop(route); + collector.observe_response(20, 0); + collector.observe_response(20, 4); + drop(collector); + + // A request that never produces tokens must neither log a TTFT nor + // add samples, even though it carries the full request context. + let mut cancelled = metrics.clone().create_response_collector(model); + cancelled.attach_ttft_timeline(&mut request); + cancelled.observe_response(20, 0); + drop(cancelled); + assert_ttft_stage_cohort(&metrics, model, 1); + + // Other endpoints still record metrics, but without the attached + // HTTP request context they must not emit an anonymous chat log. + let mut unattached = metrics + .clone() + .create_response_collector("ttft-unattached-model"); + unattached.observe_response(20, 1); + drop(unattached); + }); + + let events = events.lock().unwrap(); + assert_eq!( + events.len(), + 1, + "later tokens/drop/cancellation must not log" + ); + let (level, fields) = &events[0]; + assert_eq!(*level, tracing::Level::INFO); + assert_eq!(fields.text["message"], "frontend first token"); + assert_eq!(fields.text["request_id"], "client-request-42"); + assert_eq!(fields.text["model"], model); + assert_eq!( + fields.numbers.len(), + 18, + "17 numeric stage fields plus total" + ); + let total_ms = fields.numbers["ttft_ms"]; + let histogram_ttft_ms = metrics + .time_to_first_token + .with_label_values(&[model]) + .get_sample_sum() + * 1000.0; + assert_eq!(total_ms, histogram_ttft_ms); + let mut stage_sum_ms = 0.0; + for (stage, phase) in EXPECTED_TTFT_STAGE_LABELS { + let field = if phase.is_empty() { + format!("{stage}_ms") + } else { + format!("{stage}_{phase}_ms") + }; + let logged_ms = fields.numbers[field.as_str()]; + let histogram_ms = metrics + .ttft_stage + .with_label_values(&[model, stage, phase]) + .get_sample_sum() + * 1000.0; + assert_eq!(logged_ms, histogram_ms, "numeric field {field}"); + stage_sum_ms += logged_ms; + } + assert!((stage_sum_ms - total_ms).abs() <= 1e-9 * total_ms.max(1.0)); + assert!(fields.numbers["template_tokenize_ms"] > 0.0); + assert!(fields.numbers["python_preprocess_ms"] > 0.0); + } + + #[test] + fn ttft_log_disabled_does_not_change_metrics() { + let events = Arc::new(std::sync::Mutex::new(Vec::new())); + let subscriber = TtftLogSubscriber { + info_enabled: false, + events: events.clone(), + }; + let metrics = Arc::new(Metrics::new_with_prefix(None)); + let model = "ttft-disabled-log-model"; + tracing::subscriber::with_default(subscriber, || { + let mut collector = metrics.clone().create_response_collector(model); + let mut request = dynamo_runtime::pipeline::Context::new(()); + collector.attach_ttft_timeline(&mut request); + let timeline = request.get::>(TTFT_CONTEXT_KEY).unwrap(); + let route = timeline.start_span(TtftStage::RouteAggregated); + collector.observe_response(20, 0); + collector.observe_response(20, 1); + drop(route); + collector.observe_response(20, 2); + drop(collector); + }); + assert!(events.lock().unwrap().is_empty()); + assert_ttft_stage_cohort(&metrics, model, 1); + } + fn model_ready_value_with_name( registry: &Registry, metric_name: &str, diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index bc7c23658f5a..5b107b951906 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -1126,6 +1126,15 @@ fn copy_context_metadata( ); } + if let Ok(timings) = source.get::>( + crate::protocols::common::ttft::TTFT_CONTEXT_KEY, + ) { + target.insert( + crate::protocols::common::ttft::TTFT_CONTEXT_KEY, + Arc::clone(timings.as_ref()), + ); + } + if let Ok(agent_context) = source.get::(AGENT_CONTEXT_CONTEXT_KEY) { target.insert(AGENT_CONTEXT_CONTEXT_KEY, agent_context.as_ref().clone()); } @@ -1308,6 +1317,7 @@ async fn completions_single( let mut response_collector = state .metrics_clone() .create_response_collector(&metric_model); + response_collector.attach_ttft_timeline(&mut request); // prepare to process any annotations let annotations = request.annotations(); @@ -1602,6 +1612,7 @@ async fn completions_batch( let mut response_collector = state .metrics_clone() .create_response_collector(&metric_model); + response_collector.attach_ttft_timeline(&mut request); // prepare to process any annotations let annotations = request.annotations(); @@ -3590,6 +3601,7 @@ async fn chat_completions( let mut response_collector = state .metrics_clone() .create_response_collector(&metric_model); + response_collector.attach_ttft_timeline(&mut request); let annotations = request.annotations(); @@ -6529,6 +6541,18 @@ mod tests { assert!(err.to_string().contains("unknown field `agent_context`")); } + #[test] + fn completion_batch_child_shares_parent_ttft_timeline() { + use crate::protocols::common::ttft::{TTFT_CONTEXT_KEY, TtftTimings}; + let timeline = Arc::new(TtftTimings::new(std::time::Instant::now())); + let mut source = Context::new(()); + source.insert(TTFT_CONTEXT_KEY, timeline.clone()); + let mut child = Context::new(()); + copy_context_metadata(&source, &mut child); + let copied = child.get::>(TTFT_CONTEXT_KEY).unwrap(); + assert!(Arc::ptr_eq(copied.as_ref(), &timeline)); + } + #[test] fn test_copy_context_metadata_preserves_agent_context() { let mut source = Context::new(()); diff --git a/lib/llm/src/kv_router/routing_host.rs b/lib/llm/src/kv_router/routing_host.rs index 5d49e86a3d3e..74d5f77767a6 100644 --- a/lib/llm/src/kv_router/routing_host.rs +++ b/lib/llm/src/kv_router/routing_host.rs @@ -38,6 +38,7 @@ use crate::{ llm_backend::LLMEngineOutput, preprocessor::owned_abort_error, timing::{RequestPhase, RoutingData, WORKER_TYPE_DECODE, WORKER_TYPE_PREFILL}, + ttft::{TtftSpan, TtftStage}, }, session_affinity::{ AffinityCoordinator, AffinityTarget, Hold, SessionAffinityMode, affinity_id, @@ -851,6 +852,7 @@ impl AsyncEngine, ManyOut, ManyOut stream, @@ -925,6 +927,16 @@ impl AsyncEngine, ManyOut Option { + let timings = request.tracker.as_ref()?.ttft()?; + let stage = match phase { + RequestPhase::Prefill => TtftStage::RoutePrefill, + RequestPhase::Decode => TtftStage::RouteDecode, + RequestPhase::Aggregated => TtftStage::RouteAggregated, + }; + Some(timings.start_span(stage)) +} + enum ResponseItemOutcome { /// The stream is healthy and must keep running. Healthy, diff --git a/lib/llm/src/kv_router/routing_host/builtin.rs b/lib/llm/src/kv_router/routing_host/builtin.rs index 6273458c0bfd..743efaf8de82 100644 --- a/lib/llm/src/kv_router/routing_host/builtin.rs +++ b/lib/llm/src/kv_router/routing_host/builtin.rs @@ -286,6 +286,7 @@ impl RoutingHost { F: FnOnce(&mut PreprocessedRequest, AffinityTarget) -> Result, { let phase_label = phase.to_string(); + let ttft_route = start_ttft_route(&request, phase); let route_guard = StageGuard::new(STAGE_ROUTE, &phase_label); let explicit = explicit_target(request.content(), phase)?; let has_affinity_session = self.affinity.is_some() && affinity_id(&request)?.is_some(); @@ -365,6 +366,7 @@ impl RoutingHost { occupancy_reservation, lora_load, &request, + phase, ); let tracker = request.tracker.clone(); let request_context = request.context().clone(); @@ -374,6 +376,7 @@ impl RoutingHost { .observe(request.token_ids.len() as f64); drop(route_guard); + drop(ttft_route); guard.start_dispatch(&phase_label); guard.record_prefill_start(request.content()); let dispatch_result = if is_direct && !has_affinity_session { diff --git a/lib/llm/src/kv_router/routing_host/kv.rs b/lib/llm/src/kv_router/routing_host/kv.rs index 7f8ef88c3cd9..98fd28b7a3a4 100644 --- a/lib/llm/src/kv_router/routing_host/kv.rs +++ b/lib/llm/src/kv_router/routing_host/kv.rs @@ -211,6 +211,7 @@ impl RoutingHost { self.validate_explicit_worker(request.content(), phase)?; let phase_label = phase.to_string(); + let _ttft_route = start_ttft_route(request, phase); let route_guard = StageGuard::new(STAGE_ROUTE, &phase_label); let (outcome, _) = self .select_with_session_affinity(request, phase, true, &budget, |target| { @@ -257,6 +258,7 @@ impl RoutingHost { let phase = preview.phase; let phase_label = phase.to_string(); + let _ttft_route = start_ttft_route(request, phase); let route_guard = StageGuard::new(STAGE_ROUTE, &phase_label); let planned_worker = preview.signals.worker; let select = || { @@ -308,6 +310,7 @@ impl RoutingHost { budget, .. } = plan; + let ttft_route = start_ttft_route(&request, request.phase()); let selected_target = route_target(selection.worker); let guard = match self .track_planned_selection(&request, &mut selection, cleanup, &budget) @@ -317,7 +320,7 @@ impl RoutingHost { Err(error) => return Err(error), }; let stream = match self - .dispatch_selection(request, selection, guard, &budget) + .dispatch_selection(request, selection, guard, &budget, ttft_route) .await { Ok(stream) => stream, @@ -437,6 +440,7 @@ impl RoutingHost { request, kv_route, selection.request_lifecycle.take(), + phase, ); let record_result: Result<(), Error> = async { @@ -538,14 +542,14 @@ impl RoutingHost { selection: WorkerSelection, mut guard: RequestGuard, budget: &CleanupBudget, + ttft_route: Option, ) -> Result>, Error> { let context_id = request.context().id().to_string(); let request_context = request.context().clone(); let route_trace_context = get_route_trace_context(&request); - let phase = request.phase(); + let phase = guard.phase(); let staged_kv = StagedKv::for_request(request.content()); let phase_label = phase.to_string(); - guard.start_dispatch(&phase_label); self.warn_if_output_replay_annotation_ignored(&request, &selection); let (mut backend_input, context) = request.into_parts(); @@ -553,6 +557,8 @@ impl RoutingHost { backend_input.kv_hint = selection.kv_hint; let updated_request = context.map(|_| backend_input); guard.record_prefill_start(updated_request.content()); + drop(ttft_route); + guard.start_dispatch(&phase_label); let dispatch = self .inner @@ -659,6 +665,7 @@ impl RoutingHost { let budget = CleanupBudget::default(); let phase = RequestPhase::Prefill; let phase_label = phase.to_string(); + let ttft_route = start_ttft_route(&request, phase); let route_guard = StageGuard::new(STAGE_ROUTE, &phase_label); let is_query_only = request.get_annotation_value("query_instance_id").is_some(); let (mut selection, mut operation) = self @@ -681,7 +688,7 @@ impl RoutingHost { }; drop(route_guard); let stream = match self - .dispatch_selection(request, selection, guard, &budget) + .dispatch_selection(request, selection, guard, &budget, ttft_route) .await { Ok(stream) => stream, diff --git a/lib/llm/src/kv_router/routing_host/request_guard.rs b/lib/llm/src/kv_router/routing_host/request_guard.rs index 2cd5f77251c8..fd4b83bb8e4c 100644 --- a/lib/llm/src/kv_router/routing_host/request_guard.rs +++ b/lib/llm/src/kv_router/routing_host/request_guard.rs @@ -16,6 +16,7 @@ use crate::{ llm_backend::LLMEngineOutput, preprocessor::MigrationState, timing::{RequestPhase, RequestTracker}, + ttft::{TtftSpan, TtftStage, TtftTimings}, }, }; use dynamo_kv_router::{ @@ -273,6 +274,83 @@ impl CanonicalOutputTracker { } } +/// Captures the dispatch phase before prefill and decode can run concurrently. +/// Span handles are consumed at the first response/token, so later tokens do +/// not read the clock or lock the TTFT timeline. +struct DispatchTimings { + phase: RequestPhase, + timings: Option>, + backend: Option, + first_token: Option, + postprocess: Option, +} + +impl DispatchTimings { + fn new(phase: RequestPhase, timings: Option>) -> Self { + Self { + phase, + timings, + backend: None, + first_token: None, + postprocess: None, + } + } + + fn start(&mut self) { + if let Some(timings) = &self.timings { + let stage = match self.phase { + RequestPhase::Prefill => TtftStage::BackendPrefill, + RequestPhase::Decode => TtftStage::BackendDecode, + RequestPhase::Aggregated => TtftStage::BackendAggregated, + }; + self.backend = Some(timings.start_span(stage)); + } + } + + fn observe_response(&mut self, has_tokens: bool) { + if let Some(backend) = self.backend.take() { + drop(backend); + // If the first response already carries tokens, there is no + // separate first-response-to-first-token interval. + if !has_tokens && let Some(timings) = &self.timings { + let stage = match self.phase { + RequestPhase::Prefill => TtftStage::FirstTokenPrefill, + RequestPhase::Decode => TtftStage::FirstTokenDecode, + RequestPhase::Aggregated => TtftStage::FirstTokenAggregated, + }; + self.first_token = Some(timings.start_span(stage)); + } + } + } + + fn observe_first_token(&mut self) { + self.first_token.take(); + if self.phase != RequestPhase::Prefill + && let Some(timings) = &self.timings + { + self.postprocess = Some(timings.start_span(TtftStage::ResponsePostprocess)); + } + } + + fn finish(&mut self) { + self.backend.take(); + self.first_token.take(); + // A successful single-token stream can end before Python/HTTP sees it. + // Preserve forwarding time after the dispatch guard is dropped. + if let Some(postprocess) = self.postprocess.take() { + postprocess.leave_open(); + } + } + + fn abort(&mut self) { + self.backend.take(); + self.first_token.take(); + // A terminal frame may contain tokens that are withheld then discarded + // on a migratable error. It must not mask the next attempt's routing/wait. + self.postprocess.take(); + } +} + /// Owns request-scoped timing and metrics state. struct RequestObservability { tracker: Option>, @@ -282,13 +360,16 @@ struct RequestObservability { first_token_recorded: bool, dispatch_guard: Option, dispatched: bool, + ttft: DispatchTimings, } impl RequestObservability { fn new( tracker: Option>, request_metrics: Arc, + phase: RequestPhase, ) -> Self { + let timings = tracker.as_ref().and_then(|tracker| tracker.ttft()).cloned(); Self { tracker, request_metrics, @@ -297,6 +378,7 @@ impl RequestObservability { first_token_recorded: false, dispatch_guard: None, dispatched: false, + ttft: DispatchTimings::new(phase, timings), } } @@ -306,6 +388,7 @@ impl RequestObservability { fn start_dispatch(&mut self, phase_label: &str) { self.dispatch_guard = Some(StageGuard::new(STAGE_DISPATCH, phase_label)); + self.ttft.start(); } /// Record prefill start for dispatches that actually run prefill. @@ -317,7 +400,7 @@ impl RequestObservability { let Some(tracker) = &self.tracker else { return; }; - let includes_prefill = tracker.phase() != RequestPhase::Decode + let includes_prefill = self.ttft.phase != RequestPhase::Decode || request .annotations .iter() @@ -332,16 +415,18 @@ impl RequestObservability { self.dispatched = true; } - fn observe_response(&mut self) { + fn observe_response(&mut self, has_tokens: bool) { // Taking the guard ends dispatch latency exactly once; later responses see None. self.dispatch_guard.take(); + self.ttft.observe_response(has_tokens); } fn observe_tokens(&mut self, new_tokens: usize) { if !self.first_token_recorded && new_tokens > 0 { + self.ttft.observe_first_token(); if let Some(tracker) = &self.tracker { tracker.record_first_token(); - if tracker.phase() == RequestPhase::Decode { + if self.ttft.phase == RequestPhase::Decode { tracker.record_decode_first_token(); } if let Some(ttft) = tracker.ttft_ms() { @@ -618,6 +703,7 @@ impl RequestGuard { request: &PreprocessedRequest, kv_route: Option, mut request_lifecycle: Option>, + phase: RequestPhase, ) -> Self { if let Some(lifecycle) = request_lifecycle.as_mut() { lifecycle.observe_context_tokens(request.expanded_prompt_token_count()); @@ -653,7 +739,11 @@ impl RequestGuard { .map(|_| CanonicalOutputTracker::new(request, block_size as u32, chooser.is_eagle())); Self { cleanup: RequestCleanup::Kv(cleanup), - observability: RequestObservability::new(request.tracker.clone(), request_metrics), + observability: RequestObservability::new( + request.tracker.clone(), + request_metrics, + phase, + ), output_blocks: OutputBlockTracker::new( track_output_blocks, isl_tokens, @@ -676,6 +766,7 @@ impl RequestGuard { occupancy_reservation: Option, lora_load: Option, request: &PreprocessedRequest, + phase: RequestPhase, ) -> Self { request_metrics.requests_started_total.inc(); Self { @@ -686,7 +777,11 @@ impl RequestGuard { }, None => RequestCleanup::Stateless { worker_id }, }, - observability: RequestObservability::new(request.tracker.clone(), request_metrics), + observability: RequestObservability::new( + request.tracker.clone(), + request_metrics, + phase, + ), // Builtin policies do not track scheduler blocks. Emit one final ITL sample // when the request completes rather than observing every streamed token. output_blocks: OutputBlockTracker::new(false, request.token_ids.len(), 1, None), @@ -714,6 +809,10 @@ impl RequestGuard { self.observability.request_metrics() } + pub(super) fn phase(&self) -> RequestPhase { + self.observability.ttft.phase + } + pub(super) fn start_dispatch(&mut self, phase_label: &str) { self.observability.start_dispatch(phase_label); } @@ -760,9 +859,10 @@ impl RequestGuard { } pub(super) async fn on_item(&mut self, item: &Annotated) { - self.observability.observe_response(); - let new_tokens = item.data.as_ref().map_or(0, |data| data.token_ids.len()); + // Observe arrival before async scheduler work, which belongs to postprocessing. + self.observability.observe_response(new_tokens > 0); + self.observability.observe_tokens(new_tokens); if new_tokens > 0 && let Some(lifecycle) = self.cleanup.lifecycle() { @@ -828,7 +928,6 @@ impl RequestGuard { "Failed to materialize approximate LRU output blocks" ); } - self.observability.observe_tokens(new_tokens); self.capture_kv_worker_hit(item); let cumulative_osl = self.observability.cumulative_osl(); let Some(update) = self.output_blocks.observe(cumulative_osl) else { @@ -865,6 +964,7 @@ impl RequestGuard { self.observability .record_metrics(self.record_itl_at_completion); self.cleanup.finish().await; + self.observability.ttft.finish(); } pub(super) async fn abort(&mut self) { @@ -881,6 +981,7 @@ impl RequestGuard { lifecycle.prepare_retry(); migration_state.store_request_lifecycle(lifecycle); self.cleanup.finish().await; + self.observability.ttft.abort(); true } @@ -889,6 +990,7 @@ impl RequestGuard { lifecycle.abort(error.map(crate::protocols::common::preprocessor::owned_abort_error)); } self.cleanup.finish().await; + self.observability.ttft.abort(); } /// Count a worker report once, even if the stream subsequently fails or is cancelled. @@ -1182,7 +1284,7 @@ mod prefill_start_tests { let tracker = Arc::new(RequestTracker::new()); let _permit = tracker.set_phase(phase).await; let request = test_request(tracker.clone(), annotations); - RequestObservability::new(request.tracker.clone(), test_metrics()) + RequestObservability::new(request.tracker.clone(), test_metrics(), phase) .record_prefill_start(&request); tracker } @@ -1227,14 +1329,127 @@ mod prefill_start_tests { let request = test_request(tracker.clone(), Vec::new()); let prefill_permit = tracker.set_phase(RequestPhase::Prefill).await; - RequestObservability::new(Some(tracker.clone()), metrics.clone()) - .record_prefill_start(&request); + RequestObservability::new( + Some(tracker.clone()), + metrics.clone(), + RequestPhase::Prefill, + ) + .record_prefill_start(&request); let recorded_by_prefill = tracker.prefill_wait_time_ms(); assert!(recorded_by_prefill.is_some()); drop(prefill_permit); let _decode_permit = tracker.set_phase(RequestPhase::Decode).await; - RequestObservability::new(Some(tracker.clone()), metrics).record_prefill_start(&request); + RequestObservability::new(Some(tracker.clone()), metrics, RequestPhase::Decode) + .record_prefill_start(&request); assert_eq!(tracker.prefill_wait_time_ms(), recorded_by_prefill); } } + +#[cfg(test)] +mod ttft_tests { + use super::*; + use std::time::{Duration, Instant}; + + #[test] + fn first_token_keeps_postprocessing_open_after_dispatch_is_dropped() { + let start = Instant::now(); + let timeline = Arc::new(TtftTimings::new(start)); + let mut dispatch = DispatchTimings::new(RequestPhase::Decode, Some(timeline.clone())); + dispatch.start(); + dispatch.observe_response(true); + dispatch.observe_first_token(); + dispatch.finish(); + drop(dispatch); + let end = Instant::now() + Duration::from_secs(1); + let stages = timeline.finish(end); + assert_eq!(stages[TtftStage::FirstTokenDecode as usize], Duration::ZERO); + assert!(stages[TtftStage::ResponsePostprocess as usize] >= Duration::from_secs(1)); + assert_eq!(stages.into_iter().sum::(), end - start); + } + + #[test] + fn control_response_starts_a_distinct_wait_for_the_first_token() { + let start = Instant::now(); + let timeline = Arc::new(TtftTimings::new(start)); + let mut dispatch = DispatchTimings::new(RequestPhase::Aggregated, Some(timeline.clone())); + dispatch.start(); + dispatch.observe_response(false); + // Freeze while waiting for a token: metadata alone must not begin postprocessing. + let end = Instant::now() + Duration::from_secs(1); + let stages = timeline.finish(end); + assert!(stages[TtftStage::FirstTokenAggregated as usize] >= Duration::from_secs(1)); + assert_eq!( + stages[TtftStage::ResponsePostprocess as usize], + Duration::ZERO + ); + assert_eq!(stages.into_iter().sum::(), end - start); + dispatch.observe_response(true); + dispatch.observe_first_token(); + assert_eq!(timeline.finish(end + Duration::from_secs(1)), stages); + } + + #[tokio::test] + async fn discarded_terminal_tokens_do_not_mask_retry_backend_wait() { + let start = Instant::now(); + let timeline = Arc::new(TtftTimings::new(start)); + let tracker = Arc::new(RequestTracker::new()); + tracker.set_ttft(timeline.clone()); + let request = PreprocessedRequest::builder() + .model("test".to_string()) + .token_ids(vec![1]) + .stop_conditions(Default::default()) + .sampling_options(Default::default()) + .output_options(Default::default()) + .tracker(Some(tracker)) + .build() + .unwrap(); + let metrics = RouterRequestMetrics::for_test(&dynamo_runtime::MetricsRegistry::new()); + let mut guard = + RequestGuard::new_builtin(metrics, 1, None, None, &request, RequestPhase::Aggregated); + guard.start_dispatch("aggregated"); + // monitor_response_stream withholds this token-bearing terminal frame; + // a following migratable error discards it before HTTP sees a token. + guard + .on_item(&Annotated::from_data(LLMEngineOutput { + token_ids: vec![2], + finish_reason: Some(crate::protocols::common::FinishReason::Cancelled), + ..Default::default() + })) + .await; + guard.abort().await; + // The previous guard can still exist while Migration starts another attempt. + let _retry_route = timeline.start_span(TtftStage::RouteAggregated); + drop(_retry_route); + let _retry_backend = timeline.start_span(TtftStage::BackendAggregated); + let end = Instant::now() + Duration::from_secs(1); + let stages = timeline.finish(end); + assert!(stages[TtftStage::BackendAggregated as usize] >= Duration::from_secs(1)); + assert_eq!(stages.into_iter().sum::(), end - start); + } + + #[tokio::test] + async fn late_prefill_token_uses_dispatch_phase_after_decode_starts() { + let start = Instant::now(); + let timeline = Arc::new(TtftTimings::new(start)); + let tracker = Arc::new(RequestTracker::new()); + tracker.set_ttft(timeline.clone()); + drop(tracker.set_phase(RequestPhase::Prefill).await); + let metrics = RouterRequestMetrics::for_test(&dynamo_runtime::MetricsRegistry::new()); + let mut observability = + RequestObservability::new(Some(tracker.clone()), metrics, RequestPhase::Prefill); + observability.start_dispatch("prefill"); + drop(tracker.set_phase(RequestPhase::Decode).await); + observability.observe_response(true); + observability.observe_tokens(1); + let end = Instant::now() + Duration::from_secs(1); + let stages = timeline.finish(end); + assert_eq!(stages[TtftStage::BackendDecode as usize], Duration::ZERO); + assert_eq!(stages[TtftStage::FirstTokenDecode as usize], Duration::ZERO); + assert_eq!( + stages[TtftStage::ResponsePostprocess as usize], + Duration::ZERO + ); + assert_eq!(stages.into_iter().sum::(), end - start); + } +} diff --git a/lib/llm/src/kv_router/routing_host/tests.rs b/lib/llm/src/kv_router/routing_host/tests.rs index eb3b8b003db5..cf801201be36 100644 --- a/lib/llm/src/kv_router/routing_host/tests.rs +++ b/lib/llm/src/kv_router/routing_host/tests.rs @@ -164,6 +164,7 @@ async fn builtin_host_constructs_only_declared_capabilities() { Some(selection.reservation), None, &request(), + RequestPhase::Aggregated, ); assert_eq!(guard.retarget_worker(1), Some(1)); guard.abort().await; @@ -686,6 +687,7 @@ async fn terminal_item_does_not_skip_transport_eof() { selected_router_tokens: 0, }), None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -747,6 +749,7 @@ async fn run_kv_hit_attempt(final_frame: LLMEngineOutput) -> (KvHitSnapshot, KvH selected_router_tokens: 60, }), None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -844,6 +847,7 @@ async fn kv_cache_hit_counts_immediately_once_in_selection_phase(#[case] reused: selected_router_tokens: 1, }), None, + RequestPhase::Prefill, ); drop(permit); let _permit = tracker.set_phase(RequestPhase::Decode).await; @@ -944,6 +948,7 @@ async fn shutdown_cancellation_drains_trailing_engine_shutdown_error() { &request(), None, None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -994,6 +999,7 @@ async fn client_cancellation_still_ends_stream_without_draining() { &request(), None, None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -1036,6 +1042,7 @@ async fn drain_without_trailing_error_gives_up_at_the_deadline() { &request(), None, None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -1095,6 +1102,7 @@ async fn trailing_error_within_the_drain_window_still_reaches_migration() { &request(), None, None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -1145,6 +1153,7 @@ async fn always_ready_terminals_cannot_starve_the_drain_deadline() { &request(), None, None, + RequestPhase::Aggregated, ); let monitored = monitor_response_stream(source, context, guard); tokio::pin!(monitored); @@ -1754,7 +1763,7 @@ async fn kv_cancellation_after_admission_still_stops_the_aggregated_dispatch() { .await; let error = router - .dispatch_selection(request, selection, guard, &CleanupBudget::default()) + .dispatch_selection(request, selection, guard, &CleanupBudget::default(), None) .await .expect_err("aggregated dispatch must still be cancelled before it is issued"); assert!(match_error_chain( @@ -2629,7 +2638,8 @@ async fn router_request_counters_follow_admission_and_completion_lifecycle() { failed_request, failed_selection, failed_dispatch_guard, - &CleanupBudget::default() + &CleanupBudget::default(), + None, ) .await .is_err() @@ -2646,8 +2656,14 @@ async fn router_request_counters_follow_admission_and_completion_lifecycle() { assert_eq!(metrics.requests_started_total.get(), 3); assert_eq!(metrics.requests_total.get(), 1); - let mut builtin_guard = - RequestGuard::new_builtin(Arc::clone(&metrics), 7, None, None, &request()); + let mut builtin_guard = RequestGuard::new_builtin( + Arc::clone(&metrics), + 7, + None, + None, + &request(), + RequestPhase::Aggregated, + ); assert_eq!(metrics.requests_started_total.get(), 4); builtin_guard.abort().await; drop(builtin_guard); diff --git a/lib/llm/src/preprocessor.rs b/lib/llm/src/preprocessor.rs index e6a67b6f3b45..1c2f09ce9a85 100644 --- a/lib/llm/src/preprocessor.rs +++ b/lib/llm/src/preprocessor.rs @@ -73,6 +73,7 @@ use crate::protocols::common::preprocessor::{ MultimodalData, MultimodalDataMap, MultimodalUuidMap, PreprocessedRequestBuilder, RoutingHints, }; use crate::protocols::common::timing::RequestTracker; +use crate::protocols::common::ttft::{TTFT_CONTEXT_KEY, TtftStage, TtftTimings}; use crate::tokenizers::Encoding; use dynamo_parsers::{ @@ -1565,6 +1566,16 @@ fn attach_image_cache_scope_from_context( } } +/// Retain the response generator's tracker while sharing the HTTP first-token timeline. +fn attach_ttft_tracker( + tracker: &RequestTracker, + context: &PipelineContext, +) { + if let Ok(timings) = context.get::>(TTFT_CONTEXT_KEY) { + tracker.set_ttft(Arc::clone(timings.as_ref())); + } +} + fn attach_request_context_metadata( request: &mut PreprocessedRequest, context: &PipelineContext<()>, @@ -3041,6 +3052,9 @@ impl OpenAIPreprocessor { let template_start = Instant::now(); let formatted_prompt = { + let _ttft_stage = tracker + .and_then(RequestTracker::ttft) + .map(|timings| timings.start_span(TtftStage::TemplateTokenize)); let _nvtx = dynamo_nvtx_range!("preprocess.template"); off_runtime(|| self.apply_template(request)) .with_context(|| "Failed to apply prompt template")? @@ -3071,6 +3085,7 @@ impl OpenAIPreprocessor { &mut builder, formatted_prompt.as_ref().map(RenderedPrompt::as_str), &token_ids, + tracker, ) .await .with_context(|| "Failed to gather multimodal data")?; @@ -3580,6 +3595,7 @@ impl OpenAIPreprocessor { builder, formatted_prompt, token_ids, + None, ) .await?; Ok(entries @@ -3607,6 +3623,7 @@ impl OpenAIPreprocessor { builder: &mut PreprocessedRequestBuilder, formatted_prompt: Option<&str>, token_ids: &[crate::protocols::TokenIdType], + tracker: Option<&RequestTracker>, ) -> Result<(Vec, Option)> { // `token_ids` is only consumed by exact MM-routing construction below. #[cfg(not(feature = "mm-routing"))] @@ -3659,6 +3676,7 @@ impl OpenAIPreprocessor { return Ok((Vec::new(), None)); }; let has_media_loader = self.media_loader.is_some(); + let mut ttft_multimodal = None; let message_order = self.formatter.media_message_order(request); for index in 0..messages.len() { @@ -3670,6 +3688,11 @@ impl OpenAIPreprocessor { let Some((type_str, url, uuid)) = content_part.media_info() else { continue; }; + if ttft_multimodal.is_none() { + ttft_multimodal = tracker + .and_then(RequestTracker::ttft) + .map(|timings| timings.start_span(TtftStage::MultimodalPrepare)); + } #[cfg(feature = "mm-routing")] if type_str == "image_url" { @@ -4019,6 +4042,11 @@ impl OpenAIPreprocessor { extra_args_obj.extend(backend_extra_args); } + #[cfg(feature = "mm-routing")] + let _ttft_routing = tracker + .and_then(RequestTracker::ttft) + .map(|timings| timings.start_span(TtftStage::MmRoutingBuild)); + // Build and install routing info + worker hashes atomically. If any // mixed-media precondition misses, neither side receives exact keys // and the request cleanly uses text-prefix routing. @@ -4704,6 +4732,9 @@ impl OpenAIPreprocessor { formatted_prompt: Option<&RenderedPrompt>, tracker: Option<&RequestTracker>, ) -> Result<(Vec, HashMap)> { + let _ttft_stage = tracker + .and_then(RequestTracker::ttft) + .map(|timings| timings.start_span(TtftStage::TemplateTokenize)); let mut annotations = HashMap::new(); let mut tokens_out: Vec = Vec::new(); // match request type before any conversion/processing @@ -7671,7 +7702,9 @@ impl // create a response generator let response_generator = request.response_generator(context.id().to_string()); - let tracker = Some(response_generator.tracker()); + let request_tracker = response_generator.tracker(); + attach_ttft_tracker(&request_tracker, &context); + let tracker = Some(request_tracker); let preprocess_options = PreprocessRequestOptions { preserve_omitted_max_tokens: context .get::(PRESERVE_OMITTED_MAX_TOKENS_CONTEXT_KEY) @@ -7885,7 +7918,9 @@ impl // create a response generator let response_generator = request.response_generator(request_id.clone()); let mut response_generator = Box::new(response_generator); - let tracker = Some(response_generator.tracker()); + let request_tracker = response_generator.tracker(); + attach_ttft_tracker(&request_tracker, &context); + let tracker = Some(request_tracker); // convert the chat completion request to a common completion request let mut builder = self.builder_with_lora( &request, @@ -8523,7 +8558,7 @@ mod extra_args_media_copy_tests { .sampling_options(Default::default()) .output_options(Default::default()); preprocessor - .gather_multi_modal_data_with_image_tokens(&request, &mut builder, None, &[]) + .gather_multi_modal_data_with_image_tokens(&request, &mut builder, None, &[], None) .await .unwrap(); let result = builder.build().unwrap(); @@ -12021,7 +12056,7 @@ mod tests { .output_options(Default::default()); let (routing_entries, _) = preprocessor - .gather_multi_modal_data_with_image_tokens(&request, &mut builder, None, &[]) + .gather_multi_modal_data_with_image_tokens(&request, &mut builder, None, &[], None) .await .unwrap(); let preprocessed = builder.build().unwrap(); @@ -15962,3 +15997,87 @@ mod tests { ])); } } + +#[cfg(test)] +mod ttft_tests { + use super::*; + use std::time::Duration; + + #[derive(Default)] + struct RecordingBackend(Mutex>>); + + #[async_trait] + impl AsyncEngine, ManyOut>, Error> + for RecordingBackend + { + async fn generate( + &self, + request: SingleIn, + ) -> Result>, Error> { + let wire = serde_json::to_value(request.content()).unwrap(); + assert!(wire.get("tracker").is_none()); + assert!(wire.get("ttft").is_none()); + self.0 + .lock() + .unwrap() + .push(request.tracker.clone().unwrap()); + Ok(ResponseStream::new( + Box::pin(stream::empty()), + request.context(), + )) + } + } + + #[tokio::test] + async fn native_chat_and_completion_share_http_ttft_with_the_router() { + let mdc = ModelDeploymentCard::load_from_disk( + "tests/data/sample-models/mock-llama-3.1-8b-instruct", + None, + ) + .unwrap(); + let preprocessor = OpenAIPreprocessor::new(mdc).unwrap(); + let backend = Arc::new(RecordingBackend::default()); + let next: Arc< + dyn AsyncEngine, ManyOut>, Error>, + > = backend.clone(); + + let chat: NvCreateChatCompletionRequest = serde_json::from_value(serde_json::json!({ + "model": "test-model", + "messages": [{"role": "user", "content": "Hello"}], + "max_tokens": 1 + })) + .unwrap(); + let chat_timeline = Arc::new(TtftTimings::new(Instant::now())); + let mut chat_context = PipelineContext::new(chat); + chat_context.insert(TTFT_CONTEXT_KEY, chat_timeline.clone()); + let chat_stream = Operator::generate(preprocessor.as_ref(), chat_context, next.clone()) + .await + .unwrap(); + drop(chat_stream); + + let completion: NvCreateCompletionRequest = serde_json::from_value(serde_json::json!({ + "model": "test-model", "prompt": "Hello", "max_tokens": 1 + })) + .unwrap(); + let completion_timeline = Arc::new(TtftTimings::new(Instant::now())); + let mut completion_context = PipelineContext::new(completion); + completion_context.insert(TTFT_CONTEXT_KEY, completion_timeline.clone()); + let completion_stream = Operator::generate(preprocessor.as_ref(), completion_context, next) + .await + .unwrap(); + drop(completion_stream); + + let trackers = backend.0.lock().unwrap(); + assert_eq!(trackers.len(), 2); + for (tracker, timeline) in trackers.iter().zip([chat_timeline, completion_timeline]) { + assert!(Arc::ptr_eq(tracker.ttft().unwrap(), &timeline)); + let stages = timeline.finish(Instant::now()); + assert!(stages[TtftStage::TemplateTokenize as usize] > Duration::ZERO); + assert_eq!( + stages[TtftStage::MultimodalPrepare as usize], + Duration::ZERO + ); + assert_eq!(stages[TtftStage::MmRoutingBuild as usize], Duration::ZERO); + } + } +} diff --git a/lib/llm/src/protocols/common.rs b/lib/llm/src/protocols/common.rs index caca44f842ca..21b515d9c270 100644 --- a/lib/llm/src/protocols/common.rs +++ b/lib/llm/src/protocols/common.rs @@ -63,6 +63,7 @@ pub mod metrics; pub mod postprocessor; pub mod preprocessor; pub mod timing; +pub mod ttft; /// SamplingOptionsProvider is a trait that allows the caller to extract the sampling options from /// the object that implements it. This will mutate the object. diff --git a/lib/llm/src/protocols/common/timing.rs b/lib/llm/src/protocols/common/timing.rs index f16c625ca1ad..dc57be1c8174 100644 --- a/lib/llm/src/protocols/common/timing.rs +++ b/lib/llm/src/protocols/common/timing.rs @@ -88,6 +88,8 @@ impl std::fmt::Display for RequestPhase { /// accumulated detokenize timing, where lock-free updates are beneficial. #[derive(Debug)] pub struct RequestTracker { + /// Process-local frontend timeline, shared by preprocessing and routing. + ttft: OnceLock>, /// When the request was received (monotonic clock for duration calculations) request_received: Instant, @@ -224,6 +226,7 @@ impl RequestTracker { .unwrap_or(0); RequestTracker { + ttft: OnceLock::new(), request_received: now, request_received_epoch_ms: epoch_ms, prefill_start_time: OnceLock::new(), @@ -254,6 +257,15 @@ impl RequestTracker { } } + /// Attach the HTTP timeline without replacing an existing request timeline. + pub fn set_ttft(&self, timings: Arc) { + let _ = self.ttft.set(timings); + } + + pub fn ttft(&self) -> Option<&Arc> { + self.ttft.get() + } + /// Record when prefill started. Returns true if this was the first call. pub fn record_prefill_start(&self) -> bool { self.prefill_start_time.set(Instant::now()).is_ok() @@ -772,6 +784,20 @@ mod tests { use std::thread; use std::time::Duration; + #[test] + fn ttft_attachment_preserves_the_original_timeline_and_tracker_state() { + let tracker = RequestTracker::new(); + tracker.record_isl(12, Some(4)); + let timeline = Arc::new(super::super::ttft::TtftTimings::new(Instant::now())); + tracker.set_ttft(timeline.clone()); + tracker.set_ttft(Arc::new(super::super::ttft::TtftTimings::new( + Instant::now(), + ))); + assert!(Arc::ptr_eq(tracker.ttft().unwrap(), &timeline)); + assert_eq!(tracker.isl_tokens(), Some(12)); + assert_eq!(tracker.cached_tokens(), Some(4)); + } + #[test] fn test_record_isl_osl() { let tracker = RequestTracker::new(); diff --git a/lib/llm/src/protocols/common/ttft.rs b/lib/llm/src/protocols/common/ttft.rs new file mode 100644 index 000000000000..235a8ee45a32 --- /dev/null +++ b/lib/llm/src/protocols/common/ttft.rs @@ -0,0 +1,464 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Exclusive wall-clock attribution for the frontend's existing HTTP TTFT interval. +//! +//! Record only stage boundaries. At the first HTTP token, a single event sweep +//! partitions the interval; nested and concurrent spans never double-count time. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant}; + +use parking_lot::Mutex; + +/// Local pipeline context entry, shared with the Python processor and routed engine. +pub const TTFT_CONTEXT_KEY: &str = "frontend_ttft_timings"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(usize)] +pub enum TtftStage { + FrontendOverhead, + PythonPreprocess, + PreprocessQueue, + TemplateTokenize, + MultimodalPrepare, + MmRoutingBuild, + RoutePrefill, + RouteDecode, + RouteAggregated, + BackendPrefill, + BackendDecode, + BackendAggregated, + FirstTokenPrefill, + FirstTokenDecode, + FirstTokenAggregated, + ResponsePostprocess, + BackendOverlap, +} + +impl TtftStage { + pub const ALL: [Self; 17] = [ + Self::FrontendOverhead, + Self::PythonPreprocess, + Self::PreprocessQueue, + Self::TemplateTokenize, + Self::MultimodalPrepare, + Self::MmRoutingBuild, + Self::RoutePrefill, + Self::RouteDecode, + Self::RouteAggregated, + Self::BackendPrefill, + Self::BackendDecode, + Self::BackendAggregated, + Self::FirstTokenPrefill, + Self::FirstTokenDecode, + Self::FirstTokenAggregated, + Self::ResponsePostprocess, + Self::BackendOverlap, + ]; + + pub const fn labels(self) -> [&'static str; 2] { + match self { + Self::FrontendOverhead => ["frontend_overhead", ""], + Self::PythonPreprocess => ["python_preprocess", ""], + Self::PreprocessQueue => ["preprocess_queue", ""], + Self::TemplateTokenize => ["template_tokenize", ""], + Self::MultimodalPrepare => ["multimodal_prepare", ""], + Self::MmRoutingBuild => ["mm_routing_build", ""], + Self::RoutePrefill => ["kv_route", "prefill"], + Self::RouteDecode => ["kv_route", "decode"], + Self::RouteAggregated => ["kv_route", "aggregated"], + Self::BackendPrefill => ["backend_first_response", "prefill"], + Self::BackendDecode => ["backend_first_response", "decode"], + Self::BackendAggregated => ["backend_first_response", "aggregated"], + Self::FirstTokenPrefill => ["backend_first_token", "prefill"], + Self::FirstTokenDecode => ["backend_first_token", "decode"], + Self::FirstTokenAggregated => ["backend_first_token", "aggregated"], + Self::ResponsePostprocess => ["first_token_postprocess", ""], + Self::BackendOverlap => ["backend_overlap", "prefill_decode"], + } + } + + /// Restrict Python instrumentation to the fixed frontend stage vocabulary. + pub fn from_python(stage: &str) -> Option { + match stage { + "python_preprocess" => Some(Self::PythonPreprocess), + "preprocess_queue" => Some(Self::PreprocessQueue), + "template_tokenize" => Some(Self::TemplateTokenize), + "multimodal_prepare" => Some(Self::MultimodalPrepare), + "mm_routing_build" => Some(Self::MmRoutingBuild), + _ => None, + } + } +} + +#[derive(Debug)] +struct Interval { + stage: TtftStage, + start: Duration, + end: Option, +} + +/// Timing context for one HTTP request. No clocks or labels are updated per token. +#[derive(Debug)] +pub struct TtftTimings { + start: Instant, + intervals: Mutex>, + finished: AtomicBool, + result: OnceLock<[Duration; 17]>, +} + +impl TtftTimings { + pub fn new(start: Instant) -> Self { + Self { + start, + intervals: Mutex::new(Vec::with_capacity(16)), + finished: AtomicBool::new(false), + result: OnceLock::new(), + } + } + + pub fn start_span(self: &Arc, stage: TtftStage) -> TtftSpan { + if self.finished.load(Ordering::Acquire) { + return TtftSpan::inactive(); + } + let start = self.start.elapsed(); + let mut intervals = self.intervals.lock(); + if self.finished.load(Ordering::Acquire) { + return TtftSpan::inactive(); + } + let index = intervals.len(); + intervals.push(Interval { + stage, + start, + end: None, + }); + TtftSpan { + timings: Some(Arc::clone(self)), + index, + } + } + + /// Import a completed Python interval using ages relative to the call time. + /// + /// This avoids comparing Python and Rust monotonic clock epochs. Binding call + /// overhead shifts both boundaries equally; durations retain their precision. + pub fn record_ago(&self, stage: TtftStage, start_ago: Duration, end_ago: Duration) { + if start_ago < end_ago || self.finished.load(Ordering::Acquire) { + return; + } + let now = self.start.elapsed(); + let mut intervals = self.intervals.lock(); + if self.finished.load(Ordering::Acquire) { + return; + } + intervals.push(Interval { + stage, + start: now.saturating_sub(start_ago), + end: Some(now.saturating_sub(end_ago)), + }); + } + + /// Freeze at the HTTP first-token timestamp, including spans still in flight. + /// Repeated calls return the original result, regardless of their timestamp. + pub fn finish(&self, end: Instant) -> [Duration; 17] { + if let Some(result) = self.result.get() { + return *result; + } + let intervals = self.intervals.lock(); + if let Some(result) = self.result.get() { + return *result; + } + self.finished.store(true, Ordering::Release); + let result = partition(&intervals, end.saturating_duration_since(self.start)); + // The interval lock serializes finish calls and all interval mutations. + let _ = self.result.set(result); + result + } +} + +/// Close a stage on scope exit, or leave it active until HTTP observes a token. +#[derive(Debug)] +pub struct TtftSpan { + timings: Option>, + index: usize, +} + +impl TtftSpan { + fn inactive() -> Self { + Self { + timings: None, + index: 0, + } + } + + pub fn leave_open(mut self) { + self.timings.take(); + } +} + +impl Drop for TtftSpan { + fn drop(&mut self) { + let Some(timings) = &self.timings else { + return; + }; + if timings.finished.load(Ordering::Acquire) { + return; + } + let end = timings.start.elapsed(); + let mut intervals = timings.intervals.lock(); + if !timings.finished.load(Ordering::Acquire) { + intervals[self.index].end = Some(end); + } + } +} + +fn partition(intervals: &[Interval], total: Duration) -> [Duration; 17] { + let mut events = Vec::with_capacity(intervals.len() * 2); + for interval in intervals { + let start = interval.start.min(total); + let end = interval.end.unwrap_or(total).min(total); + if start < end { + events.push((start, interval.stage, true)); + events.push((end, interval.stage, false)); + } + } + events.sort_unstable_by_key(|event| event.0); + + let mut durations = [Duration::ZERO; 17]; + let mut active = [0usize; 17]; + let mut previous = Duration::ZERO; + for (time, stage, enters) in events { + durations[attribute(&active) as usize] += time - previous; + if enters { + active[stage as usize] += 1; + } else { + active[stage as usize] -= 1; + } + previous = time; + } + durations[attribute(&active) as usize] += total - previous; + durations +} + +fn attribute(active: &[usize; 17]) -> TtftStage { + use TtftStage::*; + // Explicit precedence keeps nested work exclusive. Decode wins rare routing + // ties; concurrent backend prefill/decode is attributed separately below. + for stage in [ + ResponsePostprocess, + RouteDecode, + RoutePrefill, + RouteAggregated, + TemplateTokenize, + MmRoutingBuild, + MultimodalPrepare, + PreprocessQueue, + PythonPreprocess, + ] { + if active[stage as usize] > 0 { + return stage; + } + } + let prefill = active[BackendPrefill as usize] > 0 || active[FirstTokenPrefill as usize] > 0; + let decode = active[BackendDecode as usize] > 0 || active[FirstTokenDecode as usize] > 0; + if (prefill && decode) || active[BackendOverlap as usize] > 0 { + return BackendOverlap; + } + for stage in [ + FirstTokenDecode, + BackendDecode, + FirstTokenPrefill, + BackendPrefill, + FirstTokenAggregated, + BackendAggregated, + ] { + if active[stage as usize] > 0 { + return stage; + } + } + FrontendOverhead +} + +#[cfg(test)] +mod tests { + use super::*; + use TtftStage::*; + + fn ms(value: u64) -> Duration { + Duration::from_millis(value) + } + + fn interval(stage: TtftStage, start: u64, end: u64) -> Interval { + Interval { + stage, + start: ms(start), + end: Some(ms(end)), + } + } + + fn assert_total(durations: &[Duration; 17], expected: u64) { + assert_eq!(durations.iter().sum::(), ms(expected)); + } + + #[test] + fn nested_preprocessing_is_exclusive_and_preserves_gaps() { + let result = partition( + &[ + interval(PythonPreprocess, 2, 18), + interval(PreprocessQueue, 3, 7), + interval(TemplateTokenize, 9, 15), + ], + ms(20), + ); + assert_eq!(result[FrontendOverhead as usize], ms(4)); + assert_eq!(result[PythonPreprocess as usize], ms(6)); + assert_eq!(result[PreprocessQueue as usize], ms(4)); + assert_eq!(result[TemplateTokenize as usize], ms(6)); + assert_eq!(result[MultimodalPrepare as usize], Duration::ZERO); + assert_total(&result, 20); + } + + #[test] + fn routing_metadata_is_exclusive_within_multimodal_preparation() { + let result = partition( + &[ + interval(PythonPreprocess, 0, 20), + interval(MultimodalPrepare, 2, 18), + interval(MmRoutingBuild, 6, 12), + ], + ms(20), + ); + assert_eq!(result[PythonPreprocess as usize], ms(4)); + assert_eq!(result[MultimodalPrepare as usize], ms(10)); + assert_eq!(result[MmRoutingBuild as usize], ms(6)); + assert_total(&result, 20); + } + + #[test] + fn overlapping_backend_phases_count_once_and_yield_to_routing() { + let result = partition( + &[ + interval(BackendPrefill, 2, 10), + interval(FirstTokenPrefill, 10, 20), + interval(BackendDecode, 6, 14), + interval(FirstTokenDecode, 14, 22), + interval(RouteDecode, 8, 12), + interval(ResponsePostprocess, 21, 24), + ], + ms(24), + ); + assert_eq!(result[FrontendOverhead as usize], ms(2)); + assert_eq!(result[BackendPrefill as usize], ms(4)); + assert_eq!(result[BackendDecode as usize], Duration::ZERO); + assert_eq!(result[FirstTokenPrefill as usize], Duration::ZERO); + assert_eq!(result[FirstTokenDecode as usize], ms(1)); + assert_eq!(result[BackendOverlap as usize], ms(10)); + assert_eq!(result[RouteDecode as usize], ms(4)); + assert_eq!(result[ResponsePostprocess as usize], ms(3)); + assert_total(&result, 24); + } + + #[test] + fn repeated_and_overlapping_attempts_of_one_stage_count_their_union() { + let result = partition( + &[ + interval(RouteAggregated, 1, 4), + interval(RouteAggregated, 3, 6), + interval(RouteAggregated, 7, 9), + interval(BackendAggregated, 6, 7), + ], + ms(10), + ); + assert_eq!(result[RouteAggregated as usize], ms(7)); + assert_eq!(result[BackendAggregated as usize], ms(1)); + assert_eq!(result[FrontendOverhead as usize], ms(2)); + assert_total(&result, 10); + } + + #[test] + fn intervals_are_clipped_to_http_ttft_and_unused_stages_are_zero() { + let result = partition( + &[ + interval(MultimodalPrepare, 0, 4), + interval(MmRoutingBuild, 4, 30), + interval(RoutePrefill, 20, 30), + ], + ms(10), + ); + assert_eq!(result[MultimodalPrepare as usize], ms(4)); + assert_eq!(result[MmRoutingBuild as usize], ms(6)); + for stage in TtftStage::ALL { + if stage != MultimodalPrepare && stage != MmRoutingBuild { + assert_eq!(result[stage as usize], Duration::ZERO); + } + } + assert_total(&result, 10); + assert_eq!(partition(&[], Duration::ZERO), [Duration::ZERO; 17]); + assert_eq!(partition(&[], ms(10))[FrontendOverhead as usize], ms(10)); + } + + #[test] + fn finish_freezes_open_spans_and_ignores_late_writes() { + let start = Instant::now(); + let timings = Arc::new(TtftTimings::new(start)); + timings.intervals.lock().push(Interval { + stage: BackendDecode, + start: ms(2), + end: None, + }); + let span = TtftSpan { + timings: Some(Arc::clone(&timings)), + index: 0, + }; + let result = timings.finish(start + ms(10)); + drop(span); + drop(timings.start_span(RouteDecode)); + timings.record_ago(MultimodalPrepare, ms(3), ms(1)); + assert_eq!(timings.intervals.lock().len(), 1); + assert!(timings.intervals.lock()[0].end.is_none()); + assert_eq!(timings.finish(start + ms(100)), result); + assert_eq!(result[BackendDecode as usize], ms(8)); + assert_total(&result, 10); + } + + #[test] + fn leaving_a_span_open_does_not_retain_the_context() { + let timings = Arc::new(TtftTimings::new(Instant::now())); + let span = timings.start_span(ResponsePostprocess); + assert_eq!(Arc::strong_count(&timings), 2); + span.leave_open(); + assert_eq!(Arc::strong_count(&timings), 1); + assert!(timings.intervals.lock()[0].end.is_none()); + } + + #[test] + fn python_stage_names_are_fixed_and_invalid_ages_are_rejected() { + for stage in [ + PythonPreprocess, + PreprocessQueue, + TemplateTokenize, + MultimodalPrepare, + MmRoutingBuild, + ] { + assert_eq!(TtftStage::from_python(stage.labels()[0]), Some(stage)); + } + assert_eq!(TtftStage::from_python("request_id"), None); + assert_eq!(TtftStage::from_python("backend_first_response"), None); + let timings = TtftTimings::new(Instant::now()); + timings.record_ago(PythonPreprocess, ms(1), ms(2)); + assert!(timings.intervals.lock().is_empty()); + } + + #[test] + fn imported_ages_preserve_duration_and_clip_before_request_start() { + let timings = TtftTimings::new(Instant::now() - ms(100)); + timings.record_ago(TemplateTokenize, ms(7), ms(2)); + timings.record_ago(MultimodalPrepare, Duration::MAX, Duration::MAX); + let intervals = timings.intervals.lock(); + assert_eq!(intervals[0].end.unwrap() - intervals[0].start, ms(5)); + assert_eq!(intervals[1].start, Duration::ZERO); + assert_eq!(intervals[1].end, Some(Duration::ZERO)); + } +} diff --git a/lib/runtime/src/config/environment_names.rs b/lib/runtime/src/config/environment_names.rs index 7ff6407ac0fc..18a7975d1fc6 100644 --- a/lib/runtime/src/config/environment_names.rs +++ b/lib/runtime/src/config/environment_names.rs @@ -579,6 +579,8 @@ pub mod llm { /// See [`DYN_METRICS_REQUEST_DURATION`]. pub const DYN_METRICS_TTFT: &str = "DYN_METRICS_TTFT"; /// See [`DYN_METRICS_REQUEST_DURATION`]. + pub const DYN_METRICS_TTFT_STAGE: &str = "DYN_METRICS_TTFT_STAGE"; + /// See [`DYN_METRICS_REQUEST_DURATION`]. pub const DYN_METRICS_ITL: &str = "DYN_METRICS_ITL"; /// Comma-separated inter-token latency histogram bucket boundaries. pub const DYN_METRICS_ITL_BUCKETS: &str = "DYN_METRICS_ITL_BUCKETS"; @@ -1189,6 +1191,7 @@ mod tests { llm::metrics::DYN_METRICS_INPUT_SEQUENCE, llm::metrics::DYN_METRICS_OUTPUT_SEQUENCE, llm::metrics::DYN_METRICS_TTFT, + llm::metrics::DYN_METRICS_TTFT_STAGE, llm::metrics::DYN_METRICS_ITL, llm::metrics::DYN_METRICS_ITL_BUCKETS, llm::metrics::DYN_METRICS_EMBEDDING_LATENCY, diff --git a/lib/runtime/src/metrics/prometheus_names.rs b/lib/runtime/src/metrics/prometheus_names.rs index 9bab99a1d364..e443e23dc2f7 100644 --- a/lib/runtime/src/metrics/prometheus_names.rs +++ b/lib/runtime/src/metrics/prometheus_names.rs @@ -254,6 +254,9 @@ pub mod frontend_service { /// Time to first token in seconds pub const TIME_TO_FIRST_TOKEN_SECONDS: &str = "time_to_first_token_seconds"; + /// Exclusive contributions to HTTP time to first token, labeled by stage and phase. + pub const TTFT_STAGE_SECONDS: &str = "ttft_stage_seconds"; + /// Inter-token latency in seconds pub const INTER_TOKEN_LATENCY_SECONDS: &str = "inter_token_latency_seconds"; From d80ee7b53deced7e64002d1f877ee3f8be114571 Mon Sep 17 00:00:00 2001 From: ssssnow <38564764+ssssnow@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:40:08 +0800 Subject: [PATCH 2/5] fix(frontend): address TTFT stage review feedback Signed-off-by: ssssnow <38564764+ssssnow@users.noreply.github.com> --- components/src/dynamo/frontend/sglang_processor.py | 5 ++--- .../src/dynamo/frontend/tests/test_ttft_stages.py | 14 ++++++++++++-- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/components/src/dynamo/frontend/sglang_processor.py b/components/src/dynamo/frontend/sglang_processor.py index 8666bf35ff5c..312ce48aee41 100644 --- a/components/src/dynamo/frontend/sglang_processor.py +++ b/components/src/dynamo/frontend/sglang_processor.py @@ -324,14 +324,13 @@ class SglangPreprocessWorkerResult: class _WorkerPreprocessTiming: - """Collect template boundaries in the preprocessing process pool.""" - def __init__(self) -> None: self.intervals_ns: list[tuple[int, int]] = [] @contextmanager def start_ttft_stage(self, stage: str): - assert stage == "template_tokenize" + if stage != "template_tokenize": + raise ValueError(f"Unsupported TTFT stage: {stage}") start_ns = time.monotonic_ns() try: yield diff --git a/components/src/dynamo/frontend/tests/test_ttft_stages.py b/components/src/dynamo/frontend/tests/test_ttft_stages.py index 481fc8c65966..cd92413aa305 100644 --- a/components/src/dynamo/frontend/tests/test_ttft_stages.py +++ b/components/src/dynamo/frontend/tests/test_ttft_stages.py @@ -225,6 +225,15 @@ def test_sglang_worker_returns_picklable_monotonic_intervals( ) +def test_sglang_worker_rejects_unsupported_ttft_stage(sglang_modules): + module, _ = sglang_modules + timing = module._WorkerPreprocessTiming() + with pytest.raises(ValueError, match="Unsupported TTFT stage: python_preprocess"): + with timing.start_ttft_stage("python_preprocess"): + pass + assert timing.intervals_ns == [] + + def test_sglang_pool_reattaches_worker_intervals(sglang_modules, monkeypatch): module, _ = sglang_modules context = RecordingContext() @@ -377,8 +386,9 @@ def vllm_modules(frontend_stubs): return processor, prepost -@pytest.mark.parametrize("pretokenized", [False, True]) -@pytest.mark.parametrize("fail_render", [False, True]) +@pytest.mark.parametrize( + "pretokenized,fail_render", [(False, False), (True, False), (False, True)] +) def test_vllm_template_tokenizer_preserves_tokens_and_closes_on_error( vllm_modules, monkeypatch, pretokenized, fail_render ): From 55fecadf1d7bea4e21029085e946035e9831a3b8 Mon Sep 17 00:00:00 2001 From: ssssnow <38564764+ssssnow@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:42:35 +0800 Subject: [PATCH 3/5] fix(observability): make TTFT breakdown logs opt-in Signed-off-by: ssssnow <38564764+ssssnow@users.noreply.github.com> --- .../pages/reference/observability/logging.mdx | 16 ++-- .../observability/metrics-catalog.mdx | 9 ++- lib/llm/src/http/service/metrics.rs | 74 +++++++++---------- 3 files changed, 49 insertions(+), 50 deletions(-) diff --git a/docs/fern/pages/reference/observability/logging.mdx b/docs/fern/pages/reference/observability/logging.mdx index da9fc1a589a3..8312869f8931 100644 --- a/docs/fern/pages/reference/observability/logging.mdx +++ b/docs/fern/pages/reference/observability/logging.mdx @@ -65,8 +65,8 @@ export DYN_LOG='info,dynamo_runtime::system_status_server=trace' ## Per-Request First-Token Log -For HTTP chat and text completion requests, the frontend emits one INFO event when it first observes a -nonzero output token count. The target is `dynamo_frontend::ttft` and the message is +When enabled, the frontend emits one DEBUG event for each HTTP chat or text completion request when +it first observes a nonzero output token count. The target is `dynamo_frontend::ttft` and the message is `frontend first token`. Requests that end, fail, or are canceled before that observation do not emit the event. Later tokens do not emit additional TTFT events. @@ -79,17 +79,19 @@ unused stages. These frozen values match the [TTFT stage histogram](metrics-catalog.mdx#ttft-stage-durations), and their sum equals `ttft_ms`, subject to floating-point rounding. -The default INFO level includes these events. To disable them while retaining metrics and other -INFO logs, set: +These events are disabled at the default INFO level. TTFT timing and Prometheus stage histograms +remain enabled regardless of the log filter. To enable TTFT events while keeping other logs at INFO, +set this environment variable before starting the frontend: ```bash -export DYN_LOG='info,dynamo_frontend::ttft=off' +export DYN_LOG='info,dynamo_frontend::ttft=debug' ``` -To retain these events with a WARN default, set: +A global DEBUG or TRACE level also enables these events. To disable them while retaining other +DEBUG logs, set: ```bash -export DYN_LOG='warn,dynamo_frontend::ttft=info' +export DYN_LOG='debug,dynamo_frontend::ttft=off' ``` Set `DYN_LOGGING_CONSOLE_FORMAT=jsonl` for machine-readable records; duration fields remain JSON diff --git a/docs/fern/pages/reference/observability/metrics-catalog.mdx b/docs/fern/pages/reference/observability/metrics-catalog.mdx index 9dbc3b661dec..7ff5bbe22e1e 100644 --- a/docs/fern/pages/reference/observability/metrics-catalog.mdx +++ b/docs/fern/pages/reference/observability/metrics-catalog.mdx @@ -158,10 +158,11 @@ HTTP first-token observation, including any remaining overlapping work. If that aborted or retried before HTTP observes a token, its postprocessing interval closes so the replacement attempt retains its own routing and backend contributions. -The metric is enabled by default on the frontend's existing `/metrics` endpoint. The SGLang and -vLLM Python chat processors and native Rust preprocessor provide preprocessing timings. Chat and -text completions carry the HTTP timing context through the in-process routing host, including KV, -round-robin, random, direct, and other built-in policies. These paths record worker selection, +The metric is enabled by default on the frontend's existing `/metrics` endpoint, independently of +the per-request TTFT log filter. The SGLang and vLLM Python chat processors and native Rust +preprocessor provide preprocessing timings. Chat and text completions carry the HTTP timing context +through the in-process routing host, including KV, round-robin, random, direct, and other built-in +policies. These paths record worker selection, backend waits, and first-token postprocessing. The `kv_route` label also covers the built-in policies. Batched text completion children share the parent HTTP timeline, so overlapping child work is counted once. diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index e89c94c05c80..d4a7a27be2b2 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -2068,14 +2068,15 @@ impl ResponseMetricCollector { } /// One structured event for the same frozen durations sent to Prometheus. - /// Its dedicated target can be disabled without disabling metrics. The - /// tracing macro evaluates numeric fields only when this event is enabled. + /// DEBUG keeps it off at the default INFO level; enable its dedicated target + /// with DYN_LOG=info,dynamo_frontend::ttft=debug. Metrics are independent, and + /// the tracing macro evaluates numeric fields only when this event is enabled. fn log_ttft_stages(&self, total: Duration, stages: &[Duration; TtftStage::ALL.len()]) { let Some(request_id) = self.ttft_request_id.as_deref() else { return; }; let ms = |stage: TtftStage| stages[stage as usize].as_secs_f64() * 1000.0; - tracing::info!( + tracing::debug!( target: "dynamo_frontend::ttft", request_id, model = self.model.as_str(), @@ -2917,15 +2918,16 @@ mod tests { /// or adding a test-only logging dependency. #[derive(Clone)] struct TtftLogSubscriber { - info_enabled: bool, + filter: tracing_subscriber::filter::Targets, events: Arc>>, } impl tracing::Subscriber for TtftLogSubscriber { fn enabled(&self, metadata: &tracing::Metadata<'_>) -> bool { - self.info_enabled - && metadata.target() == "dynamo_frontend::ttft" - && *metadata.level() == tracing::Level::INFO + metadata.target() == "dynamo_frontend::ttft" + && self + .filter + .would_enable(metadata.target(), metadata.level()) } fn register_callsite( @@ -2937,14 +2939,6 @@ mod tests { tracing::subscriber::Interest::sometimes() } - fn max_level_hint(&self) -> Option { - Some(if self.info_enabled { - tracing::level_filters::LevelFilter::INFO - } else { - tracing::level_filters::LevelFilter::OFF - }) - } - fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::span::Id { tracing::span::Id::from_u64(1) } @@ -2973,7 +2967,7 @@ mod tests { fn ttft_log_is_once_at_first_token_and_matches_histograms_and_request_identity() { let events = Arc::new(std::sync::Mutex::new(Vec::new())); let subscriber = TtftLogSubscriber { - info_enabled: true, + filter: "info,dynamo_frontend::ttft=debug".parse().unwrap(), events: events.clone(), }; let metrics = Arc::new(Metrics::new_with_prefix(None)); @@ -3035,7 +3029,7 @@ mod tests { "later tokens/drop/cancellation must not log" ); let (level, fields) = &events[0]; - assert_eq!(*level, tracing::Level::INFO); + assert_eq!(*level, tracing::Level::DEBUG); assert_eq!(fields.text["message"], "frontend first token"); assert_eq!(fields.text["request_id"], "client-request-42"); assert_eq!(fields.text["model"], model); @@ -3073,28 +3067,30 @@ mod tests { } #[test] - fn ttft_log_disabled_does_not_change_metrics() { - let events = Arc::new(std::sync::Mutex::new(Vec::new())); - let subscriber = TtftLogSubscriber { - info_enabled: false, - events: events.clone(), - }; - let metrics = Arc::new(Metrics::new_with_prefix(None)); - let model = "ttft-disabled-log-model"; - tracing::subscriber::with_default(subscriber, || { - let mut collector = metrics.clone().create_response_collector(model); - let mut request = dynamo_runtime::pipeline::Context::new(()); - collector.attach_ttft_timeline(&mut request); - let timeline = request.get::>(TTFT_CONTEXT_KEY).unwrap(); - let route = timeline.start_span(TtftStage::RouteAggregated); - collector.observe_response(20, 0); - collector.observe_response(20, 1); - drop(route); - collector.observe_response(20, 2); - drop(collector); - }); - assert!(events.lock().unwrap().is_empty()); - assert_ttft_stage_cohort(&metrics, model, 1); + fn ttft_log_disabled_by_default_or_explicit_filter_does_not_change_metrics() { + for directive in ["info", "debug,dynamo_frontend::ttft=off"] { + let events = Arc::new(std::sync::Mutex::new(Vec::new())); + let subscriber = TtftLogSubscriber { + filter: directive.parse().unwrap(), + events: events.clone(), + }; + let metrics = Arc::new(Metrics::new_with_prefix(None)); + let model = "ttft-disabled-log-model"; + tracing::subscriber::with_default(subscriber, || { + let mut collector = metrics.clone().create_response_collector(model); + let mut request = dynamo_runtime::pipeline::Context::new(()); + collector.attach_ttft_timeline(&mut request); + let timeline = request.get::>(TTFT_CONTEXT_KEY).unwrap(); + let route = timeline.start_span(TtftStage::RouteAggregated); + collector.observe_response(20, 0); + collector.observe_response(20, 1); + drop(route); + collector.observe_response(20, 2); + drop(collector); + }); + assert!(events.lock().unwrap().is_empty()); + assert_ttft_stage_cohort(&metrics, model, 1); + } } fn model_ready_value_with_name( From c96342b46c39a8d251a2da4ddc5887add079c0f7 Mon Sep 17 00:00:00 2001 From: ssssnow <38564764+ssssnow@users.noreply.github.com> Date: Fri, 23 Oct 2026 17:39:26 +0800 Subject: [PATCH 4/5] fix(observability): repair TTFT test builds Pass the optional request tracker in the mm-routing regression test. Enable PyO3 extension-module through maturin configuration rather than unconditionally, so cargo test links libpython for native timing tests while Python wheels retain extension-module linking semantics. Signed-off-by: ssssnow <38564764+ssssnow@users.noreply.github.com> --- lib/bindings/python/Cargo.toml | 4 ++-- lib/bindings/python/pyproject.toml | 2 ++ lib/llm/src/preprocessor/mm_routing/image.rs | 2 +- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/lib/bindings/python/Cargo.toml b/lib/bindings/python/Cargo.toml index 0605704716a3..83c0beece6ae 100644 --- a/lib/bindings/python/Cargo.toml +++ b/lib/bindings/python/Cargo.toml @@ -102,13 +102,13 @@ dynamo-worker-selection-policy-catalog = { path = "../../router-plugins/catalog" clap = { version = "4.5", features = ["derive"], optional = true } tracing-subscriber = { version = "0.3", features = ["env-filter"], optional = true } -# "extension-module" tells pyo3 we want to build an extension module (skips linking against libpython.so) +# Maturin enables "extension-module" in pyproject.toml for wheel builds. +# Plain cargo tests must link libpython to execute native Python API tests. # "abi3-py311" tells pyo3 (and maturin) to build using the stable ABI with minimum Python version 3.11, which is the minimum version in pyproject.toml pyo3 = { version = "0.23.4", default-features = false, features = [ "macros", "experimental-async", "experimental-inspect", - "extension-module", "py-clone", "abi3-py311" ] } diff --git a/lib/bindings/python/pyproject.toml b/lib/bindings/python/pyproject.toml index 019cc9094bb6..7fb5f5e93acc 100644 --- a/lib/bindings/python/pyproject.toml +++ b/lib/bindings/python/pyproject.toml @@ -45,6 +45,8 @@ classifiers = [ keywords = ["llm", "genai", "inference", "nvidia", "distributed", "dynamo"] [tool.maturin] +# Skip libpython linkage for extension builds, but retain it for cargo tests. +features = ["pyo3/extension-module"] module-name = "dynamo._core" manifest-path = "Cargo.toml" python-packages = ["dynamo"] diff --git a/lib/llm/src/preprocessor/mm_routing/image.rs b/lib/llm/src/preprocessor/mm_routing/image.rs index 4ca9c08d5a9e..d1c995867e1a 100644 --- a/lib/llm/src/preprocessor/mm_routing/image.rs +++ b/lib/llm/src/preprocessor/mm_routing/image.rs @@ -701,7 +701,7 @@ mod tests { .sampling_options(Default::default()) .output_options(Default::default()); let (_, image_tokens) = preprocessor - .gather_multi_modal_data_with_image_tokens(&request, &mut builder, None, &ids) + .gather_multi_modal_data_with_image_tokens(&request, &mut builder, None, &ids, None) .await .unwrap(); let result = builder.build().unwrap(); From 5ea5210a8fdda5673b85539e8e715cf2f174e910 Mon Sep 17 00:00:00 2001 From: ssssnow <38564764+ssssnow@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:56:24 +0800 Subject: [PATCH 5/5] test(observability): isolate TTFT metric compatibility snapshots Signed-off-by: ssssnow <38564764+ssssnow@users.noreply.github.com> --- lib/llm/src/http/service/metrics.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/lib/llm/src/http/service/metrics.rs b/lib/llm/src/http/service/metrics.rs index 3687c1f9144e..e42260c909d8 100644 --- a/lib/llm/src/http/service/metrics.rs +++ b/lib/llm/src/http/service/metrics.rs @@ -3547,6 +3547,17 @@ mod tests { let metrics = Arc::new(Metrics::build_with_detail(None, &fake_env(&[]), mode)); let registry = Registry::new(); metrics.register(®istry).unwrap(); + // Registries otherwise share this counter's lazily created label sets + // with concurrent error-rendering tests. Keep the same metric schema + // in the snapshot, but give each mode its own counter instance. + registry + .unregister(Box::new(DYNAM_FAILURES_TOTAL.clone())) + .unwrap(); + let failures = new_failure_counter(None); + failures + .with_label_values(&["InvalidRequest", "request.invalid"]) + .inc(); + registry.register(Box::new(failures)).unwrap(); for emits_token in [false, true] { let mut collector = response_collector(&metrics, "compatibility"); let mut request = dynamo_runtime::pipeline::Context::new(()); @@ -6268,6 +6279,14 @@ mod tests { .into_iter() .find(|family| family.name() == "dynamo_frontend_failures_total") .expect("dynamo_frontend_failures_total metric"); + assert_eq!( + family.get_field_type(), + prometheus::proto::MetricType::COUNTER + ); + assert_eq!( + family.help(), + "Total number of terminal semantic request failures" + ); let metric = family .get_metric() .iter()