Support elastic fleets (#2967)

* Do not auto-delete empty fleets that allow 0 nodes

* Refactor offers filtering

* Respect fleet nodes.max

* Fix replicas typing

* Support provisioning in empty fleets

* Rebase migrations

* Prioritize fleet choice by capacity and offer price

* Unassign run from fleet for retrying

* Comment on optimal fleet choice

* Refactor fleet/instance selects to fix for update with outer join

* If no fleets have available offers, create a new fleet

* Revert respect nodes.max

* Set nodes for autocreated fleet

* Forbid new fleets creation with fleets specified

* Use Range for replicas

* Ensure models Config is inherited

* Rebase migrations

* Use fleets with no instances if fleets specified

* Fix tests

* Fix tests

Author: r4victor

src/dstack/_internal/core/backends/aws/compute.py

@@ -383,10 +383,7 @@ def is_suitable_placement_group(
     ) -> bool:
         if not _offer_supports_placement_group(instance_offer, placement_group):
             return False
-        return (
-            placement_group.configuration.backend == BackendType.AWS
-            and placement_group.configuration.region == instance_offer.region
-        )
+        return placement_group.configuration.region == instance_offer.region
 
     def create_gateway(
         self,

src/dstack/_internal/core/backends/base/compute.py

@@ -263,10 +263,6 @@ def is_suitable_placement_group(
         Checks if the instance offer can be provisioned in the placement group.
 
         Should return immediately, without performing API calls.
-
-        Can be called with an offer originating from a different backend, because some backends
-        (BackendType.DSTACK) produce offers on behalf of other backends. Should return `False`
-        in that case.
         """
         pass
 

src/dstack/_internal/core/backends/gcp/compute.py

@@ -448,10 +448,7 @@ def is_suitable_placement_group(
         placement_group: PlacementGroup,
         instance_offer: InstanceOffer,
     ) -> bool:
-        return (
-            placement_group.configuration.backend == BackendType.GCP
-            and placement_group.configuration.region == instance_offer.region
-        )
+        return placement_group.configuration.region == instance_offer.region
 
     def create_gateway(
         self,

src/dstack/_internal/core/backends/nebius/compute.py

@@ -298,10 +298,7 @@ def is_suitable_placement_group(
         placement_group: PlacementGroup,
         instance_offer: InstanceOffer,
     ) -> bool:
-        if not (
-            placement_group.configuration.backend == BackendType.NEBIUS
-            and placement_group.configuration.region == instance_offer.region
-        ):
+        if placement_group.configuration.region != instance_offer.region:
             return False
         assert placement_group.provisioning_data is not None
         backend_data = NebiusPlacementGroupBackendData.load(

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

@@ -102,7 +102,7 @@ class RegistryAuth(CoreModel):
         password (str): The password or access token
     """
 
-    class Config:
+    class Config(CoreModel.Config):
         frozen = True
 
     username: Annotated[str, Field(description="The username")]

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

@@ -20,6 +20,7 @@
 from dstack._internal.core.models.unix import UnixUser
 from dstack._internal.core.models.volumes import MountPoint, VolumeConfiguration, parse_mount_point
 from dstack._internal.utils.common import has_duplicates
+from dstack._internal.utils.json_schema import add_extra_schema_types
 from dstack._internal.utils.json_utils import (
     pydantic_orjson_dumps_with_indent,
 )
@@ -561,7 +562,7 @@ class ServiceConfigurationParams(CoreModel):
     )
     auth: Annotated[bool, Field(description="Enable the authorization")] = True
     replicas: Annotated[
-        Union[conint(ge=1), constr(regex=r"^[0-9]+..[1-9][0-9]*$"), Range[int]],
+        Range[int],
         Field(
             description="The number of replicas. Can be a number (e.g. `2`) or a range (`0..4` or `1..8`). "
             "If it's a range, the `scaling` property is required"
@@ -592,20 +593,13 @@ def convert_model(cls, v: Optional[Union[AnyModel, str]]) -> Optional[AnyModel]:
         return v
 
     @validator("replicas")
-    def convert_replicas(cls, v: Any) -> Range[int]:
-        if isinstance(v, str) and ".." in v:
-            min, max = v.replace(" ", "").split("..")
-            v = Range(min=min or 0, max=max or None)
-        elif isinstance(v, (int, float)):
-            v = Range(min=v, max=v)
+    def convert_replicas(cls, v: Range[int]) -> Range[int]:
         if v.max is None:
             raise ValueError("The maximum number of replicas is required")
+        if v.min is None:
+            v.min = 0
         if v.min < 0:
             raise ValueError("The minimum number of replicas must be greater than or equal to 0")
-        if v.max < v.min:
-            raise ValueError(
-                "The maximum number of replicas must be greater than or equal to the minimum number of replicas"
-            )
         return v
 
     @validator("gateway")
@@ -622,9 +616,9 @@ def validate_gateway(
     def validate_scaling(cls, values):
         scaling = values.get("scaling")
         replicas = values.get("replicas")
-        if replicas.min != replicas.max and not scaling:
+        if replicas and replicas.min != replicas.max and not scaling:
             raise ValueError("When you set `replicas` to a range, ensure to specify `scaling`.")
-        if replicas.min == replicas.max and scaling:
+        if replicas and replicas.min == replicas.max and scaling:
             raise ValueError("To use `scaling`, `replicas` must be set to a range.")
         return values
 
@@ -655,6 +649,14 @@ class ServiceConfiguration(
 ):
     type: Literal["service"] = "service"
 
+    class Config(CoreModel.Config):
+        @staticmethod
+        def schema_extra(schema: Dict[str, Any]):
+            add_extra_schema_types(
+                schema["properties"]["replicas"],
+                extra_types=[{"type": "integer"}, {"type": "string"}],
+            )
+
 
 AnyRunConfiguration = Union[DevEnvironmentConfiguration, TaskConfiguration, ServiceConfiguration]
 
@@ -715,7 +717,7 @@ class DstackConfiguration(CoreModel):
         Field(discriminator="type"),
     ]
 
-    class Config:
+    class Config(CoreModel.Config):
         json_loads = orjson.loads
         json_dumps = pydantic_orjson_dumps_with_indent
 

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

@@ -234,7 +234,7 @@ class InstanceGroupParams(CoreModel):
     termination_policy: Annotated[Optional[TerminationPolicy], Field(exclude=True)] = None
     termination_idle_time: Annotated[Optional[Union[str, int]], Field(exclude=True)] = None
 
-    class Config:
+    class Config(CoreModel.Config):
         @staticmethod
         def schema_extra(schema: Dict[str, Any], model: Type):
             del schema["properties"]["termination_policy"]
@@ -279,7 +279,7 @@ class FleetSpec(CoreModel):
     # TODO: make merged_profile a computed field after migrating to pydanticV2
     merged_profile: Annotated[Profile, Field(exclude=True)] = None
 
-    class Config:
+    class Config(CoreModel.Config):
         @staticmethod
         def schema_extra(schema: Dict[str, Any], model: Type) -> None:
             prop = schema.get("properties", {})

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

@@ -122,7 +122,7 @@ class SSHConnectionParams(CoreModel):
     username: str
     port: int
 
-    class Config:
+    class Config(CoreModel.Config):
         frozen = True
 
 

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

@@ -339,7 +339,7 @@ class ProfileParams(CoreModel):
     termination_policy: Annotated[Optional[TerminationPolicy], Field(exclude=True)] = None
     termination_idle_time: Annotated[Optional[Union[str, int]], Field(exclude=True)] = None
 
-    class Config:
+    class Config(CoreModel.Config):
         @staticmethod
         def schema_extra(schema: Dict[str, Any]) -> None:
             del schema["properties"]["pool_name"]
@@ -379,7 +379,7 @@ class Profile(ProfileProps, ProfileParams):
 class ProfilesConfig(CoreModel):
     profiles: List[Profile]
 
-    class Config:
+    class Config(CoreModel.Config):
         json_loads = orjson.loads
         json_dumps = pydantic_orjson_dumps_with_indent
 

src/dstack/_internal/core/models/repos/remote.py

@@ -32,7 +32,7 @@ class RemoteRepoCreds(CoreModel):
     # TODO: remove in 0.20. Left for compatibility with CLI <=0.18.44
     protocol: Annotated[Optional[str], Field(exclude=True)] = None
 
-    class Config:
+    class Config(CoreModel.Config):
         @staticmethod
         def schema_extra(schema: Dict[str, Any]) -> None:
             del schema["properties"]["protocol"]
@@ -47,7 +47,7 @@ class RemoteRepoInfo(BaseRepoInfo):
     repo_port: Annotated[Optional[int], Field(exclude=True)] = None
     repo_user_name: Annotated[Optional[str], Field(exclude=True)] = None
 
-    class Config:
+    class Config(BaseRepoInfo.Config):
         @staticmethod
         def schema_extra(schema: Dict[str, Any]) -> None:
             del schema["properties"]["repo_host_name"]