test(sidecar): cover DEP rank-aware KV routing

.github/workflows/nightly-ci.yml

@@ -903,19 +903,23 @@ jobs:
     name: sidecar-runtime
     needs: [vllm-build, sidecar-build, resolve-source-sha, compute-release-mode]
     if: ${{ needs.compute-release-mode.outputs.run_tests == 'true' }}
+    strategy:
+      fail-fast: false
+      matrix:
+        gpu_count: [1, 2]
     uses: $/.github/workflows/shared-test.yml
     with:
       dd_env: nightly
       dd_flaky_retry_enabled: ${{ 'false' }}
       test_suite_name: vllm
       framework: vllm
-      test_type: Sidecar E2E Test
-      amd_runner: prod-tester-amd-gpu-v2
+      test_type: ${{ format('vLLM Sidecar {0}-GPU E2E Test', matrix.gpu_count) }}
+      amd_runner: ${{ matrix.gpu_count == 2 && 'prod-tester-amd-gpu-2-v2' || 'prod-tester-amd-gpu-v2' }}
       target_tag_plain: ${{ needs.vllm-build.outputs.target_tag_plain }}
       cuda_version: '["13.0"]'
       platform: '["amd64"]'
       run_sanity_check: false
-      gpu_test_markers: sidecar and vllm and gpu_1
+      gpu_test_markers: ${{ format('sidecar and vllm and gpu_{0}', matrix.gpu_count) }}
       gpu_test_timeout_minutes: 45
       source_ref: ${{ needs.resolve-source-sha.outputs.source_sha }}
       image_tag_suffix: '-nightly'
@@ -927,19 +931,23 @@ jobs:
     name: sidecar-runtime
     needs: [sglang-build, sidecar-build, resolve-source-sha, compute-release-mode]
     if: ${{ needs.compute-release-mode.outputs.run_tests == 'true' }}
+    strategy:
+      fail-fast: false
+      matrix:
+        gpu_count: [1, 2]
     uses: $/.github/workflows/shared-test.yml
     with:
       dd_env: nightly
       dd_flaky_retry_enabled: ${{ 'false' }}
       test_suite_name: sglang
       framework: sglang
-      test_type: Sidecar E2E Test
-      amd_runner: prod-tester-amd-gpu-v2
+      test_type: ${{ format('SGLang Sidecar {0}-GPU E2E Test', matrix.gpu_count) }}
+      amd_runner: ${{ matrix.gpu_count == 2 && 'prod-tester-amd-gpu-2-v2' || 'prod-tester-amd-gpu-v2' }}
       target_tag_plain: ${{ needs.sglang-build.outputs.target_tag_plain }}
       cuda_version: '["13.0"]'
       platform: '["amd64"]'
       run_sanity_check: false
-      gpu_test_markers: sidecar and sglang and gpu_1
+      gpu_test_markers: ${{ format('sidecar and sglang and gpu_{0}', matrix.gpu_count) }}
       gpu_test_timeout_minutes: 45
       source_ref: ${{ needs.resolve-source-sha.outputs.source_sha }}
       image_tag_suffix: '-nightly'
@@ -998,7 +1006,7 @@ jobs:
       platform: '["amd64"]' # No ARM GPUs available
       enable_coverage: true
       run_sanity_check: false
-      gpu_test_markers: vllm and gpu_2 and not h100
+      gpu_test_markers: vllm and gpu_2 and not h100 and not sidecar
       gpu_test_timeout_minutes: 45
       source_ref: ${{ needs.resolve-source-sha.outputs.source_sha }}
       image_tag_suffix: '-nightly'
@@ -1102,7 +1110,7 @@ jobs:
       platform: '["amd64"]' # No ARM GPUs available
       enable_coverage: true
       run_sanity_check: false
-      gpu_test_markers: sglang and gpu_2 and not h100
+      gpu_test_markers: sglang and gpu_2 and not h100 and not sidecar
       gpu_test_timeout_minutes: 45
       source_ref: ${{ needs.resolve-source-sha.outputs.source_sha }}
       image_tag_suffix: '-nightly'

