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",