Rolling deployment for services (dstackai#2821)

Support rolling deployment when some service
configuration properties are changed. During the
deployment, `dstack` updates service replicas one
by one. It first starts a new version of the
replica, then waits until it is running, then
stops the old version of the replica.

As part of the implementation, introduce a new run
and job property - `deployment_num`. When a new
configuration is applied, the `deployment_num` of
the run is incremented. Then, `dstack` gradually
updates the jobs so that their `deployment_num`
matches that of the run. Some jobs are updated
in-place, if the new configuration does not affect
their spec, others are redeployed as described
above.

```shell
> dstack apply

Active run test-service already exists. Detected configuration changes that can be updated in-place: ['image', 'env', 'commands']
Update the run? [y/n]: y
⠋ Launching test-service...
 NAME                            BACKEND          RESOURCES                        PRICE    STATUS       SUBMITTED
 test-service deployment=1                                                                  running      11 mins ago
   replica=0 job=0 deployment=0  aws (us-west-2)  cpu=2 mem=1GB disk=100GB (spot)  $0.0026  terminating  11 mins ago
   replica=1 job=0 deployment=1  aws (us-west-2)  cpu=2 mem=1GB disk=100GB (spot)  $0.0026  running      1 min ago
```

Author: jvstme

src/dstack/_internal/cli/commands/attach.py

@@ -52,9 +52,8 @@ def _register(self):
         )
         self._parser.add_argument(
             "--replica",
-            help="The replica number. Defaults to 0.",
+            help="The replica number. Defaults to any running replica.",
             type=int,
-            default=0,
         )
         self._parser.add_argument(
             "--job",
@@ -129,14 +128,15 @@ def _print_finished_message_when_available(run: Run) -> None:
 def _print_attached_message(
     run: Run,
     bind_address: Optional[str],
-    replica_num: int,
+    replica_num: Optional[int],
     job_num: int,
 ):
     if bind_address is None:
         bind_address = "localhost"
 
-    output = f"Attached to run [code]{run.name}[/] (replica={replica_num} job={job_num})\n"
     job = get_or_error(run._find_job(replica_num=replica_num, job_num=job_num))
+    replica_num = job.job_spec.replica_num
+    output = f"Attached to run [code]{run.name}[/] (replica={replica_num} job={job_num})\n"
     name = run.name
     if replica_num != 0 or job_num != 0:
         name = job.job_spec.job_name

src/dstack/_internal/cli/services/configurators/run.py

@@ -599,6 +599,7 @@ def _is_ready_to_attach(run: Run) -> bool:
         ]
         or run._run.jobs[0].job_submissions[-1].status
         in [JobStatus.SUBMITTED, JobStatus.PROVISIONING, JobStatus.PULLING]
+        or run._run.is_deployment_in_progress()
     )
 
 

src/dstack/_internal/cli/utils/run.py