.github/workflows/post-merge-ci.yml

@@ -475,18 +475,22 @@ jobs:
   sidecar-vllm-test:
     name: sidecar-runtime
     needs: [vllm-build, sidecar-build]
+    strategy:
+      fail-fast: false
+      matrix:
+        gpu_count: [1, 2]
     uses: $/.github/workflows/shared-test.yml
     with:
       dd_env: post-merge
       test_suite_name: vllm
       framework: vllm
-      test_type: Sidecar E2E Test
-      amd_runner: prod-tester-amd-gpu-v2
+      test_type: ${{ format('vLLM Sidecar {0}-GPU E2E Test', matrix.gpu_count) }}
+      amd_runner: ${{ matrix.gpu_count == 2 && 'prod-tester-amd-gpu-2-v2' || 'prod-tester-amd-gpu-v2' }}
       target_tag_plain: ${{ needs.vllm-build.outputs.target_tag_plain }}
       cuda_version: '["13.0"]'
       platform: '["amd64"]'
       run_sanity_check: false
-      gpu_test_markers: (pre_merge or post_merge) and sidecar and vllm and gpu_1
+      gpu_test_markers: ${{ format('(pre_merge or post_merge) and sidecar and vllm and gpu_{0}', matrix.gpu_count) }}
       gpu_test_timeout_minutes: 45
       sidecar_backend: vllm
       sidecar_binary_artifact: ${{ needs.sidecar-build.outputs.binary_artifact }}
@@ -495,18 +499,22 @@ jobs:
   sidecar-sglang-test:
     name: sidecar-runtime
     needs: [sglang-build, sidecar-build]
+    strategy:
+      fail-fast: false
+      matrix:
+        gpu_count: [1, 2]
     uses: $/.github/workflows/shared-test.yml
     with:
       dd_env: post-merge
       test_suite_name: sglang
       framework: sglang
-      test_type: Sidecar E2E Test
-      amd_runner: prod-tester-amd-gpu-v2
+      test_type: ${{ format('SGLang Sidecar {0}-GPU E2E Test', matrix.gpu_count) }}
+      amd_runner: ${{ matrix.gpu_count == 2 && 'prod-tester-amd-gpu-2-v2' || 'prod-tester-amd-gpu-v2' }}
       target_tag_plain: ${{ needs.sglang-build.outputs.target_tag_plain }}
       cuda_version: '["13.0"]'
       platform: '["amd64"]'
       run_sanity_check: false
-      gpu_test_markers: (pre_merge or post_merge) and sidecar and sglang and gpu_1
+      gpu_test_markers: ${{ format('(pre_merge or post_merge) and sidecar and sglang and gpu_{0}', matrix.gpu_count) }}
       gpu_test_timeout_minutes: 45
       sidecar_backend: sglang
       sidecar_binary_artifact: ${{ needs.sidecar-build.outputs.binary_artifact }}
@@ -553,7 +561,7 @@ jobs:
       cuda_version: '["13.0"]'
       platform: '["amd64"]' # No ARM GPUs available
       run_sanity_check: false
-      gpu_test_markers: (pre_merge or post_merge) and vllm and gpu_2
+      gpu_test_markers: (pre_merge or post_merge) and vllm and gpu_2 and not sidecar
       gpu_test_timeout_minutes: 60
     secrets: inherit
 
@@ -634,7 +642,7 @@ jobs:
       cuda_version: '["13.0"]'
       platform: '["amd64"]' # No ARM GPUs available
       run_sanity_check: false
-      gpu_test_markers: (pre_merge or post_merge) and sglang and gpu_2
+      gpu_test_markers: (pre_merge or post_merge) and sglang and gpu_2 and not sidecar
       gpu_test_timeout_minutes: 60
     secrets: inherit
 

.github/workflows/pr.yaml

