From 424fd05add3c986d730e83715ec1ef6ee9121954 Mon Sep 17 00:00:00 2001 From: Dennis Ramdass Date: Mon, 26 Oct 2026 11:50:53 -0700 Subject: [PATCH] Replace engine copies that aren't serving when a rollout stalls A Standalone engine's Deployment kept Kubernetes' default rolling update: surge 25% rounded up, and 25% unavailable rounded down. For an engine of fewer than four copies that's one surge pod and no copy allowed down. The controller doesn't remove an old pod until the new one is available. It cleans up unhealthy old pods only inside that budget, which is zero while the surge pod is Pending. On a pool with no free devices the surge pod can't schedule until an old pod releases its claim. In both incidents in #496 the old pod was restarting and not Ready while it held the GPUs the new pod needed, so a fix to the engine's config never rolled out and nothing served. This sets the Deployment's strategy from its observed status. Once the Deployment controller marks a rollout Progressing False with reason ProgressDeadlineExceeded (at least 600 seconds after its last progress by default), maxUnavailable becomes the number of copies that aren't available, when that's more than the default allows. The controller then removes the old pods that aren't serving, and their devices go to the new pods. Surge stays at 25%. Waiting for the deadline means a copy that's briefly unready while a rollout still progresses isn't removed. The controller marks the rollout Progressing True again once it moves, and the strategy goes back to the defaults. Until a rollout stalls, the strategy is the one the API server defaults to. Existing Deployments keep their spec on upgrade. A strategy change doesn't start a new ReplicaSet. A rollout of a healthy engine on a full pool still stalls with the old pod serving, as before. Every copy is available, so none counts as down even past the deadline. Backend.build takes the observed resources as a keyword argument. The llm-d and Grove backends ignore them, because their workloads replace a gang before starting its successor by default. Fixes #496. Signed-off-by: Dennis Ramdass Co-Authored-By: Claude Opus 5.5 --- .../function/backends/base.py | 52 ++++ .../function/backends/grove.py | 5 + .../function/backends/llmd.py | 5 + .../function/backends/native.py | 11 +- .../compose-model-replica/function/fn.py | 2 +- .../tests/test_backends.py | 259 +++++++++++++++++- .../compose-model-replica/tests/test_fn.py | 149 ++++++++-- 7 files changed, 460 insertions(+), 23 deletions(-) diff --git a/functions/compose-model-replica/function/backends/base.py b/functions/compose-model-replica/function/backends/base.py index f962652b5..02c8edcce 100644 --- a/functions/compose-model-replica/function/backends/base.py +++ b/functions/compose-model-replica/function/backends/base.py @@ -55,6 +55,8 @@ def build( provider_config: str, serving_label: str, stack: str, + *, + observed: Mapping[str, fnv1.Resource], ) -> dict[str, k8sobjv1alpha1.Object]: ... def ready_copies(self, engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource]) -> int: @@ -589,6 +591,56 @@ def ready_replicas(engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource return int(status.get("readyReplicas") or 0) +def stalled_copies_down(engine: v1alpha1.Engine, observed: Mapping[str, fnv1.Resource], key: str) -> int: + """How many of an engine's copies aren't available once its Deployment's + rollout has stalled, or 0 while it hasn't. key is the response key the + Deployment is composed under. + + A rollout has stalled once the Deployment controller marks it Progressing + False with reason ProgressDeadlineExceeded. It checks the deadline when it + next syncs the Deployment after progressDeadlineSeconds (600 by default) + with no progress, so this can take longer than the deadline. It marks it + Progressing True again as soon as the rollout makes progress, such as a pod + becoming available or an old pod being removed. + """ + status = observed_status(observed, key) or {} + stalled = any( + c.get("type") == "Progressing" and c.get("status") == "False" and c.get("reason") == "ProgressDeadlineExceeded" + for c in status.get("conditions") or [] + ) + if not stalled: + return 0 + copies = int(engine.copies or 1) + return copies - min(int(status.get("availableReplicas") or 0), copies) + + +# Kubernetes' default for both of a rolling update's bounds. Surge rounds up and +# unavailable rounds down, so an engine of fewer than four copies starts a new +# pod before it stops an old one. +_ROLLING_UPDATE_DEFAULT = "25%" + + +def rolling_update(copies: int, down: int) -> dict: + """A Deployment strategy that surges, and replaces the copies down in a + stalled rollout without waiting for their replacements. + + Waiting would deadlock when the only devices a new pod fits on are held by + an old pod that isn't serving, such as one crash-looping on a bad config. + Counting copies down only once the rollout has stalled means a copy that's + briefly unready while the rollout still progresses isn't removed. A copy + that's available stays up until its replacement is available. + + The bounds stay at Kubernetes' defaults until more copies are down than + they allow, so a healthy engine's Deployment is composed as it would be + defaulted. + """ + max_unavailable: int | str = down if down > copies // 4 else _ROLLING_UPDATE_DEFAULT + return { + "type": "RollingUpdate", + "rollingUpdate": {"maxSurge": _ROLLING_UPDATE_DEFAULT, "maxUnavailable": max_unavailable}, + } + + def serving_label(replica: v1alpha1.ModelReplica) -> str: """The serving label value a replica's serving pods share. diff --git a/functions/compose-model-replica/function/backends/grove.py b/functions/compose-model-replica/function/backends/grove.py index 72d80d75b..71d5c2415 100644 --- a/functions/compose-model-replica/function/backends/grove.py +++ b/functions/compose-model-replica/function/backends/grove.py @@ -61,6 +61,11 @@ def build( provider_config: str, serving_label: str, stack: str, + *, + # Unused here: the workload's own default rollout replaces a gang + # before it starts the new one, so it never waits on devices an old + # gang holds. + observed: Mapping[str, fnv1.Resource], # noqa: ARG002 ) -> dict[str, k8sobjv1alpha1.Object]: leader = base.engine_member(engine, base.ROLE_LEADER) worker = base.engine_member(engine, base.ROLE_WORKER) diff --git a/functions/compose-model-replica/function/backends/llmd.py b/functions/compose-model-replica/function/backends/llmd.py index d40ef0669..bcdd6d2d9 100644 --- a/functions/compose-model-replica/function/backends/llmd.py +++ b/functions/compose-model-replica/function/backends/llmd.py @@ -60,6 +60,11 @@ def build( # backend. Unused here: this backend is Standard-only, so there's no # stack to switch on. stack: str, # noqa: ARG002 + *, + # Unused here: the workload's own default rollout replaces a gang + # before it starts the new one, so it never waits on devices an old + # gang holds. + observed: Mapping[str, fnv1.Resource], # noqa: ARG002 ) -> dict[str, k8sobjv1alpha1.Object]: leader = base.engine_member(engine, base.ROLE_LEADER) worker = base.engine_member(engine, base.ROLE_WORKER) diff --git a/functions/compose-model-replica/function/backends/native.py b/functions/compose-model-replica/function/backends/native.py index 5c8a3133a..5c1df847c 100644 --- a/functions/compose-model-replica/function/backends/native.py +++ b/functions/compose-model-replica/function/backends/native.py @@ -42,6 +42,8 @@ def build( provider_config: str, serving_label: str, stack: str, + *, + observed: Mapping[str, fnv1.Resource], ) -> dict[str, k8sobjv1alpha1.Object]: member = base.engine_member(engine, base.ROLE_STANDALONE) # select_backend dispatches the native backend only for an engine with a @@ -103,13 +105,16 @@ def build( if tmpl.spec.imagePullSecrets: pod_spec["imagePullSecrets"] = [s.model_dump(exclude_none=True) for s in tmpl.spec.imagePullSecrets] + copies = int(engine.copies or 1) + key = base.workload_key(engine) deployment = { "apiVersion": "apps/v1", "kind": "Deployment", "metadata": {"name": name, "namespace": base.remote_namespace(replica)}, "spec": { - "replicas": int(engine.copies or 1), + "replicas": copies, "selector": {"matchLabels": selector}, + "strategy": base.rolling_update(copies, base.stalled_copies_down(engine, observed, key)), "template": { "metadata": base.pod_metadata(member, pod_labels, replica=replica, engine=engine), "spec": pod_spec, @@ -118,9 +123,7 @@ def build( } composed = { - base.workload_key(engine): base.wrap_object( - provider_config, deployment, cel_query=base.ALL_COPIES_READY_CEL, watch=True - ), + key: base.wrap_object(provider_config, deployment, cel_query=base.ALL_COPIES_READY_CEL, watch=True), } # Gated like the container's resources above, and for the same reason. if member.deviceRequests: diff --git a/functions/compose-model-replica/function/fn.py b/functions/compose-model-replica/function/fn.py index 6ed0814c1..d359ed6cd 100644 --- a/functions/compose-model-replica/function/fn.py +++ b/functions/compose-model-replica/function/fn.py @@ -192,7 +192,7 @@ def compose_model_serving(self) -> None: composed: dict[str, k8sobjv1alpha1.Object] = {} for engine in self.xr.spec.engines: backend = _BACKENDS[base.select_backend(engine, stack)]() - composed.update(backend.build(self.xr, engine, pc, label, stack)) + composed.update(backend.build(self.xr, engine, pc, label, stack, observed=self.req.observed.resources)) composed = routing.apply(composed, self.xr, pc) for key, obj in composed.items(): resource.update(self.rsp.desired.resources[key], obj) diff --git a/functions/compose-model-replica/tests/test_backends.py b/functions/compose-model-replica/tests/test_backends.py index 743d86430..001de67e2 100644 --- a/functions/compose-model-replica/tests/test_backends.py +++ b/functions/compose-model-replica/tests/test_backends.py @@ -144,6 +144,28 @@ class ReadyCopiesCase: want: int +@dataclasses.dataclass +class StalledCopiesDownCase: + """A test case for base.stalled_copies_down.""" + + name: str + reason: str + observed: dict[str, fnv1.Resource] + key: str + want: int + + +@dataclasses.dataclass +class RollingUpdateCase: + """A test case for base.rolling_update.""" + + name: str + reason: str + copies: int + down: int + want: dict + + @dataclasses.dataclass class Dns1035LabelCase: """A test case for base.dns_1035_label.""" @@ -436,11 +458,15 @@ def _deployment( name: str, claim_template_name: str, replicas: int, + max_unavailable: int | str, pod_metadata: dict, containers: list[dict], volumes: list[dict], ) -> dict: - """The Object composing a Standalone engine's Deployment.""" + """The Object composing a Standalone engine's Deployment. + + Its rolling update surges by 25%, and lets max_unavailable copies go down. + """ return { "spec": { "providerConfigRef": {"kind": "ClusterProviderConfig", "name": "cluster-a-pc"}, @@ -460,6 +486,10 @@ def _deployment( "spec": { "replicas": replicas, "selector": {"matchLabels": {"modelplane.ai/workload": name}}, + "strategy": { + "type": "RollingUpdate", + "rollingUpdate": {"maxSurge": "25%", "maxUnavailable": max_unavailable}, + }, "template": { "metadata": pod_metadata, "spec": { @@ -764,6 +794,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -861,6 +892,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -970,6 +1002,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -1071,6 +1104,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -1164,6 +1198,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -1252,6 +1287,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="dep-clusterA-main-3d1d5", claim_template_name="dep-clusterA-main-standalone-devices-145eb", replicas=1, @@ -1341,6 +1377,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="dep-clusterB-main-d6c52", claim_template_name="dep-clusterB-main-standalone-devices-5a8a8", replicas=1, @@ -1465,6 +1502,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -1496,6 +1534,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: ), "resource-claim-prefill-standalone": _claim_template(name="r-prefill-standalone-devices-f62af", count=1), "model-serving-decode": _deployment( + max_unavailable="25%", name="r-decode-4b27b", claim_template_name="r-decode-standalone-devices-63392", replicas=1, @@ -1577,6 +1616,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -1696,6 +1736,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -1793,6 +1834,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Standard", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -1899,6 +1941,7 @@ def _sorted(*, objects: dict[str, dict]) -> dict[str, dict]: stack="Dynamo", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -5215,7 +5258,9 @@ def test_build(case: BuildCase) -> None: # build composes one engine, so build each of the replica's engines in turn. got: dict[str, k8sobjv1alpha1.Object] = {} for engine in case.replica.spec.engines: - got.update(case.backend.build(case.replica, engine, case.provider_config, case.serving_label, case.stack)) + got.update( + case.backend.build(case.replica, engine, case.provider_config, case.serving_label, case.stack, observed={}) + ) assert _sorted(objects=_to_dicts(objects=got)) == _sorted(objects=case.want), case.reason @@ -5650,6 +5695,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -5681,6 +5727,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), "resource-claim-prefill-standalone": _claim_template(name="r-prefill-standalone-devices-f62af", count=1), "model-serving-decode": _deployment( + max_unavailable="25%", name="r-decode-4b27b", claim_template_name="r-decode-standalone-devices-63392", replicas=1, @@ -5819,6 +5866,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -5853,6 +5901,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), "resource-claim-prefill-standalone": _claim_template(name="r-prefill-standalone-devices-f62af", count=1), "model-serving-decode": _deployment( + max_unavailable="25%", name="r-decode-4b27b", claim_template_name="r-decode-standalone-devices-63392", replicas=1, @@ -5914,6 +5963,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -5945,6 +5995,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), "resource-claim-prefill-standalone": _claim_template(name="r-prefill-standalone-devices-f62af", count=1), "model-serving-decode": _deployment( + max_unavailable="25%", name="r-decode-4b27b", claim_template_name="r-decode-standalone-devices-63392", replicas=1, @@ -6090,6 +6141,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -6124,6 +6176,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), "resource-claim-prefill-standalone": _claim_template(name="r-prefill-standalone-devices-f62af", count=1), "model-serving-decode": _deployment( + max_unavailable="25%", name="r-decode-4b27b", claim_template_name="r-decode-standalone-devices-63392", replicas=1, @@ -6248,6 +6301,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-alpha": _deployment( + max_unavailable="25%", name="r-alpha-b36ce", claim_template_name="r-alpha-standalone-devices-9ec1c", replicas=1, @@ -6279,6 +6333,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), "resource-claim-alpha-standalone": _claim_template(name="r-alpha-standalone-devices-9ec1c", count=1), "model-serving-beta": _deployment( + max_unavailable="25%", name="r-beta-52d85", claim_template_name="r-beta-standalone-devices-9d8b1", replicas=1, @@ -6417,6 +6472,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-alpha": _deployment( + max_unavailable="25%", name="r-alpha-b36ce", claim_template_name="r-alpha-standalone-devices-9ec1c", replicas=1, @@ -6451,6 +6507,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), "resource-claim-alpha-standalone": _claim_template(name="r-alpha-standalone-devices-9ec1c", count=1), "model-serving-beta": _deployment( + max_unavailable="25%", name="r-beta-52d85", claim_template_name="r-beta-standalone-devices-9d8b1", replicas=1, @@ -6514,6 +6571,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -6805,6 +6863,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-prefill": _deployment( + max_unavailable="25%", name="r-prefill-d90b0", claim_template_name="r-prefill-standalone-devices-f62af", replicas=1, @@ -6998,6 +7057,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7075,6 +7135,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7152,6 +7213,7 @@ def test_cache_env(case: CacheEnvCase) -> None: reason="An engine block size above 64 sets the default picker configuration's blockSizeTokens.", composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7229,6 +7291,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7306,6 +7369,7 @@ def test_cache_env(case: CacheEnvCase) -> None: reason="An engine context length above 131072 leaves the default picker configuration's maxPrefixTokensToMatch at 131072.", composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7383,6 +7447,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7463,6 +7528,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7571,6 +7637,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7674,6 +7741,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7767,6 +7835,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7844,6 +7913,7 @@ def test_cache_env(case: CacheEnvCase) -> None: reason="A Unified replica labeled with its deployment and index gets an endpoint picker labeled with both.", composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -7923,6 +7993,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=1, @@ -8009,6 +8080,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=2, @@ -8086,6 +8158,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="r-main-bb4e3", claim_template_name="r-main-standalone-devices-f456f", replicas=2, @@ -8422,6 +8495,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="qwen2.5-7b-abcde-main-e910a", claim_template_name="qwen2.5-7b-abcde-main-standalone-devices-771a9", replicas=1, @@ -8501,6 +8575,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="qwen2.5-7b-abcde-main-e910a", claim_template_name="qwen2.5-7b-abcde-main-standalone-devices-771a9", replicas=1, @@ -8591,6 +8666,7 @@ def test_cache_env(case: CacheEnvCase) -> None: ), composed={ "model-serving-main": _deployment( + max_unavailable="25%", name="a" * 57 + "-88380", claim_template_name="a" * 57 + "-05ced", replicas=1, @@ -8668,6 +8744,7 @@ def test_cache_env(case: CacheEnvCase) -> None: provider_config="cluster-a-pc", want={ "model-serving-main": _deployment( + max_unavailable="25%", name="a" * 57 + "-88380", claim_template_name="a" * 57 + "-05ced", replicas=1, @@ -9437,3 +9514,181 @@ def _main_engine() -> v1alpha1.Engine: def test_ready_copies(case: ReadyCopiesCase) -> None: """A backend counts an engine's ready copies from its observed workload.""" assert case.backend.ready_copies(_main_engine(), case.observed) == case.want, case.reason + + +def _progressing(*, status: str, reason: str) -> dict: + """A Deployment's Progressing condition, as its controller sets it.""" + return {"type": "Progressing", "status": status, "reason": reason} + + +STALLED_COPIES_DOWN_CASES = [ + StalledCopiesDownCase( + name="NotObserved", + reason="Until its Deployment is observed, no copies count as down in a stalled rollout.", + observed={}, + key="model-serving-main", + want=0, + ), + StalledCopiesDownCase( + name="SteadyState", + reason="A Deployment whose last rollout completed has no stalled rollout.", + observed={ + "model-serving-main": _observed( + status={ + "availableReplicas": 2, + "conditions": [_progressing(status="True", reason="NewReplicaSetAvailable")], + }, + ready=True, + ) + }, + key="model-serving-main", + want=0, + ), + StalledCopiesDownCase( + name="CrashLoopBeforeDeadline", + reason="Copies down in a rollout that hasn't passed its progress deadline don't count yet.", + observed={ + "model-serving-main": _observed( + status={"conditions": [_progressing(status="True", reason="ReplicaSetUpdated")]}, + ready=False, + ) + }, + key="model-serving-main", + want=0, + ), + StalledCopiesDownCase( + name="CrashLoopPastDeadline", + reason="Once a rollout passes its progress deadline, every copy that isn't available counts as down.", + observed={ + "model-serving-main": _observed( + status={"conditions": [_progressing(status="False", reason="ProgressDeadlineExceeded")]}, + ready=False, + ) + }, + key="model-serving-main", + want=2, + ), + StalledCopiesDownCase( + name="OneDownPastDeadline", + reason="Past the deadline, only the copy that isn't available counts as down.", + observed={ + "model-serving-main": _observed( + status={ + "availableReplicas": 1, + "conditions": [_progressing(status="False", reason="ProgressDeadlineExceeded")], + }, + ready=False, + ) + }, + key="model-serving-main", + want=1, + ), + StalledCopiesDownCase( + name="HealthyStallPastDeadline", + reason="A rollout past its deadline with every copy available has no copy down.", + observed={ + "model-serving-main": _observed( + status={ + "availableReplicas": 2, + "conditions": [_progressing(status="False", reason="ProgressDeadlineExceeded")], + }, + ready=True, + ) + }, + key="model-serving-main", + want=0, + ), + StalledCopiesDownCase( + name="SurgeOvershoot", + reason="A surge pod available beside the old ones leaves no copy down.", + observed={ + "model-serving-main": _observed( + status={ + "availableReplicas": 3, + "conditions": [_progressing(status="False", reason="ProgressDeadlineExceeded")], + }, + ready=True, + ) + }, + key="model-serving-main", + want=0, + ), + StalledCopiesDownCase( + name="OtherKey", + reason="The Deployment is read under the key it's composed under, whatever else is observed.", + observed={ + "model-serving-main": _observed( + status={ + "availableReplicas": 2, + "conditions": [_progressing(status="True", reason="NewReplicaSetAvailable")], + }, + ready=True, + ), + "workload-native-main": _observed( + status={ + "availableReplicas": 1, + "conditions": [_progressing(status="False", reason="ProgressDeadlineExceeded")], + }, + ready=False, + ), + }, + key="workload-native-main", + want=1, + ), +] + + +@pytest.mark.parametrize("case", STALLED_COPIES_DOWN_CASES, ids=lambda case: case.name) +def test_stalled_copies_down(case: StalledCopiesDownCase) -> None: + """An engine's copies down in a stalled rollout come from its observed Deployment.""" + assert base.stalled_copies_down(_main_engine(), case.observed, case.key) == case.want, case.reason + + +def _rolling_update(*, max_unavailable: int | str) -> dict: + """A Deployment strategy that surges by 25% and lets max_unavailable copies go down.""" + return {"type": "RollingUpdate", "rollingUpdate": {"maxSurge": "25%", "maxUnavailable": max_unavailable}} + + +ROLLING_UPDATE_CASES = [ + RollingUpdateCase( + name="NoneDown", + reason="With no copy down in a stalled rollout, the strategy is Kubernetes' default.", + copies=1, + down=0, + want=_rolling_update(max_unavailable="25%"), + ), + RollingUpdateCase( + name="OneCopyDown", + reason="A copy down in a stalled rollout is replaced without waiting for its replacement.", + copies=1, + down=1, + want=_rolling_update(max_unavailable=1), + ), + RollingUpdateCase( + name="ThreeCopiesOneDown", + reason="Only the copy that's down may be replaced first.", + copies=3, + down=1, + want=_rolling_update(max_unavailable=1), + ), + RollingUpdateCase( + name="EightCopiesOneDown", + reason="One copy down is within the 25% of eight copies a rollout may take down anyway.", + copies=8, + down=1, + want=_rolling_update(max_unavailable="25%"), + ), + RollingUpdateCase( + name="EightCopiesThreeDown", + reason="Three copies down is more than the 25% of eight, so all three may be replaced first.", + copies=8, + down=3, + want=_rolling_update(max_unavailable=3), + ), +] + + +@pytest.mark.parametrize("case", ROLLING_UPDATE_CASES, ids=lambda case: case.name) +def test_rolling_update(case: RollingUpdateCase) -> None: + """A rollout surges, and replaces the copies down in a stalled rollout first.""" + assert base.rolling_update(case.copies, case.down) == case.want, case.reason diff --git a/functions/compose-model-replica/tests/test_fn.py b/functions/compose-model-replica/tests/test_fn.py index c2fd025c7..640736da9 100644 --- a/functions/compose-model-replica/tests/test_fn.py +++ b/functions/compose-model-replica/tests/test_fn.py @@ -210,11 +210,20 @@ def _cluster(*, provider_config_ref: str | None) -> fnv1.Resource: return fnv1.Resource(resource=resource.dict_to_struct(cluster)) -def _deployment(*, ready: fnv1.Ready, name: str, claim_template_name: str, labels: dict, copies: int) -> fnv1.Resource: +def _deployment( + *, + ready: fnv1.Ready, + max_unavailable: int | str, + name: str, + claim_template_name: str, + labels: dict, + copies: int, +) -> fnv1.Resource: """The Object composing an engine's Deployment. Its pod carries the labels the collector attributes the engine's metrics by: - the deployment, the engine and the member's role. + the deployment, the engine and the member's role. Its rolling update surges + by 25%, and lets max_unavailable copies go down. """ return fnv1.Resource( resource=resource.dict_to_struct( @@ -239,6 +248,10 @@ def _deployment(*, ready: fnv1.Ready, name: str, claim_template_name: str, label "spec": { "replicas": copies, "selector": {"matchLabels": {"modelplane.ai/workload": name}}, + "strategy": { + "type": "RollingUpdate", + "rollingUpdate": {"maxSurge": "25%", "maxUnavailable": max_unavailable}, + }, "template": { "metadata": {"labels": labels}, "spec": { @@ -695,15 +708,23 @@ def _epp_service(*, ready: fnv1.Ready) -> fnv1.Resource: ) -def _observed_deployment(*, ready: bool, ready_replicas: int) -> fnv1.Resource: +def _observed_deployment(*, ready: bool, ready_replicas: int, deadline_exceeded: bool) -> fnv1.Resource: """The engine's Deployment Object as observed back. ready sets the Object's derived Ready condition, and ready_replicas the - Deployment's status.readyReplicas, which a Deployment leaves out at 0. + Deployment's status.readyReplicas and status.availableReplicas, which a + Deployment leaves out at 0. With no minReadySeconds a ready pod is + available. deadline_exceeded sets the Progressing condition a rollout gets + once it passes its progress deadline. """ observed_manifest: dict = {"kind": "Deployment"} + status: dict = {} if ready_replicas: - observed_manifest["status"] = {"readyReplicas": ready_replicas} + status = {"readyReplicas": ready_replicas, "availableReplicas": ready_replicas} + if deadline_exceeded: + status["conditions"] = [{"type": "Progressing", "status": "False", "reason": "ProgressDeadlineExceeded"}] + if status: + observed_manifest["status"] = status return fnv1.Resource( resource=resource.dict_to_struct( { @@ -771,8 +792,10 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_prefill_decode_replica(serving=v1alpha1.Serving(mode="PrefillDecode")), resources={ - "model-serving-prefill": _observed_deployment(ready=True, ready_replicas=1), - "model-serving-decode": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-prefill": _observed_deployment( + ready=True, ready_replicas=1, deadline_exceeded=False + ), + "model-serving-decode": _observed_deployment(ready=True, ready_replicas=1, deadline_exceeded=False), }, ), required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")])}, @@ -818,8 +841,12 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_prefill_decode_replica(serving=v1alpha1.Serving(mode="PrefillDecode")), resources={ - "model-serving-prefill": _observed_deployment(ready=True, ready_replicas=1), - "model-serving-decode": _observed_deployment(ready=False, ready_replicas=0), + "model-serving-prefill": _observed_deployment( + ready=True, ready_replicas=1, deadline_exceeded=False + ), + "model-serving-decode": _observed_deployment( + ready=False, ready_replicas=0, deadline_exceeded=False + ), }, ), required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")])}, @@ -948,6 +975,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-prefill": _deployment( + max_unavailable="25%", copies=1, ready=fnv1.READY_UNSPECIFIED, name="test-replica-prefill-10d7e", @@ -966,6 +994,7 @@ def _to_dict(msg: message.Message) -> dict: ready=fnv1.READY_UNSPECIFIED, name="test-replica-prefill-standalone-devices-51762" ), "model-serving-decode": _deployment( + max_unavailable="25%", copies=1, ready=fnv1.READY_UNSPECIFIED, name="test-replica-decode-44185", @@ -1075,6 +1104,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", copies=1, ready=fnv1.READY_UNSPECIFIED, name="test-replica-main-6b608", @@ -1215,7 +1245,7 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=1), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, deadline_exceeded=False), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), }, @@ -1227,6 +1257,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", copies=1, ready=fnv1.READY_TRUE, name="test-replica-main-6b608", @@ -1290,7 +1321,7 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=1), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, deadline_exceeded=False), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1309,6 +1340,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", copies=1, ready=fnv1.READY_TRUE, name="test-replica-main-6b608", @@ -1372,7 +1404,7 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=1), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=1, deadline_exceeded=False), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1391,6 +1423,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", copies=1, ready=fnv1.READY_TRUE, name="test-replica-main-6b608", @@ -1451,7 +1484,7 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=2), resources={ - "model-serving-main": _observed_deployment(ready=False, ready_replicas=1), + "model-serving-main": _observed_deployment(ready=False, ready_replicas=1, deadline_exceeded=False), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1470,6 +1503,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", ready=fnv1.READY_UNSPECIFIED, copies=2, name="test-replica-main-6b608", @@ -1527,7 +1561,87 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=2), resources={ - "model-serving-main": _observed_deployment(ready=False, ready_replicas=0), + "model-serving-main": _observed_deployment(ready=False, ready_replicas=0, deadline_exceeded=False), + "model-route": _observed_object(ready=True), + "resource-claim-main-standalone": _observed_object(ready=True), + "inference-pool": _observed_object(ready=True), + "epp": _observed_object(ready=True), + "epp-config": _observed_object(ready=True), + "epp-role": _observed_object(ready=True), + "epp-rolebinding": _observed_object(ready=True), + "epp-serviceaccount": _observed_object(ready=True), + "epp-service": _observed_object(ready=True), + }, + ), + required_resources={"cluster": fnv1.Resources(items=[_cluster(provider_config_ref="cluster-a-pc")])}, + ), + want=fnv1.RunFunctionResponse( + meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)), + desired=fnv1.State( + resources={ + "model-serving-main": _deployment( + max_unavailable="25%", + ready=fnv1.READY_UNSPECIFIED, + copies=2, + name="test-replica-main-6b608", + claim_template_name="test-replica-main-standalone-devices-e609d", + labels={ + "modelplane.ai/deployment": "my-deployment", + "modelplane.ai/engine": "main", + "modelplane.ai/role": "Standalone", + "modelplane.ai/serving": "test-replica", + "modelplane.ai/workload": "test-replica-main-6b608", + }, + ), + "model-route": _route(ready=fnv1.READY_TRUE), + "resource-claim-main-standalone": _claim_template( + ready=fnv1.READY_TRUE, name="test-replica-main-standalone-devices-e609d" + ), + "inference-pool": _inference_pool(ready=fnv1.READY_TRUE), + "epp": _epp( + ready=fnv1.READY_TRUE, + config_checksum="c4f1cefa4a1fe67069eaa6e9dde3e5b9bc7e79d3f807ecd076dd021259b2ab39", + ), + "epp-config": _epp_config(ready=fnv1.READY_TRUE), + "epp-role": _epp_role(ready=fnv1.READY_TRUE), + "epp-rolebinding": _epp_role_binding(ready=fnv1.READY_TRUE), + "epp-serviceaccount": _epp_service_account(ready=fnv1.READY_TRUE), + "epp-service": _epp_service(ready=fnv1.READY_TRUE), + }, + ), + context=structpb.Struct(), + requirements=fnv1.Requirements( + resources={ + "cluster": fnv1.ResourceSelector( + api_version="modelplane.ai/v1alpha1", + kind="InferenceCluster", + match_name="cluster-a", + ), + }, + ), + conditions=[ + fnv1.Condition(type="ModelAccepted", status=fnv1.STATUS_CONDITION_TRUE, reason="Accepted"), + fnv1.Condition(type="ModelServing", status=fnv1.STATUS_CONDITION_FALSE, reason="ModelStarting"), + fnv1.Condition( + type="ModelReady", + status=fnv1.STATUS_CONDITION_FALSE, + reason="ModelStarting", + message="0 of 2 engine copies ready", + ), + ], + ), + ), + Case( + name="RolloutPastDeadline", + reason=( + "A rollout past its progress deadline with no copy available replaces both copies without waiting for " + "their replacements, which may need their devices." + ), + req=fnv1.RunFunctionRequest( + observed=fnv1.State( + composite=_model_replica(copies=2), + resources={ + "model-serving-main": _observed_deployment(ready=False, ready_replicas=0, deadline_exceeded=True), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1546,6 +1660,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable=2, ready=fnv1.READY_UNSPECIFIED, copies=2, name="test-replica-main-6b608", @@ -1606,7 +1721,7 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=2), resources={ - "model-serving-main": _observed_deployment(ready=False, ready_replicas=2), + "model-serving-main": _observed_deployment(ready=False, ready_replicas=2, deadline_exceeded=False), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1625,6 +1740,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", ready=fnv1.READY_UNSPECIFIED, copies=2, name="test-replica-main-6b608", @@ -1682,7 +1798,7 @@ def _to_dict(msg: message.Message) -> dict: observed=fnv1.State( composite=_model_replica(copies=2), resources={ - "model-serving-main": _observed_deployment(ready=True, ready_replicas=2), + "model-serving-main": _observed_deployment(ready=True, ready_replicas=2, deadline_exceeded=False), "model-route": _observed_object(ready=True), "resource-claim-main-standalone": _observed_object(ready=True), "inference-pool": _observed_object(ready=True), @@ -1701,6 +1817,7 @@ def _to_dict(msg: message.Message) -> dict: desired=fnv1.State( resources={ "model-serving-main": _deployment( + max_unavailable="25%", ready=fnv1.READY_TRUE, copies=2, name="test-replica-main-6b608",