Introduce effective_spec for runs and fleets (#2579)

* Introduce effective_spec for runs and fleets

* Fix spec copy

Author: r4victor

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

@@ -246,32 +246,34 @@ def _print_plan_header(plan: FleetPlan):
     def th(s: str) -> str:
         return f"[bold]{s}[/bold]"
 
+    spec = plan.get_effective_spec()
+
     configuration_table = Table(box=None, show_header=False)
     configuration_table.add_column(no_wrap=True)  # key
     configuration_table.add_column()  # value
 
     configuration_table.add_row(th("Project"), plan.project_name)
     configuration_table.add_row(th("User"), plan.user)
-    configuration_table.add_row(th("Configuration"), plan.spec.configuration_path or "?")
-    configuration_table.add_row(th("Type"), plan.spec.configuration.type)
+    configuration_table.add_row(th("Configuration"), spec.configuration_path or "?")
+    configuration_table.add_row(th("Type"), spec.configuration.type)
 
     fleet_type = "cloud"
-    nodes = plan.spec.configuration.nodes or "-"
-    placement = plan.spec.configuration.placement or InstanceGroupPlacement.ANY
-    reservation = plan.spec.configuration.reservation
+    nodes = spec.configuration.nodes or "-"
+    placement = spec.configuration.placement or InstanceGroupPlacement.ANY
+    reservation = spec.configuration.reservation
     backends = None
-    if plan.spec.configuration.backends is not None:
-        backends = ", ".join(b.value for b in plan.spec.configuration.backends)
+    if spec.configuration.backends is not None:
+        backends = ", ".join(b.value for b in spec.configuration.backends)
     regions = None
-    if plan.spec.configuration.regions is not None:
-        regions = ", ".join(plan.spec.configuration.regions)
+    if spec.configuration.regions is not None:
+        regions = ", ".join(spec.configuration.regions)
     resources = None
-    if plan.spec.configuration.resources is not None:
-        resources = plan.spec.configuration.resources.pretty_format()
-    spot_policy = plan.spec.merged_profile.spot_policy
-    if plan.spec.configuration.ssh_config is not None:
+    if spec.configuration.resources is not None:
+        resources = spec.configuration.resources.pretty_format()
+    spot_policy = spec.merged_profile.spot_policy
+    if spec.configuration.ssh_config is not None:
         fleet_type = "ssh"
-        nodes = len(plan.spec.configuration.ssh_config.hosts)
+        nodes = len(spec.configuration.ssh_config.hosts)
         resources = None
         spot_policy = None
 

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

@@ -27,6 +27,7 @@
 def print_run_plan(
     run_plan: RunPlan, max_offers: Optional[int] = None, include_run_properties: bool = True
 ):
+    run_spec = run_plan.get_effective_run_spec()
     job_plan = run_plan.job_plans[0]
 
     props = Table(box=None, show_header=False)
@@ -43,18 +44,18 @@ def print_run_plan(
     )
     if include_run_properties:
         inactivity_duration = None
-        if isinstance(run_plan.run_spec.configuration, DevEnvironmentConfiguration):
+        if isinstance(run_spec.configuration, DevEnvironmentConfiguration):
             inactivity_duration = "-"
-            if isinstance(run_plan.run_spec.configuration.inactivity_duration, int):
+            if isinstance(run_spec.configuration.inactivity_duration, int):
                 inactivity_duration = format_pretty_duration(
-                    run_plan.run_spec.configuration.inactivity_duration
+                    run_spec.configuration.inactivity_duration
                 )
         if job_plan.job_spec.retry is None:
             retry = "-"
         else:
             retry = escape(job_plan.job_spec.retry.pretty_format())
 
-        profile = run_plan.run_spec.merged_profile
+        profile = run_spec.merged_profile
         creation_policy = profile.creation_policy
         # FIXME: This assumes the default idle_duration is the same for client and server.
         # If the server changes idle_duration, old clients will see incorrect value.
@@ -79,8 +80,8 @@ def th(s: str) -> str:
     props.add_row(th("Project"), run_plan.project_name)
     props.add_row(th("User"), run_plan.user)
     if include_run_properties:
-        props.add_row(th("Configuration"), run_plan.run_spec.configuration_path)
-        props.add_row(th("Type"), run_plan.run_spec.configuration.type)
+        props.add_row(th("Configuration"), run_spec.configuration_path)
+        props.add_row(th("Type"), run_spec.configuration.type)
     props.add_row(th("Resources"), pretty_req)
     props.add_row(th("Spot policy"), spot_policy)
     props.add_row(th("Max price"), max_price)
@@ -91,7 +92,7 @@ def th(s: str) -> str:
         props.add_row(th("Max duration"), max_duration)
         if inactivity_duration is not None:  # None means n/a
             props.add_row(th("Inactivity duration"), inactivity_duration)
-    props.add_row(th("Reservation"), run_plan.run_spec.configuration.reservation or "-")
+    props.add_row(th("Reservation"), run_spec.configuration.reservation or "-")
 
     offers = Table(box=None)
     offers.add_column("#")

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

@@ -312,11 +312,17 @@ class FleetPlan(CoreModel):
     project_name: str
     user: str
     spec: FleetSpec
+    effective_spec: Optional[FleetSpec] = None
     current_resource: Optional[Fleet] = None
     offers: List[InstanceOfferWithAvailability]
     total_offers: int
     max_offer_price: Optional[float] = None
 
+    def get_effective_spec(self) -> FleetSpec:
+        if self.effective_spec is not None:
+            return self.effective_spec
+        return self.spec
+
 
 class ApplyFleetPlanInput(CoreModel):
     spec: FleetSpec

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

@@ -455,10 +455,16 @@ class RunPlan(CoreModel):
     project_name: str
     user: str
     run_spec: RunSpec
+    effective_run_spec: Optional[RunSpec] = None
     job_plans: List[JobPlan]
     current_resource: Optional[Run] = None
     action: ApplyAction
 
+    def get_effective_run_spec(self) -> RunSpec:
+        if self.effective_run_spec is not None:
+            return self.effective_run_spec
+        return self.run_spec
+
 
 class ApplyRunPlanInput(CoreModel):
     run_spec: RunSpec

src/dstack/_internal/server/services/fleets.py

@@ -234,32 +234,34 @@ async def get_plan(
     user: UserModel,
     spec: FleetSpec,
 ) -> FleetPlan:
+    effective_spec = FleetSpec.parse_obj(spec.dict())
     current_fleet: Optional[Fleet] = None
     current_fleet_id: Optional[uuid.UUID] = None
-    if spec.configuration.name is not None:
+    if effective_spec.configuration.name is not None:
         current_fleet_model = await get_project_fleet_model_by_name(
-            session=session, project=project, name=spec.configuration.name
+            session=session, project=project, name=effective_spec.configuration.name
         )
         if current_fleet_model is not None:
             current_fleet = fleet_model_to_fleet(current_fleet_model)
             current_fleet_id = current_fleet_model.id
-    await _check_ssh_hosts_not_yet_added(session, spec, current_fleet_id)
+    await _check_ssh_hosts_not_yet_added(session, effective_spec, current_fleet_id)
 
     offers = []
-    if spec.configuration.ssh_config is None:
+    if effective_spec.configuration.ssh_config is None:
         offers_with_backends = await get_create_instance_offers(
             project=project,
-            profile=spec.merged_profile,
-            requirements=_get_fleet_requirements(spec),
-            fleet_spec=spec,
-            blocks=spec.configuration.blocks,
+            profile=effective_spec.merged_profile,
+            requirements=_get_fleet_requirements(effective_spec),
+            fleet_spec=effective_spec,
+            blocks=effective_spec.configuration.blocks,
         )
         offers = [offer for _, offer in offers_with_backends]
-    _remove_fleet_spec_sensitive_info(spec)
+    _remove_fleet_spec_sensitive_info(effective_spec)
     plan = FleetPlan(
         project_name=project.name,
         user=user.name,
         spec=spec,
+        effective_spec=effective_spec,
         current_resource=current_fleet,
         offers=offers[:50],
         total_offers=len(offers),
@@ -335,7 +337,7 @@ async def create_fleet(
 
     lock_namespace = f"fleet_names_{project.name}"
     if get_db().dialect_name == "sqlite":
-        # Start new transaction to see commited changes after lock
+        # Start new transaction to see committed changes after lock
         await session.commit()
     elif get_db().dialect_name == "postgresql":
         await session.execute(

src/dstack/_internal/server/services/runs.py

@@ -279,45 +279,44 @@ async def get_plan(
     run_spec: RunSpec,
     max_offers: Optional[int],
 ) -> RunPlan:
-    _validate_run_spec_and_set_defaults(run_spec)
+    effective_run_spec = RunSpec.parse_obj(run_spec.dict())
+    _validate_run_spec_and_set_defaults(effective_run_spec)
 
-    profile = run_spec.merged_profile
+    profile = effective_run_spec.merged_profile
     creation_policy = profile.creation_policy
 
     current_resource = None
     action = ApplyAction.CREATE
-    if run_spec.run_name is not None:
+    if effective_run_spec.run_name is not None:
         current_resource = await get_run_by_name(
             session=session,
             project=project,
-            run_name=run_spec.run_name,
+            run_name=effective_run_spec.run_name,
         )
         if (
             current_resource is not None
             and not current_resource.status.is_finished()
-            and _can_update_run_spec(current_resource.run_spec, run_spec)
+            and _can_update_run_spec(current_resource.run_spec, effective_run_spec)
         ):
             action = ApplyAction.UPDATE
 
-    # TODO(egor-s): do we need to generate all replicas here?
-    jobs = await get_jobs_from_run_spec(run_spec, replica_num=0)
+    jobs = await get_jobs_from_run_spec(effective_run_spec, replica_num=0)
 
     volumes = await get_job_configured_volumes(
         session=session,
         project=project,
-        run_spec=run_spec,
+        run_spec=effective_run_spec,
         job_num=0,
     )
 
     pool_offers = await _get_pool_offers(
         session=session,
         project=project,
-        run_spec=run_spec,
+        run_spec=effective_run_spec,
         job=jobs[0],
         volumes=volumes,
     )
-    run_name = run_spec.run_name  # preserve run_name
-    run_spec.run_name = "dry-run"  # will regenerate jobs on submission
+    effective_run_spec.run_name = "dry-run"  # will regenerate jobs on submission
 
     # Get offers once for all jobs
     offers = []
@@ -330,7 +329,7 @@ async def get_plan(
             multinode=jobs[0].job_spec.jobs_per_replica > 1,
             volumes=volumes,
             privileged=jobs[0].job_spec.privileged,
-            instance_mounts=check_run_spec_requires_instance_mounts(run_spec),
+            instance_mounts=check_run_spec_requires_instance_mounts(effective_run_spec),
         )
 
     job_plans = []
@@ -351,11 +350,12 @@ async def get_plan(
         )
         job_plans.append(job_plan)
 
-    run_spec.run_name = run_name  # restore run_name
+    effective_run_spec.run_name = run_spec.run_name  # restore run_name
     run_plan = RunPlan(
         project_name=project.name,
         user=user.name,
         run_spec=run_spec,
+        effective_run_spec=effective_run_spec,
         job_plans=job_plans,
         current_resource=current_resource,
         action=action,

src/tests/_internal/server/routers/test_fleets.py

@@ -851,6 +851,7 @@ async def test_returns_plan(self, test_db, session: AsyncSession, client: AsyncC
             "project_name": project.name,
             "user": user.name,
             "spec": spec.dict(),
+            "effective_spec": spec.dict(),
             "current_resource": None,
             "offers": [json.loads(o.json()) for o in offers],
             "total_offers": len(offers),