@@ -162,9 +162,16 @@ def get_runs_table(
 
     for run in runs:
         run = run._run  # TODO(egor-s): make public attribute
+        show_deployment_num = (
+            verbose
+            and run.run_spec.configuration.type == "service"
+            or run.is_deployment_in_progress()
+        )
+        merge_job_rows = len(run.jobs) == 1 and not show_deployment_num
 
         run_row: Dict[Union[str, int], Any] = {
-            "NAME": run.run_spec.run_name,
+            "NAME": run.run_spec.run_name
+            + (f" [secondary]deployment={run.deployment_num}[/]" if show_deployment_num else ""),
             "SUBMITTED": format_date(run.submitted_at),
             "STATUS": (
                 run.latest_job_submission.status_message
@@ -174,7 +181,7 @@ def get_runs_table(
         }
         if run.error:
             run_row["ERROR"] = run.error
-        if len(run.jobs) != 1:
+        if not merge_job_rows:
             add_row_from_dict(table, run_row)
 
         for job in run.jobs:
@@ -184,7 +191,12 @@ def get_runs_table(
                 inactive_for = format_duration_multiunit(latest_job_submission.inactivity_secs)
                 status += f" (inactive for {inactive_for})"
             job_row: Dict[Union[str, int], Any] = {
-                "NAME": f"  replica={job.job_spec.replica_num} job={job.job_spec.job_num}",
+                "NAME": f"  replica={job.job_spec.replica_num} job={job.job_spec.job_num}"
+                + (
+                    f" deployment={latest_job_submission.deployment_num}"
+                    if show_deployment_num
+                    else ""
+                ),
                 "STATUS": latest_job_submission.status_message,
                 "SUBMITTED": format_date(latest_job_submission.submitted_at),
                 "ERROR": latest_job_submission.error,
@@ -208,7 +220,7 @@ def get_runs_table(
                         "PRICE": f"${jpd.price:.4f}".rstrip("0").rstrip("."),
                     }
                 )
-            if len(run.jobs) == 1:
+            if merge_job_rows:
                 # merge rows
                 job_row.update(run_row)
             add_row_from_dict(table, job_row, style="secondary" if len(run.jobs) != 1 else None)

src/dstack/_internal/core/compatibility/runs.py

@@ -19,6 +19,8 @@ def get_apply_plan_excludes(plan: ApplyRunPlanInput) -> Optional[Dict]:
     if current_resource is not None:
         current_resource_excludes = {}
         current_resource_excludes["status_message"] = True
+        if current_resource.deployment_num == 0:
+            current_resource_excludes["deployment_num"] = True
         apply_plan_excludes["current_resource"] = current_resource_excludes
         current_resource_excludes["run_spec"] = get_run_spec_excludes(current_resource.run_spec)
         job_submissions_excludes = {}
@@ -36,6 +38,8 @@ def get_apply_plan_excludes(plan: ApplyRunPlanInput) -> Optional[Dict]:
             }
         if all(js.exit_status is None for js in job_submissions):
             job_submissions_excludes["exit_status"] = True
+        if all(js.deployment_num == 0 for js in job_submissions):
+            job_submissions_excludes["deployment_num"] = True
         latest_job_submission = current_resource.latest_job_submission
         if latest_job_submission is not None:
             latest_job_submission_excludes = {}
@@ -50,6 +54,8 @@ def get_apply_plan_excludes(plan: ApplyRunPlanInput) -> Optional[Dict]:
                 }
             if latest_job_submission.exit_status is None:
                 latest_job_submission_excludes["exit_status"] = True
+            if latest_job_submission.deployment_num == 0:
+                latest_job_submission_excludes["deployment_num"] = True
     return {"plan": apply_plan_excludes}
 
 

src/dstack/_internal/core/models/runs.py

@@ -289,6 +289,7 @@ class ClusterInfo(CoreModel):
 class JobSubmission(CoreModel):
     id: UUID4
     submission_num: int
+    deployment_num: int = 0  # default for compatibility with pre-0.19.14 servers
     submitted_at: datetime
     last_processed_at: datetime
     finished_at: Optional[datetime]
@@ -516,6 +517,7 @@ class Run(CoreModel):
     latest_job_submission: Optional[JobSubmission]
     cost: float = 0
     service: Optional[ServiceSpec] = None
+    deployment_num: int = 0  # default for compatibility with pre-0.19.14 servers
     # TODO: make error a computed field after migrating to pydanticV2
     error: Optional[str] = None
     deleted: Optional[bool] = None
@@ -578,6 +580,13 @@ def _get_status_message(
             return "retrying"
         return status.value
 
+    def is_deployment_in_progress(self) -> bool:
+        return any(
+            not j.job_submissions[-1].status.is_finished()
+            and j.job_submissions[-1].deployment_num != self.deployment_num
+            for j in self.jobs
+        )
+
 
 class JobPlan(CoreModel):
     job_spec: JobSpec