feat(BA-3435): Implement Rolling Update deployment strategy

changes/9997.feature.md

@@ -0,0 +1 @@
+Implement Rolling Update deployment strategy

src/ai/backend/manager/models/deployment_policy/__init__.py

@@ -1,11 +1,13 @@
 from .row import (
     BlueGreenSpec,
     DeploymentPolicyRow,
+    DeploymentStrategySpec,
     RollingUpdateSpec,
 )
 
 __all__ = (
     "BlueGreenSpec",
     "DeploymentPolicyRow",
+    "DeploymentStrategySpec",
     "RollingUpdateSpec",
 )

src/ai/backend/manager/models/deployment_policy/row.py

@@ -26,6 +26,7 @@
 __all__ = (
     "BlueGreenSpec",
     "DeploymentPolicyRow",
+    "DeploymentStrategySpec",
     "RollingUpdateSpec",
 )
 
@@ -61,6 +62,9 @@ class BlueGreenSpec(BaseModel):
     promote_delay_seconds: int = 0
 
 
+DeploymentStrategySpec = RollingUpdateSpec | BlueGreenSpec
+
+
 def _get_endpoint_join_condition() -> sa.ColumnElement[bool]:
     from ai.backend.manager.models.endpoint import EndpointRow
 

src/ai/backend/manager/sokovan/deployment/strategy/blue_green.py

@@ -13,23 +13,20 @@
     DeploymentInfo,
     RouteInfo,
 )
-from ai.backend.manager.models.deployment_policy import BlueGreenSpec
+from ai.backend.manager.models.deployment_policy import DeploymentStrategySpec
 
 from .types import AbstractDeploymentStrategy, StrategyCycleResult
 
 
 class BlueGreenStrategy(AbstractDeploymentStrategy):
     """Blue-green deployment strategy FSM."""
 
-    def __init__(self, spec: BlueGreenSpec) -> None:
-        super().__init__(spec)
-        self._spec = spec
-
     @override
     def evaluate_cycle(
         self,
         deployment: DeploymentInfo,
         routes: Sequence[RouteInfo],
+        spec: DeploymentStrategySpec,
     ) -> StrategyCycleResult:
         """Evaluate one cycle of blue-green deployment for a single deployment."""
         raise NotImplementedError("Blue-green deployment strategy is not yet implemented")

src/ai/backend/manager/sokovan/deployment/strategy/evaluator.py

@@ -25,9 +25,16 @@
     InvalidDeploymentStrategy,
     InvalidDeploymentStrategySpec,
 )
-from ai.backend.manager.models.deployment_policy.conditions import DeploymentPolicyConditions
-from ai.backend.manager.models.routing.conditions import RouteConditions
-from ai.backend.manager.repositories.base import BatchQuerier, NoPagination
+from ai.backend.manager.repositories.base import (
+    BatchQuerier,
+    NoPagination,
+    QueryCondition,
+    combine_conditions_or,
+)
+from ai.backend.manager.repositories.deployment.options import (
+    DeploymentPolicyConditions,
+    RouteConditions,
+)
 from ai.backend.manager.repositories.deployment.repository import DeploymentRepository
 from ai.backend.manager.sokovan.deployment.recorder import DeploymentRecorderContext
 
@@ -79,15 +86,33 @@ async def evaluate(
             )
         )
         policy_map = {policy.endpoint: policy for policy in policy_search.items}
-        # Fetch all non-terminated routes so the strategy can detect rollback
-        # conditions (e.g. FAILED_TO_START routes after a coordinator crash).
+        # Fetch non-terminated routes + terminated routes belonging to a
+        # deploying revision.  The FSM needs terminated new-revision routes
+        # to count accumulated failures for rollback detection, but old
+        # terminated routes are irrelevant and would bloat the result set.
+        deploying_revision_ids = {
+            deployment.deploying_revision_id
+            for deployment in deployments
+            if deployment.deploying_revision_id is not None
+        }
+        route_conditions: list[QueryCondition] = [
+            RouteConditions.by_endpoint_ids(endpoint_ids),
+        ]
+        if deploying_revision_ids:
+            route_conditions.append(
+                combine_conditions_or([
+                    RouteConditions.exclude_statuses([RouteStatus.TERMINATED]),
+                    RouteConditions.by_revision_ids(deploying_revision_ids),
+                ])
+            )
+        else:
+            route_conditions.append(
+                RouteConditions.exclude_statuses([RouteStatus.TERMINATED]),
+            )
         route_search = await self._deployment_repo.search_routes(
             BatchQuerier(
                 pagination=NoPagination(),
-                conditions=[
-                    RouteConditions.by_endpoint_ids(endpoint_ids),
-                    RouteConditions.exclude_statuses([RouteStatus.TERMINATED]),
-                ],
+                conditions=route_conditions,
             )
         )
         route_map: defaultdict[UUID, list[RouteInfo]] = defaultdict(list)
@@ -108,7 +133,7 @@ async def evaluate(
 
             try:
                 strategy = self._create_strategy(policy.strategy, policy)
-                cycle_result = strategy.evaluate_cycle(deployment, routes)
+                cycle_result = strategy.evaluate_cycle(deployment, routes, policy.strategy_spec)
             except BackendAIError as e:
                 log.warning("deployment {}: evaluation error — {}", deployment.id, e)
                 result.errors.append(EvaluationErrorData(deployment=deployment, reason=str(e)))
@@ -170,4 +195,4 @@ def _create_strategy(
                     f" got {type(spec).__name__}"
                 ),
             )
-        return entry.strategy_cls(spec)
+        return entry.strategy_cls()