Support targeting specific run instances

Co-authored-by: Federico D'Agostino <fede.dagos@gmail.com>

Author: Andrey Cheptsov

mkdocs/docs/reference/dstack.yml/dev-environment.md

@@ -34,6 +34,40 @@ The `dev-environment` configuration type allows running [dev environments](../..
       type:
         required: true
 
+### `instances[n]` { #_instances data-toc-label="instances" }
+
+When `instances` is set, the run is placed only on matching existing fleet instances.
+
+=== "By name"
+
+    #SCHEMA# dstack._internal.core.models.profiles.InstanceNameSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+=== "By hostname"
+
+    #SCHEMA# dstack._internal.core.models.profiles.InstanceHostnameSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+=== "By fleet and instance number"
+
+    #SCHEMA# dstack._internal.core.models.profiles.FleetInstanceSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+??? info "Short syntax"
+
+    The short syntax for instances is an instance name string.
+
+    * `my-fleet-1`, same as `{name: my-fleet-1}`
+
 ### `resources`
 
 #SCHEMA# dstack._internal.core.models.resources.ResourcesSpec

mkdocs/docs/reference/dstack.yml/service.md

@@ -114,6 +114,40 @@ The `service` configuration type allows running [services](../../concepts/servic
       type:
         required: true
 
+### `instances[n]` { #_instances data-toc-label="instances" }
+
+When `instances` is set, the run is placed only on matching existing fleet instances.
+
+=== "By name"
+
+    #SCHEMA# dstack._internal.core.models.profiles.InstanceNameSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+=== "By hostname"
+
+    #SCHEMA# dstack._internal.core.models.profiles.InstanceHostnameSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+=== "By fleet and instance number"
+
+    #SCHEMA# dstack._internal.core.models.profiles.FleetInstanceSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+??? info "Short syntax"
+
+    The short syntax for instances is an instance name string.
+
+    * `my-fleet-1`, same as `{name: my-fleet-1}`
+
 ### `resources`
 
 #SCHEMA# dstack._internal.core.models.resources.ResourcesSpec

mkdocs/docs/reference/dstack.yml/task.md

@@ -34,6 +34,40 @@ The `task` configuration type allows running [tasks](../../concepts/tasks.md).
       type:
         required: true
 
+### `instances[n]` { #_instances data-toc-label="instances" }
+
+When `instances` is set, the run is placed only on matching existing fleet instances.
+
+=== "By name"
+
+    #SCHEMA# dstack._internal.core.models.profiles.InstanceNameSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+=== "By hostname"
+
+    #SCHEMA# dstack._internal.core.models.profiles.InstanceHostnameSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+=== "By fleet and instance number"
+
+    #SCHEMA# dstack._internal.core.models.profiles.FleetInstanceSelector
+        overrides:
+          show_root_heading: false
+          type:
+            required: true
+
+??? info "Short syntax"
+
+    The short syntax for instances is an instance name string.
+
+    * `my-fleet-1`, same as `{name: my-fleet-1}`
+
 ### `resources`
 
 #SCHEMA# dstack._internal.core.models.resources.ResourcesSpec

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

@@ -10,6 +10,8 @@ def get_profile_excludes(profile: Optional[ProfileParams]) -> IncludeExcludeSetT
         return excludes
     if profile.backend_options is None:
         excludes.add("backend_options")
+    if profile.instances is None:
+        excludes.add("instances")
     return excludes
 
 

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

@@ -83,6 +83,8 @@ def get_run_spec_excludes(run_spec: RunSpec) -> IncludeExcludeDictType:
     spec_excludes: IncludeExcludeDictType = {}
     configuration_excludes: IncludeExcludeDictType = {}
     profile_excludes = get_profile_excludes(run_spec.profile)
+    for field in get_profile_excludes(run_spec.configuration):
+        configuration_excludes[field] = True
 
     if run_spec.configuration.backend_options is None:
         configuration_excludes["backend_options"] = True

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

@@ -234,6 +234,58 @@ def crons(self) -> List[str]:
         return self.cron
 
 
+class InstanceNameSelector(CoreModel):
+    name: Annotated[str, Field(description="The fleet instance name", min_length=1)]
+
+
+class InstanceHostnameSelector(CoreModel):
+    hostname: Annotated[
+        str, Field(description="The fleet instance hostname or IP address", min_length=1)
+    ]
+
+
+def _parse_fleet_instance_selector_fleet(v: Any) -> Any:
+    if isinstance(v, str):
+        return EntityReference.parse(v)
+    return v
+
+
+class FleetInstanceSelectorConfig(CoreConfig):
+    @staticmethod
+    def schema_extra(schema: Dict[str, Any]):
+        add_extra_schema_types(
+            schema["properties"]["fleet"],
+            extra_types=[{"type": "string", "minLength": 1}],
+        )
+
+
+class FleetInstanceSelector(generate_dual_core_model(FleetInstanceSelectorConfig)):
+    fleet: Annotated[
+        EntityReference,
+        Field(
+            description=(
+                "The fleet reference. For fleets owned by the current project, specify"
+                " the fleet name. For a fleet from another project, specify"
+                " `<project name>/<fleet name>` or an object with `project` and `name`."
+            ),
+        ),
+    ]
+    instance: Annotated[int, Field(description="The fleet instance number", ge=0)]
+
+    _validate_fleet = validator("fleet", pre=True, allow_reuse=True)(
+        _parse_fleet_instance_selector_fleet
+    )
+
+
+InstanceSelector = Union[InstanceNameSelector, InstanceHostnameSelector, FleetInstanceSelector]
+
+
+def parse_instance_selector(v: Union[InstanceSelector, str]) -> InstanceSelector:
+    if isinstance(v, str):
+        return InstanceNameSelector(name=v)
+    return v
+
+
 class ProfileParamsConfig(CoreConfig):
     @staticmethod
     def schema_extra(schema: Dict[str, Any]):
@@ -249,6 +301,10 @@ def schema_extra(schema: Dict[str, Any]):
             schema["properties"]["idle_duration"],
             extra_types=[{"type": "string"}],
         )
+        add_extra_schema_types(
+            schema["properties"]["instances"]["items"],
+            extra_types=[{"type": "string", "minLength": 1}],
+        )
 
 
 class ProfileParams(CoreModel):
@@ -391,6 +447,18 @@ class ProfileParams(CoreModel):
             ),
         ),
     ] = None
