From 2198ae6a211005a92ed6f036ce0e654723810b08 Mon Sep 17 00:00:00 2001 From: Michal Guzek Date: Sun, 1 Nov 2026 14:19:13 -0800 Subject: [PATCH] [TRTLLM-17163][test] Muse Spark Open stress settings and disagg cancellation marathon attach mode - stress_test.py: a Muse-Spark-Open-816B-A42B-NVFP4 TP4 case with the model's single-node GB300 recipe (TP4/EP4, FP8 KV cache with block reuse, chunked prefill, TRTLLM attention and MoE, NCCL all-reduce, CUDA graph padding), the large-MoE timeouts, a 0.4 KV cache fraction and a 30 req/s cap on the stress stage's request count. - The disaggregated-cancellation marathon can attach to an externally launched cluster (stress_config.external_cluster): the harness probes the given orchestrator and tails the given worker logs instead of launching workers. Attach mode supports log_only only. - marathon_python_v2_musespark_open_attach.yaml: Muse Spark Open 1P1D in attach mode, skipped unless the cluster's environment variables are set. - Tests for the attach mode's parsing, validation, setup and log scanning. Signed-off-by: Michal Guzek --- .../disagg_cancel/configs/README.md | 52 ++++ ...athon_python_v2_musespark_open_attach.yaml | 62 +++++ .../defs/stress_test/disagg_cancel/harness.py | 262 ++++++++++++++++++ .../test_disagg_cancel_stress.py | 147 ++++++++++ .../defs/stress_test/stress_test.py | 76 ++++- 5 files changed, 591 insertions(+), 8 deletions(-) create mode 100644 tests/integration/defs/stress_test/disagg_cancel/configs/marathon_python_v2_musespark_open_attach.yaml diff --git a/tests/integration/defs/stress_test/disagg_cancel/configs/README.md b/tests/integration/defs/stress_test/disagg_cancel/configs/README.md index 0d875d9ad8d8..fce39b2ba0dd 100644 --- a/tests/integration/defs/stress_test/disagg_cancel/configs/README.md +++ b/tests/integration/defs/stress_test/disagg_cancel/configs/README.md @@ -5,6 +5,7 @@ Per-marathon YAML files consumed by ``test_disagg_cancel_stress.py``. | File | Model | KV cache | Transceiver | |------|-------|----------|-------------| | `marathon_python_v2_deepseek.yaml` | `DeepSeek-V3-Lite/bf16` (MLA) | V2 | Python | +| `marathon_python_v2_musespark_open_attach.yaml` | `musespark_open` (served name; attaches to an externally launched 1P1D cluster) | V2 | Python | Only configurations listed in `_MARATHON_CONFIGS` in `test_disagg_cancel_stress.py` are actually parametrized; the other @@ -43,6 +44,57 @@ When switching from `log_only` to `full_cancel_poison`, update the test-list timeout together. The top-level README has the exact checklist. +## Attach mode: `stress_config.external_cluster` + +By default the harness launches the disaggregated cluster locally via +`setup_disagg_cluster` (one node, `Popen` workers). Models whose +per-instance GPU footprint exceeds one node (e.g. TP4 instances on +4-GPU nodes, where 1P1D needs 8 GPUs) cannot be launched that way; for +those, boot the cluster externally (one node per instance) and let the +marathon **attach** to it: + +```yaml +stress_config: + mode: log_only # full_cancel_poison is rejected (see below) + external_cluster: + server_url: ${TRTLLM_TEST_EXTERNAL_CLUSTER_URL} + server_log: ${TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR}/orch/server.log # optional + worker_logs: # required, >=1 entry; feeds the log scanner + - role: ctx # ctx | gen; per-role index = YAML order + path: ${TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR}/ctx/server.log + - role: gen + path: ${TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR}/gen1/server.log + worker_health_urls: [] # optional; each is GET-checked once at setup() + model_name: musespark_open # optional; falls back to top-level `model` +``` + +When the block is present, `harness.setup()` skips the local launcher +entirely: it binds the orchestrator endpoint for the probe thread, +verifies `server_url` + `worker_health_urls` respond with HTTP 200, +and registers the log paths for the log-scanner thread. The attach +test itself needs **no GPUs** — only HTTP reachability to the +orchestrator and filesystem access to the worker logs (a shared +filesystem when the cluster spans nodes). + +**Endpoint indirection is via environment variables**, not YAML +templating: string fields may contain `${VAR}` references, expanded by +the harness at `setup()` time. An unset variable is a hard error at +setup, but parse/validation is env-independent, so checked-in attach +YAMLs pass `test_all_marathon_yamls_parse_and_validate` (and pytest +collection) with nothing exported; the parametrized marathon test +**skips** when a referenced variable is unset. The conventional +variables used by the shipped attach YAML are +`TRTLLM_TEST_EXTERNAL_CLUSTER_URL` (orchestrator base URL) and +`TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR` (serve directory containing the +worker/orchestrator logs). + +Attach mode supports `mode: log_only` only. `full_cancel_poison` is +rejected by `StressConfig.validate()` because fault injection signals +locally-launched worker subprocesses by PID and respawns them through +the shared launcher, neither of which exists for an external cluster. +Teardown of the external cluster is likewise the launcher's job, not +the harness's — `harness.stop()` only stops its own threads. + ## Backend-knob axis: KV-cache manager × transceiver runtime The registered marathon config explicitly selects `NIXL` with the `PYTHON` diff --git a/tests/integration/defs/stress_test/disagg_cancel/configs/marathon_python_v2_musespark_open_attach.yaml b/tests/integration/defs/stress_test/disagg_cancel/configs/marathon_python_v2_musespark_open_attach.yaml new file mode 100644 index 000000000000..5651f0c9e334 --- /dev/null +++ b/tests/integration/defs/stress_test/disagg_cancel/configs/marathon_python_v2_musespark_open_attach.yaml @@ -0,0 +1,62 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# Muse Spark Open marathon in attach mode: V2 KV cache + Python transceiver, 1P1D. +# +# At TP4 per instance, 1P1D needs 8 GPUs. On 4-GPU nodes the two instances +# cannot share one node (two weight-resident TP4 engines exceed device memory), +# so the cluster is launched externally, one node per instance, and the +# marathon attaches to it through stress_config.external_cluster instead of +# launching workers itself. That is also why no context_servers or +# generation_servers sizing appears here. +# +# The run environment provides the endpoints through ${VAR} references, which +# the harness resolves at setup() time (see README.md in this directory): +# TRTLLM_TEST_EXTERNAL_CLUSTER_URL e.g. http://nodename:8340 +# TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR serve directory containing +# ctx/server.log, gen1/server.log and +# orch/server.log +# +# model: is the name the external cluster serves (the request envelope), not a +# checkpoint path: attach mode loads no weights. + +model: musespark_open +backend: pytorch + +# ============================================================================ +# Stress harness configuration (consumed by harness.py::StressConfig). +# Schema documented in ../README.md. +# ============================================================================ +stress_config: + mode: log_only + duration_min: 10 + + kv_cache_manager: v2 + transceiver: python + + external_cluster: + server_url: ${TRTLLM_TEST_EXTERNAL_CLUSTER_URL} + server_log: ${TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR}/orch/server.log + worker_logs: + - role: ctx + path: ${TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR}/ctx/server.log + - role: gen + path: ${TRTLLM_TEST_EXTERNAL_CLUSTER_LOG_DIR}/gen1/server.log + + log_only_probe: + interval_s: 30 + prompt: "Write one sentence about reliable distributed inference." + max_tokens: 32 + seed: 42 + request_timeout_s: 300 # first probe pays one-time JIT/warmup cost + + log_scan: + hard_zero_patterns: + - "Broken promise" + - "Segfault" + - "Segmentation fault" + - "SIGSEGV" + - "0xffffffffffffffff" + - "use-after-free" + - "heap-use-after-free" + - "AddressSanitizer:.*use-after-free" + - "double[- ]free" diff --git a/tests/integration/defs/stress_test/disagg_cancel/harness.py b/tests/integration/defs/stress_test/disagg_cancel/harness.py index 27f7be5edd77..beb48590c22c 100644 --- a/tests/integration/defs/stress_test/disagg_cancel/harness.py +++ b/tests/integration/defs/stress_test/disagg_cancel/harness.py @@ -74,6 +74,149 @@ } +# `${VAR}` references in external_cluster string fields. Expanded at +# setup() time, not YAML-parse time, so configs with unset placeholders +# still pass the parse/validate gate (and CI collection). +_ENV_REF_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") + + +def _expand_env_refs(value: str, *, label: str) -> str: + """Expand `${VAR}` references in `value` from the environment. + + Args: + value: String possibly containing `${VAR}` references. + label: Config-field name embedded in the error message. + + Returns: + `value` with every reference replaced by the variable's value. + + Raises: + ValueError: If any referenced variable is unset. Unset + references never pass through silently — a literal + `${...}` in a URL or log path would only fail later and + further from the root cause. + """ + + def replace(match: re.Match[str]) -> str: + name = match.group(1) + env_value = os.environ.get(name) + if env_value is None: + raise ValueError(f"{label} references unset environment variable {name!r}") + return env_value + + return _ENV_REF_RE.sub(replace, value) + + +def _unset_env_refs(value: str) -> list[str]: + """Return the `${VAR}` names in `value` that are unset in the environment.""" + return [name for name in _ENV_REF_RE.findall(value) if os.environ.get(name) is None] + + +@dataclass(frozen=True) +class ExternalClusterConfig: + """Parsed `stress_config.external_cluster:` block (attach mode). + + When present, the harness attaches to an already-running + disaggregated cluster instead of launching one via + `setup_disagg_cluster`: the orchestrator URL becomes the probe + endpoint and the worker log paths feed the log-scanner thread. + This is how models whose per-instance GPU footprint exceeds one + node run the marathon — the cluster is booted externally (e.g. + one tray per instance via `srun`) and the test only needs + HTTP + filesystem access to it. + + String fields may contain `${VAR}` environment references, + resolved at `setup()` time (see `_expand_env_refs`); structural + validation here is env-independent so checked-in YAMLs with + placeholders still parse. + """ + + server_url: str + # (role, log_path) pairs in YAML order; per-role indices are + # assigned in that order at setup() time. + worker_logs: tuple[tuple[str, str], ...] + server_log: Optional[str] = None + worker_health_urls: tuple[str, ...] = () + # Model name for the OpenAI request envelope; falls back to the + # YAML's top-level `model`. Attach mode never loads weights, so + # this should be the externally-launched cluster's served model + # name rather than a checkpoint path. + model_name: Optional[str] = None + + @classmethod + def from_raw(cls, raw: Any) -> "ExternalClusterConfig": + """Structurally validate the YAML subtree; no env expansion. + + Raises: + ValueError: If the subtree is not a mapping, `server_url` + is missing/empty, `worker_logs` is missing/empty, or + any entry lacks a valid `role` / `path`. + """ + if not isinstance(raw, dict): + raise ValueError(f"external_cluster must be a mapping, got {raw!r}") + server_url = raw.get("server_url") + if not isinstance(server_url, str) or not server_url: + raise ValueError("external_cluster.server_url must be a non-empty string") + raw_logs = raw.get("worker_logs") + if not isinstance(raw_logs, list) or not raw_logs: + raise ValueError( + "external_cluster.worker_logs must be a non-empty list of {role, path} mappings" + ) + worker_logs: list[tuple[str, str]] = [] + for i, entry in enumerate(raw_logs): + if not isinstance(entry, dict): + raise ValueError(f"external_cluster.worker_logs[{i}] must be a mapping") + role = entry.get("role") + path = entry.get("path") + if role not in ("ctx", "gen"): + raise ValueError( + f"external_cluster.worker_logs[{i}].role must be 'ctx' or 'gen', got {role!r}" + ) + if not isinstance(path, str) or not path: + raise ValueError( + f"external_cluster.worker_logs[{i}].path must be a non-empty string" + ) + worker_logs.append((role, path)) + server_log = raw.get("server_log") + if server_log is not None and (not isinstance(server_log, str) or not server_log): + raise ValueError("external_cluster.server_log must be a non-empty string when set") + raw_health = raw.get("worker_health_urls", []) + if not isinstance(raw_health, list) or not all( + isinstance(u, str) and u for u in raw_health + ): + raise ValueError( + "external_cluster.worker_health_urls must be a list of non-empty strings" + ) + model_name = raw.get("model_name") + if model_name is not None and (not isinstance(model_name, str) or not model_name): + raise ValueError("external_cluster.model_name must be a non-empty string when set") + return cls( + server_url=server_url, + worker_logs=tuple(worker_logs), + server_log=server_log, + worker_health_urls=tuple(raw_health), + model_name=model_name, + ) + + def missing_env_vars(self) -> list[str]: + """Names of `${VAR}` references that are currently unset. + + Lets the pytest entry point `skip` (rather than fail) an + attach-mode marathon when no external cluster is provided — + e.g. when the YAML is parametrized but the run environment + did not export the endpoint variables. + """ + missing: list[str] = [] + fields = [self.server_url, self.server_log or "", self.model_name or ""] + fields.extend(path for _, path in self.worker_logs) + fields.extend(self.worker_health_urls) + for value in fields: + for name in _unset_env_refs(value): + if name not in missing: + missing.append(name) + return missing + + @dataclass class StressConfig: """Parsed contents of a marathon YAML's ``stress_config:`` block. @@ -90,6 +233,9 @@ class StressConfig: base_concurrency: int = 64 client_cancel_rate: float = 0.10 output_length: int = 512 + # Attach mode: bind to an externally-launched cluster instead of + # calling setup_disagg_cluster. None = launch locally (default). + external_cluster: Optional[ExternalClusterConfig] = None # Full YAML subtree, for fields the harness consumes lazily # (bursts, injections, canary, log_scan, kv_cache_growth_max). raw: dict[str, Any] = field(default_factory=dict) @@ -131,6 +277,8 @@ def from_yaml_path(cls, path: Path) -> "StressConfig": for name, coerce in _STRESS_CONFIG_COERCERS.items(): if name in sc: kwargs[name] = coerce(sc[name]) + if "external_cluster" in sc: + kwargs["external_cluster"] = ExternalClusterConfig.from_raw(sc["external_cluster"]) cfg = cls(**kwargs) cfg.validate() return cfg @@ -148,6 +296,16 @@ def validate(self) -> None: raise ValueError(f"kv_cache_manager must be 'v2', got {self.kv_cache_manager!r}") if self.transceiver != "python": raise ValueError(f"transceiver must be 'python', got {self.transceiver!r}") + if self.external_cluster is not None and self.is_full_cancel_poison: + # Fault injection signals local worker subprocesses by PID + # and respawns them via the shared launcher; an externally + # launched cluster offers neither, so the combination is a + # config error rather than a silently-degraded marathon. + raise ValueError( + "mode=full_cancel_poison is unsupported with external_cluster: " + "injection and respawn require locally-launched worker processes. " + "Use mode=log_only for attach-mode marathons." + ) @property def is_log_only(self) -> bool: @@ -986,7 +1144,14 @@ def setup(self) -> None: ``stress_config`` block is stripped from the temporary YAML passed to the shared launcher so worker config validation only sees normal ``trtllm-serve`` settings. + + When `stress_config.external_cluster` is present, no cluster + is launched at all; see `_setup_external_cluster`. """ + if self.config.external_cluster is not None: + self._setup_external_cluster() + return + from test_disaggregated import ( build_worker_config, get_default_disagg_cluster_config, @@ -1056,6 +1221,103 @@ def setup(self) -> None: server_url, ) + def _setup_external_cluster(self) -> None: + """Attach to an externally-launched disagg cluster (no local Popen). + + Expands `${VAR}` references in the `external_cluster` + block, binds the orchestrator endpoint for the probe thread, + and registers worker/server log paths as `WorkerLaunchSpec` + shadows so the log-scanner thread tails them exactly as it + would for a locally-launched cluster. No workers are tracked + (`_tracked_workers` stays empty): fault injection is + unavailable in attach mode, which `StressConfig.validate` + already enforces by rejecting `full_cancel_poison`. + + Raises: + ValueError: If an `${VAR}` reference is unset, or the + YAML lacks both `external_cluster.model_name` and a + top-level `model`. + RuntimeError: If a configured `worker_health_urls` probe + does not return HTTP 200. + """ + ext = self.config.external_cluster + assert ext is not None # guarded by setup() + + server_url = _expand_env_refs(ext.server_url, label="external_cluster.server_url") + + if ext.model_name is not None: + model_name = _expand_env_refs(ext.model_name, label="external_cluster.model_name") + else: + with self.yaml_path.open("r", encoding="utf-8") as f: + doc = yaml.safe_load(f) + model_name = str((doc or {}).get("model") or "") + if not model_name: + raise ValueError( + f"YAML at {self.yaml_path} needs external_cluster.model_name " + "or a top-level model for the request envelope" + ) + + specs: list[WorkerLaunchSpec] = [] + role_counts = {"ctx": 0, "gen": 0} + for role, raw_path in ext.worker_logs: + index = role_counts[role] + role_counts[role] += 1 + specs.append( + WorkerLaunchSpec( + role=role, + index=index, + model_name=model_name, + worker_config={}, + work_dir="", + port=0, + device="", + env={}, + log_path=_expand_env_refs( + raw_path, label=f"external_cluster.worker_logs[{role}_{index}].path" + ), + ) + ) + self._worker_specs = specs + self._tracked_workers = [] + if ext.server_log is not None: + self._server_log_path = _expand_env_refs( + ext.server_log, label="external_cluster.server_log" + ) + + for i, raw_url in enumerate(ext.worker_health_urls): + url = _expand_env_refs(raw_url, label=f"external_cluster.worker_health_urls[{i}]") + self._check_external_endpoint(url) + # The orchestrator must already be healthy — attach mode never + # waits for a boot, it verifies one that external tooling + # reported up. + self._check_external_endpoint(f"{server_url.rstrip('/')}/health") + + self.bind_server_endpoint(server_url, model_name) + logger.info( + "[harness] setup() attached to external cluster %s with %d ctx / %d gen log(s)", + server_url, + role_counts["ctx"], + role_counts["gen"], + ) + + def _check_external_endpoint(self, url: str, *, timeout_s: float = 15.0) -> None: + """GET `url` once; raise if it is not reachable with HTTP 200. + + Kept as a method so unit tests can monkeypatch it without a + live HTTP server. + + Raises: + RuntimeError: On any connection error, timeout, or + non-200 status. + """ + try: + with urllib.request.urlopen(url, timeout=timeout_s) as response: + status = response.status + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise RuntimeError(f"external cluster endpoint {url} unreachable: {exc}") from exc + if status != 200: + raise RuntimeError(f"external cluster endpoint {url} returned HTTP {status}") + def _load_sanitized_cluster_config(self) -> dict[str, Any]: """Load YAML and remove harness-only fields before cluster launch.""" with self.yaml_path.open("r", encoding="utf-8") as f: diff --git a/tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py b/tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py index 849d9e7638e5..e232b8af3cc4 100644 --- a/tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py +++ b/tests/integration/defs/stress_test/disagg_cancel/test_disagg_cancel_stress.py @@ -43,6 +43,9 @@ # Add a new entry here to wire an additional YAML into the suite. _MARATHON_CONFIGS: list[str] = [ "marathon_python_v2_deepseek.yaml", + # Attaches to an externally launched 1P1D cluster (one TP4 instance per + # 4-GPU node); skipped unless the cluster's environment variables are set. + "marathon_python_v2_musespark_open_attach.yaml", ] # ``trtllm-llmapi-launch`` exports these so an LLM created in the pytest process @@ -87,6 +90,13 @@ def test_disagg_cancellation_marathon( monkeypatch.delenv(name, raising=False) harness = DisaggCancellationStressHarness(config_path) + if harness.config.external_cluster is not None: + missing = harness.config.external_cluster.missing_env_vars() + if missing: + pytest.skip( + "attach-mode marathon needs an externally-launched cluster; " + f"unset environment variables: {missing}" + ) try: harness.setup() harness.start() @@ -187,6 +197,143 @@ def test_stress_config_rejects_unknown_mode(tmp_path: Path) -> None: ) +def _write_attach_yaml(tmp_path: Path, *, mode: str = "log_only") -> Path: + """Write a minimal attach-mode marathon YAML for external_cluster tests.""" + return _write_mode_yaml( + tmp_path, + f"""\ + mode: {mode} + duration_min: 1 + kv_cache_manager: v2 + transceiver: python + external_cluster: + server_url: ${{TEST_ATTACH_URL}} + server_log: ${{TEST_ATTACH_LOG_DIR}}/orch/server.log + worker_logs: + - role: ctx + path: ${{TEST_ATTACH_LOG_DIR}}/ctx/server.log + - role: gen + path: ${{TEST_ATTACH_LOG_DIR}}/gen1/server.log + log_only_probe: + interval_s: 0.01 + max_tokens: 8 + request_timeout_s: 1 + log_scan: + hard_zero_patterns: + - "Broken promise" + """, + ) + + +def test_external_cluster_config_parses_without_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Attach YAMLs must parse/validate even when the ${VAR} refs are unset.""" + monkeypatch.delenv("TEST_ATTACH_URL", raising=False) + monkeypatch.delenv("TEST_ATTACH_LOG_DIR", raising=False) + + cfg = StressConfig.from_yaml_path(_write_attach_yaml(tmp_path)) + + ext = cfg.external_cluster + assert ext is not None + assert ext.server_url == "${TEST_ATTACH_URL}" + assert [role for role, _ in ext.worker_logs] == ["ctx", "gen"] + assert sorted(ext.missing_env_vars()) == ["TEST_ATTACH_LOG_DIR", "TEST_ATTACH_URL"] + + +def test_external_cluster_rejects_full_cancel_poison(tmp_path: Path) -> None: + """Injection needs local worker PIDs, so attach mode only supports log_only.""" + with pytest.raises(ValueError, match="full_cancel_poison is unsupported"): + StressConfig.from_yaml_path(_write_attach_yaml(tmp_path, mode="full_cancel_poison")) + + +def test_external_cluster_rejects_bad_worker_role(tmp_path: Path) -> None: + """Structural validation catches roles other than ctx/gen at parse time.""" + with pytest.raises(ValueError, match="role must be 'ctx' or 'gen'"): + StressConfig.from_yaml_path( + _write_mode_yaml( + tmp_path, + """\ + mode: log_only + external_cluster: + server_url: http://localhost:8000 + worker_logs: + - role: orchestrator + path: /tmp/x.log + """, + ) + ) + + +def test_setup_external_cluster_binds_endpoint_and_log_specs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """setup() in attach mode expands env refs and skips the local launcher.""" + monkeypatch.setenv("TEST_ATTACH_URL", "http://127.0.0.1:8340") + monkeypatch.setenv("TEST_ATTACH_LOG_DIR", str(tmp_path / "serve")) + + h = DisaggCancellationStressHarness(_write_attach_yaml(tmp_path)) + checked_urls: list[str] = [] + monkeypatch.setattr(h, "_check_external_endpoint", lambda url, **kw: checked_urls.append(url)) + + h.setup() + + assert h._server_url == "http://127.0.0.1:8340" + assert h._model_name == "dummy" # top-level model from _write_mode_yaml + assert h._cluster is None + assert h._tracked_workers == [] + assert checked_urls == ["http://127.0.0.1:8340/health"] + assert [(s.role, s.index) for s in h._worker_specs] == [("ctx", 0), ("gen", 0)] + assert h._worker_specs[0].log_path == str(tmp_path / "serve" / "ctx" / "server.log") + assert h._server_log_path == str(tmp_path / "serve" / "orch" / "server.log") + + +def test_setup_external_cluster_fails_on_unset_env( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An unset ${VAR} at setup() time is an error, never a literal passthrough.""" + monkeypatch.delenv("TEST_ATTACH_URL", raising=False) + monkeypatch.delenv("TEST_ATTACH_LOG_DIR", raising=False) + + h = DisaggCancellationStressHarness(_write_attach_yaml(tmp_path)) + with pytest.raises(ValueError, match="unset environment variable 'TEST_ATTACH_URL'"): + h.setup() + + +def test_external_cluster_log_scanner_trips_on_pattern( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The log scanner tails externally-provided logs and fail-fasts on hits.""" + serve = tmp_path / "serve" + for sub in ("ctx", "gen1", "orch"): + (serve / sub).mkdir(parents=True) + (serve / sub / "server.log").write_text("startup ok\n") + monkeypatch.setenv("TEST_ATTACH_URL", "http://127.0.0.1:8340") + monkeypatch.setenv("TEST_ATTACH_LOG_DIR", str(serve)) + + h = DisaggCancellationStressHarness( + _write_attach_yaml(tmp_path), log_scanner_poll_interval_s=0.02 + ) + monkeypatch.setattr(h, "_check_external_endpoint", lambda url, **kw: None) + h.setup() + + scanner = threading.Thread(target=h._log_scanner_thread_body, daemon=True) + scanner.start() + time.sleep(0.1) + with (serve / "gen1" / "server.log").open("a") as f: + f.write("terminate called: Broken promise\n") + + deadline = time.monotonic() + 2.0 + while not h.failed_event.is_set() and time.monotonic() < deadline: + time.sleep(0.02) + h.stop_event.set() + scanner.join(timeout=2.0) + + assert h.failed_event.is_set() + assert "Broken promise" in (h.failure_reason or "") + assert "gen_0" in (h.failure_reason or "") + + def test_log_only_thread_sends_probe_and_stops( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/integration/defs/stress_test/stress_test.py b/tests/integration/defs/stress_test/stress_test.py index 93058e8e2ca8..ed8d841273ed 100644 --- a/tests/integration/defs/stress_test/stress_test.py +++ b/tests/integration/defs/stress_test/stress_test.py @@ -458,6 +458,11 @@ def is_port_available(port: int, ModelConfig(model_dir="DeepSeek-R1/DeepSeek-R1-0528-FP4", tp_size=4, memory_requirement=172032), + # Configuration for Muse Spark Open with NVFP4 checkpoints (4 GPU setup, requires GB300 288GB) + # memory_requirement is in MiB (192 GB = 196608 MiB) + ModelConfig(model_dir="Muse-Spark-Open-816B-A42B-NVFP4", + tp_size=4, + memory_requirement=196608), ], ids=lambda x: f"{os.path.basename(x.model_dir)}_tp{x.tp_size}") def test_run_stress_test(config, stress_time_timeout, capacity_scheduler_policy, @@ -553,11 +558,13 @@ def stress_test(config, if run_performance: performance_config = PerformanceParams() - # For DeepSeek-V3 or DeepSeek-R1 specific parameters - if "DeepSeek-V3" in config.model_dir or "DeepSeek-R1" in config.model_dir: + # For DeepSeek-V3, DeepSeek-R1 or Muse Spark Open specific parameters + if ("DeepSeek-V3" in config.model_dir + or "DeepSeek-R1" in config.model_dir + or "Muse-Spark-Open" in config.model_dir): performance_config = PerformanceParams( test_timeout= - 36000 # 10 hours for DeepSeek-V3 or DeepSeek-R1, change this value if needed + 36000 # 10 hours for large MoE models, change this value if needed ) # For DeepSeek-V3 or DeepSeek-R1 specific server parameters @@ -584,6 +591,24 @@ def stress_test(config, 28800, # DeepSeek-V3 or DeepSeek-R1 specific wait time (8 hours) health_check_timeout=test_server_config.health_check_timeout) + # Muse Spark Open server parameters: its single-node GB300 serving recipe + # (TP4/EP4), with a 0.4 KV cache fraction for the stress load + if "Muse-Spark-Open" in config.model_dir: + os.environ["PYTORCH_ALLOC_CONF"] = "expandable_segments:True" + test_server_config = ServerConfig( + port=test_server_config.port, + host=test_server_config.host, + pp_size=test_server_config.pp_size, + ep_size=config.tp_size, # TP4/EP4 on one GB300 node + max_batch_size=256, + max_num_tokens=8192, + kv_cache_free_gpu_memory_fraction=0.4, + capacity_scheduler_policy=test_server_config. + capacity_scheduler_policy, + wait_interval=test_server_config.wait_interval, + max_wait_seconds=7200, # loading the checkpoint can take ~30 min + health_check_timeout=test_server_config.health_check_timeout) + # Create a StressTestConfig with customized time parameters if provided if run_stress: # Enable accuracy test for stress-test-with-accuracy mode @@ -678,6 +703,36 @@ def stress_test(config, "print_iter_log": True, }) + # Muse Spark Open LLM-API options for the same recipe (tp/ep/batch/token/ + # kv-fraction values are passed on the CLI instead) + if "Muse-Spark-Open" in config.model_dir: + extra_llm_options.update({ + "max_seq_len": 49152, + "trust_remote_code": True, + "enable_chunked_prefill": True, + "disable_overlap_scheduler": False, + "attn_backend": "TRTLLM", + "allreduce_strategy": "NCCL", + "guided_decoding_backend": "xgrammar", + "cuda_graph_config": { + "max_batch_size": 256, + "enable_padding": True, + }, + "moe_config": { + "backend": "TRTLLM", + }, + "kv_cache_config": { + # FP8 KV cache is required on SM103 when chunked prefill or + # block reuse is enabled (head-dimension-64 paged-context + # FMHA constraint). + "dtype": "fp8", + "enable_block_reuse": True, + "enable_partial_reuse": True, + "free_gpu_memory_fraction": 0.4, + }, + "print_iter_log": True, + }) + with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as temp_file: yaml.dump(extra_llm_options, temp_file) @@ -1198,16 +1253,21 @@ def stress_stage(model_name, request_count = int(stress_request_rate * stress_time) test_timeout = stress_config.stress_timeout - # Cap request count for large MoE models (DeepSeek-V3/R1) to prevent timeout - if "DeepSeek-V3" in model_path or "DeepSeek-R1" in model_path: - max_sustainable_rate = 3.0 # req/s - conservative estimate + # Cap request count for large MoE models to prevent timeout + if ("DeepSeek-V3" in model_path or "DeepSeek-R1" in model_path + or "Muse-Spark-Open" in model_path): + if "Muse-Spark-Open" in model_path: + # The rate cap keeps the stress stage near its nominal duration + # without risking the timeout. + max_sustainable_rate = 30.0 # req/s + else: + max_sustainable_rate = 3.0 # req/s - conservative estimate max_request_count = int(max_sustainable_rate * stress_config.stress_time) if request_count > max_request_count: print_info( f"Capping request_count from {request_count} to {max_request_count} " - f"for DeepSeek V3/R1 model (sustainable rate: {max_sustainable_rate} req/s)" - ) + f"(sustainable rate: {max_sustainable_rate} req/s)") request_count = max_request_count print_info(