Consider backend offers when choosing optimal fleet

Author: r4victor

src/dstack/_internal/server/background/tasks/process_submitted_jobs.py

@@ -289,7 +289,8 @@ async def _process_submitted_job(session: AsyncSession, job_model: JobModel):
                     instance_filters=instance_filters,
                 )
             fleet_models = fleet_models_with_instances + fleet_models_without_instances
-            fleet_model, fleet_instances_with_offers = _find_optimal_fleet_with_offers(
+            fleet_model, fleet_instances_with_offers = await _find_optimal_fleet_with_offers(
+                project=project,
                 fleet_models=fleet_models,
                 run_model=run_model,
                 run_spec=run.run_spec,
@@ -492,7 +493,8 @@ async def _refetch_fleet_models_with_instances(
     return fleet_models
 
 
-def _find_optimal_fleet_with_offers(
+async def _find_optimal_fleet_with_offers(
+    project: ProjectModel,
     fleet_models: list[FleetModel],
     run_model: RunModel,
     run_spec: RunSpec,
@@ -502,58 +504,99 @@ def _find_optimal_fleet_with_offers(
 ) -> tuple[Optional[FleetModel], list[tuple[InstanceModel, InstanceOfferWithAvailability]]]:
     if run_model.fleet is not None:
         # Using the fleet that was already chosen by the master job
-        fleet_instances_with_offers = _get_fleet_instances_with_offers(
+        fleet_instances_with_pool_offers = _get_fleet_instances_with_pool_offers(
             fleet_model=run_model.fleet,
             run_spec=run_spec,
             job=job,
             master_job_provisioning_data=master_job_provisioning_data,
             volumes=volumes,
         )
-        return run_model.fleet, fleet_instances_with_offers
+        return run_model.fleet, fleet_instances_with_pool_offers
 
     if len(fleet_models) == 0:
         return None, []
 
     nodes_required_num = _get_nodes_required_num_for_run(run_spec)
-    # The current strategy is to first consider fleets that can accommodate
-    # the run without additional provisioning and choose the one with the cheapest offer.
-    # Fallback to fleet with the cheapest offer among all fleets with offers.
+    # The current strategy is first to consider fleets that can accommodate
+    # the run without additional provisioning and choose the one with the cheapest pool offer.
+    # Then choose a fleet with the cheapest pool offer among all fleets with pool offers.
+    # If there are no fleets with pool offers, choose a fleet with a cheapest backend offer.
+    # Fallback to autocreated fleet if fleets have no pool or backend offers.
+    # TODO: Consider trying all backend offers and then choosing a fleet.
     candidate_fleets_with_offers: list[
         tuple[
             Optional[FleetModel],
             list[tuple[InstanceModel, InstanceOfferWithAvailability]],
             int,
-            tuple[int, float],
+            int,
+            tuple[int, float, float],
         ]
     ] = []
     for candidate_fleet_model in fleet_models:
-        fleet_instances_with_offers = _get_fleet_instances_with_offers(
+        fleet_instances_with_pool_offers = _get_fleet_instances_with_pool_offers(
             fleet_model=candidate_fleet_model,
             run_spec=run_spec,
             job=job,
             master_job_provisioning_data=master_job_provisioning_data,
             volumes=volumes,
         )
-        fleet_available_offers = [
-            o for _, o in fleet_instances_with_offers if o.availability.is_available()
-        ]
-        fleet_has_available_capacity = nodes_required_num <= len(fleet_available_offers)
-        fleet_cheapest_offer = math.inf
-        if len(fleet_available_offers) > 0:
-            fleet_cheapest_offer = fleet_available_offers[0].price
-        fleet_priority = (not fleet_has_available_capacity, fleet_cheapest_offer)
+        fleet_has_available_capacity = nodes_required_num <= len(fleet_instances_with_pool_offers)
+        fleet_cheapest_pool_offer = math.inf
+        if len(fleet_instances_with_pool_offers) > 0:
+            fleet_cheapest_pool_offer = fleet_instances_with_pool_offers[0][1].price
+
+        candidate_fleet = fleet_model_to_fleet(candidate_fleet_model)
+        profile = combine_fleet_and_run_profiles(
+            candidate_fleet.spec.merged_profile, run_spec.merged_profile
+        )
+        fleet_requirements = get_fleet_requirements(candidate_fleet.spec)
+        requirements = combine_fleet_and_run_requirements(
+            fleet_requirements, job.job_spec.requirements
+        )
+        multinode = (
+            candidate_fleet.spec.configuration.placement == InstanceGroupPlacement.CLUSTER
+            or job.job_spec.jobs_per_replica > 1
+        )
+        fleet_backend_offers = []
+        if (
+            _check_can_create_new_instance_in_fleet(candidate_fleet)
+            and profile is not None
+            and requirements is not None
+        ):
+            fleet_backend_offers = await get_offers_by_requirements(
+                project=project,
+                profile=profile,
+                requirements=requirements,
+                exclude_not_available=True,
+                multinode=multinode,
+                master_job_provisioning_data=master_job_provisioning_data,
+                volumes=volumes,
+                privileged=job.job_spec.privileged,
+                instance_mounts=check_run_spec_requires_instance_mounts(run_spec),
+            )
+
+        fleet_cheapest_backend_offer = math.inf
+        if len(fleet_backend_offers) > 0:
+            fleet_cheapest_backend_offer = fleet_backend_offers[0][1].price
+
+        fleet_priority = (
+            not fleet_has_available_capacity,
+            fleet_cheapest_pool_offer,
+            fleet_cheapest_backend_offer,
+        )
         candidate_fleets_with_offers.append(
             (
                 candidate_fleet_model,
-                fleet_instances_with_offers,
-                len(fleet_available_offers),
+                fleet_instances_with_pool_offers,
+                len(fleet_instances_with_pool_offers),
+                len(fleet_backend_offers),
                 fleet_priority,
             )
         )
     if run_spec.merged_profile.fleets is None and all(
-        t[2] == 0 for t in candidate_fleets_with_offers
+        t[2] == 0 and t[3] == 0 for t in candidate_fleets_with_offers
     ):
-        # If fleets are not specified and no fleets have available offers, create a new fleet.
+        # If fleets are not specified and no fleets have available pool or backend offers, create a new fleet.
         # This is for compatibility with non-fleet-first UX when runs created new fleets
         # if there are no instances to reuse.
         return None, []
@@ -573,7 +616,7 @@ def _get_nodes_required_num_for_run(run_spec: RunSpec) -> int:
     return nodes_required_num
 
 
-def _get_fleet_instances_with_offers(
+def _get_fleet_instances_with_pool_offers(
     fleet_model: FleetModel,
     run_spec: RunSpec,
     job: Job,