+    instances: Annotated[
+        Optional[List[InstanceSelector]],
+        Field(
+            description=(
+                "The specific fleet instances to consider for reuse."
+                " Each value can be an instance name string, or an object with"
+                " `name`, `hostname`, or `fleet` and `instance`."
+                " When set, the run is only placed on matching existing instances."
+            ),
+            min_items=1,
+        ),
+    ] = None
     tags: Annotated[
         Optional[Dict[str, str]],
         Field(
@@ -416,11 +484,23 @@ class ProfileParams(CoreModel):
         parse_idle_duration
     )
     _validate_fleets = validator("fleets", allow_reuse=True, each_item=True)(EntityReference.parse)
+    _validate_instances = validator("instances", pre=True, allow_reuse=True, each_item=True)(
+        parse_instance_selector
+    )
     _validate_tags = validator("tags", pre=True, allow_reuse=True)(tags_validator)
     _validate_backend_options = validator("backend_options", allow_reuse=True)(
         validate_backend_options
     )
 
+    def dict(self, *args, **kwargs) -> Dict:
+        # super() does not work with pydantic-duality
+        res = CoreModel.dict(self, *args, **kwargs)
+        # Older servers reject unknown fields in profile/request models. Since `instances`
+        # did not exist before, do not serialize it when unset.
+        if self.instances is None:
+            res.pop("instances", None)
+        return res
+
 
 class ProfileProps(CoreModel):
     name: Annotated[