@@ -374,6 +374,10 @@ jobs:
   sidecar-vllm-test:
     name: sidecar-runtime # This name overlaps with other sidecar jobs to group them in the UI
     needs: [changed-files, vllm-build, sidecar-build]
+    strategy:
+      fail-fast: false
+      matrix:
+        gpu_count: ${{ fromJSON(needs.changed-files.outputs.run_multigpu_tests == 'true' && '[1,2]' || '[1]') }}
     # vllm-build must have succeeded or been intentionally skipped (a
     # lib/sidecar-only PR) -- not failed or cancelled -- or the fallback below
     # would silently test this PR's sidecar binary against main's vllm image
@@ -386,14 +390,14 @@ jobs:
     with:
       test_suite_name: vllm
       framework: vllm  # matches the framework passed to vllm-build
-      test_type: Sidecar E2E Test
-      amd_runner: prod-tester-amd-gpu-v2
+      test_type: ${{ format('Sidecar {0}-GPU E2E Test', matrix.gpu_count) }}
+      amd_runner: ${{ matrix.gpu_count == 2 && 'prod-tester-amd-gpu-2-v2' || 'prod-tester-amd-gpu-v2' }}
       target_tag_plain: ${{ needs.vllm-build.result == 'success' && needs.vllm-build.outputs.target_tag_plain || 'vllm-runtime' }}
       source_ref: ${{ needs.vllm-build.result == 'success' && github.sha || 'main' }}
       cuda_version: '["13.0"]'
       platform: '["amd64"]'
       run_sanity_check: false
-      gpu_test_markers: pre_merge and sidecar and vllm and gpu_1
+      gpu_test_markers: ${{ format('pre_merge and sidecar and vllm and gpu_{0}', matrix.gpu_count) }}
       gpu_test_timeout_minutes: 45
       sidecar_backend: vllm
       sidecar_binary_artifact: ${{ needs.sidecar-build.outputs.binary_artifact }}
@@ -402,6 +406,10 @@ jobs:
   sidecar-sglang-test:
     name: sidecar-runtime # This name overlaps with other sidecar jobs to group them in the UI
     needs: [changed-files, sglang-build, sidecar-build]
+    strategy:
+      fail-fast: false
+      matrix:
+        gpu_count: ${{ fromJSON(needs.changed-files.outputs.run_multigpu_tests == 'true' && '[1,2]' || '[1]') }}
     # See sidecar-vllm-test above: sglang-build must have succeeded or been
     # intentionally skipped, not failed or cancelled.
     if: ${{ !cancelled() && needs.changed-files.outputs.sidecar == 'true' && needs.sidecar-build.result == 'success' && (needs.sglang-build.result == 'success' || needs.sglang-build.result == 'skipped') }}
@@ -412,14 +420,14 @@ jobs:
     with:
       test_suite_name: sglang
       framework: sglang  # matches the framework passed to sglang-build
-      test_type: Sidecar E2E Test
-      amd_runner: prod-tester-amd-gpu-v2
+      test_type: ${{ format('Sidecar {0}-GPU E2E Test', matrix.gpu_count) }}
+      amd_runner: ${{ matrix.gpu_count == 2 && 'prod-tester-amd-gpu-2-v2' || 'prod-tester-amd-gpu-v2' }}
       target_tag_plain: ${{ needs.sglang-build.result == 'success' && needs.sglang-build.outputs.target_tag_plain || 'sglang-runtime' }}
       source_ref: ${{ needs.sglang-build.result == 'success' && github.sha || 'main' }}
       cuda_version: '["13.0"]'
       platform: '["amd64"]'
       run_sanity_check: false
-      gpu_test_markers: pre_merge and sidecar and sglang and gpu_1
+      gpu_test_markers: ${{ format('pre_merge and sidecar and sglang and gpu_{0}', matrix.gpu_count) }}
       gpu_test_timeout_minutes: 45
       sidecar_backend: sglang
       sidecar_binary_artifact: ${{ needs.sidecar-build.outputs.binary_artifact }}
@@ -934,7 +942,7 @@ jobs:
       cuda_version: '["13.0"]'
       platform: '["amd64"]' # No ARM GPUs available
       run_sanity_check: false
-      gpu_test_markers: pre_merge and vllm and gpu_2
+      gpu_test_markers: pre_merge and vllm and gpu_2 and not sidecar
       gpu_test_timeout_minutes: 60
     secrets: inherit
 
@@ -982,7 +990,7 @@ jobs:
       cuda_version: '["13.0"]'
       platform: '["amd64"]' # No ARM GPUs available
       run_sanity_check: false
-      gpu_test_markers: pre_merge and sglang and gpu_2
+      gpu_test_markers: pre_merge and sglang and gpu_2 and not sidecar
       gpu_test_timeout_minutes: 60
     secrets: inherit
 

lib/sidecar/vllm/launch/agg.sh

@@ -42,6 +42,7 @@ while [[ $# -gt 0 ]]; do
             echo "  DYN_SYSTEM_PORT         Dynamo sidecar system port (default: 8081)"
             echo "  VLLM_RS_HTTP_PORT       vLLM HTTP port (default: 8100)"
             echo "  VLLM_GRPC_PORT          vLLM gRPC port (default: 50051)"
+            echo "  VLLM_DATA_PARALLEL_SIZE Number of local data-parallel ranks (default: 1)"
             echo "  MAX_MODEL_LEN           Maximum model length (default: 4096)"
             echo "  MAX_CONCURRENT_SEQS     Maximum concurrent sequences (default: 2)"
             echo "  DEFAULT_KV_CACHE_BYTES  KV cache cap when not profiling (default: 1119388000)"
@@ -85,6 +86,7 @@ vllm-rs serve "$MODEL" \
     --host 127.0.0.1 \
     --port "$VLLM_RS_HTTP_PORT" \
     --grpc-port "$VLLM_GRPC_PORT" \
+    --data-parallel-size "${VLLM_DATA_PARALLEL_SIZE:-1}" \
     --max-model-len "$MAX_MODEL_LEN" \
     -- \
     --enforce-eager \

tests/router/common.py

@@ -558,9 +558,10 @@ def _test_frontend_kv_routing(
     namespace: str,
     model_name: str,
     block_size: int,
+    dp_ranks: tuple[int, ...] = (0,),
 ) -> None:
-    """Verify engine events drive HTTP routing to two independently warmed workers."""
-    assert len(system_ports) == 2
+    """Verify engine events drive HTTP routing to two independently warmed ranks."""
+    assert len(system_ports) * len(dp_ranks) == 2
     url = f"http://localhost:{frontend_port}/v1/chat/completions"
     prompts = [
         "Amber rabbits explore quiet meadows. " * 96,
@@ -571,10 +572,13 @@ async def run_test() -> None:
         with managed_runtime() as runtime:
             worker_ids = sorted(
                 await poll_for_worker_instances(
-                    runtime.endpoint(f"{namespace}.backend.generate"), 2
+                    runtime.endpoint(f"{namespace}.backend.generate"), len(system_ports)
                 )
             )
-            assert len(worker_ids) == 2, worker_ids
+            assert len(worker_ids) == len(system_ports), worker_ids
+            targets = [
+                (worker_id, rank) for worker_id in worker_ids for rank in dp_ranks
+            ]
 
             async with aiohttp.ClientSession(
                 timeout=aiohttp.ClientTimeout(total=30)
@@ -584,9 +588,9 @@ async def send(
                     prompt: str,
                     *,
                     is_query_only: bool = False,
-                    worker_id: int | None = None,
-                ) -> tuple[int, float | None]:
-                    """Send one request and return its selected worker and KV hit rate."""
+                    target: tuple[int, int] | None = None,
+                ) -> tuple[tuple[int, int], float | None]:
+                    """Send one request and return its selected target and KV hit rate."""
                     payload = {
                         "model": model_name,
                         "messages": [{"role": "user", "content": prompt}],
@@ -602,20 +606,25 @@ async def send(
                     }
                     headers = (
                         {
-                            "x-dynamo-worker-instance-id": str(worker_id),
-                            "x-dynamo-dp-rank": "0",
+                            "x-dynamo-worker-instance-id": str(target[0]),
+                            "x-dynamo-dp-rank": str(target[1]),
                         }
-                        if worker_id is not None
+                        if target is not None
                         else None
                     )
                     nvext, has_generated_text = await send_router_chat_request(
                         session, url, payload, headers
                     )
                     selected = require_router_worker_id({"nvext": nvext})
-                    selected_id = selected["decode_worker_id"]
-                    assert selected_id in worker_ids, selected
-                    assert selected["prefill_worker_id"] == selected_id, selected
-                    assert selected["decode_dp_rank"] == 0, selected
+                    selected_target = (
+                        selected["decode_worker_id"],
+                        selected["decode_dp_rank"],
+                    )
+                    assert selected_target in targets, selected
+                    assert (
+                        selected["prefill_worker_id"],
+                        selected["prefill_dp_rank"],
+                    ) == selected_target, selected
                     hit_rate = nvext.get("timing", {}).get("kv_hit_rate")
                     if is_query_only:
                         assert not has_generated_text, nvext
@@ -626,26 +635,26 @@ async def send(
                         ), "Request completed without generating text"
                         assert isinstance(hit_rate, (int, float)), nvext
                         assert 0 <= hit_rate <= 1, nvext
-                    return selected_id, hit_rate
+                    return selected_target, hit_rate
 
                 baselines = {
                     port: await get_stored_kv_event_counts(session, port)
                     for port in system_ports
                 }
                 for prompt in prompts:
                     await send(prompt, is_query_only=True)
-                for prompt, worker_id in zip(prompts, worker_ids):
-                    selected, _ = await send(prompt, worker_id=worker_id)
-                    assert selected == worker_id, (selected, worker_id)
+                for prompt, target in zip(prompts, targets):
+                    selected, _ = await send(prompt, target=target)
+                    assert selected == target, (selected, target)
 
                 deadline = time.monotonic() + 60
                 observed = []
                 counts = {}
                 while time.monotonic() < deadline:
-                    # Pinned completions expose timing without warming the other worker.
+                    # Pinned completions expose timing without warming the other target.
                     observed = [
-                        await send(prompt, worker_id=worker_id)
-                        for prompt, worker_id in zip(prompts, worker_ids)
+                        await send(prompt, target=target)
+                        for prompt, target in zip(prompts, targets)
                     ]
                     counts = {
                         port: await get_stored_kv_event_counts(session, port)
@@ -655,7 +664,7 @@ async def send(
                         selected == expected
                         and hit_rate is not None
                         and hit_rate >= 0.5
-                        for (selected, hit_rate), expected in zip(observed, worker_ids)
+                        for (selected, hit_rate), expected in zip(observed, targets)
                     ) and all(
                         all(
                             current > baseline
@@ -667,16 +676,20 @@ async def send(
                     await asyncio.sleep(0.1)
                 else:
                     raise AssertionError(
-                        f"KV events did not converge: expected workers={worker_ids}, "
+                        f"KV events did not converge: expected targets={targets}, "
                         f"routing={observed}, Stored counters={counts}, baselines={baselines}"
                     )
 
+                for prompt, target in zip(prompts, targets):
+                    selected, _ = await send(prompt, is_query_only=True)
+                    assert selected == target, (selected, target)
+
                 for prompt_index in (0, 0, 1, 0, 1, 1):
                     selected, hit_rate = await send(prompts[prompt_index])
-                    assert selected == worker_ids[prompt_index], (
+                    assert selected == targets[prompt_index], (
                         prompt_index,
                         selected,
-                        worker_ids,
+                        targets,
                     )
                     assert hit_rate is not None and hit_rate >= 0.5, (
                         prompt_index,