🐛Computational backend: re-enable service resources override (#4970)

Author: sanderegg

packages/models-library/src/models_library/api_schemas_resource_usage_tracker/pricing_plans.py

@@ -3,6 +3,7 @@
 from typing import Any, ClassVar
 
 from models_library.resource_tracker import (
+    HardwareInfo,
     PricingPlanClassification,
     PricingPlanId,
     PricingUnitCostId,
@@ -18,7 +19,7 @@ class PricingUnitGet(BaseModel):
     current_cost_per_unit: Decimal
     current_cost_per_unit_id: PricingUnitCostId
     default: bool
-    specific_info: dict
+    specific_info: HardwareInfo
 
     class Config:
         schema_extra: ClassVar[dict[str, Any]] = {
@@ -30,8 +31,9 @@ class Config:
                     "current_cost_per_unit": 5.7,
                     "current_cost_per_unit_id": 1,
                     "default": True,
-                    "specific_info": {},
+                    "specific_info": hw_config_example,
                 }
+                for hw_config_example in HardwareInfo.Config.schema_extra["examples"]
             ]
         }
 
@@ -55,9 +57,10 @@ class Config:
                     "classification": "TIER",
                     "created_at": "2023-01-11 13:11:47.293595",
                     "pricing_plan_key": "pricing-plan-sleeper",
-                    "pricing_units": [
-                        PricingUnitGet.Config.schema_extra["examples"][0]
-                    ],
+                    "pricing_units": [pricing_unit_get_example],
                 }
+                for pricing_unit_get_example in PricingUnitGet.Config.schema_extra[
+                    "examples"
+                ]
             ]
         }

packages/models-library/src/models_library/resource_tracker.py

@@ -57,7 +57,10 @@ class HardwareInfo(BaseModel):
 
     class Config:
         schema_extra: ClassVar[dict[str, Any]] = {
-            "examples": [{"aws_ec2_instances": ["c6a.4xlarge"]}]
+            "examples": [
+                {"aws_ec2_instances": ["c6a.4xlarge"]},
+                {"aws_ec2_instances": []},
+            ]
         }
 
 

packages/models-library/src/models_library/wallets.py

@@ -27,8 +27,9 @@ class Config:
 
 ZERO_CREDITS = Decimal(0)
 
-
-### DB
+#
+# DB
+#
 
 
 class WalletDB(BaseModel):

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -49,6 +49,7 @@
     ClusterNotFoundError,
     ClustersKeeperNotAvailableError,
     ComputationalRunNotFoundError,
+    ConfigurationError,
     PricingPlanUnitNotFoundError,
     ProjectNotFoundError,
     SchedulerError,
@@ -345,6 +346,8 @@ async def create_computation(  # noqa: PLR0913
         raise HTTPException(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=f"{e}"
         ) from e
+    except ConfigurationError as e:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
 @router.get(

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_tasks/_utils.py

@@ -45,7 +45,7 @@
 )
 from simcore_postgres_database.utils_projects_nodes import ProjectNodesRepo
 
-from .....core.errors import ClustersKeeperNotAvailableError
+from .....core.errors import ClustersKeeperNotAvailableError, ConfigurationError
 from .....models.comp_tasks import CompTaskAtDB, Image, NodeSchema
 from .....modules.resource_usage_tracker_client import ResourceUsageTrackerClient
 from .....utils.comp_scheduler import COMPLETED_STATES
@@ -189,7 +189,8 @@ async def _get_pricing_and_hardware_infos(
     node_key: ServiceKey,
     node_version: ServiceVersion,
 ) -> tuple[PricingInfo | None, HardwareInfo]:
-    if not is_wallet:
+    if not is_wallet or (to_node_class(node_key) == NodeClass.FRONTEND):
+        # NOTE: frontend services have no pricing plans, therefore no need to call RUT
         return None, HardwareInfo(aws_ec2_instances=[])
     project_nodes_repo = ProjectNodesRepo(project_uuid=project_id)
     output = await project_nodes_repo.get_project_node_pricing_unit_id(
@@ -204,7 +205,7 @@ async def _get_pricing_and_hardware_infos(
             product_name, pricing_plan_id, pricing_unit_id
         )
         pricing_unit_cost_id = pricing_unit_get.current_cost_per_unit_id
-        aws_ec2_instances = pricing_unit_get.specific_info["aws_ec2_instances"]
+        aws_ec2_instances = pricing_unit_get.specific_info.aws_ec2_instances
     else:
         (
             pricing_plan_id,
@@ -231,6 +232,7 @@ async def _get_pricing_and_hardware_infos(
 
 
 _RAM_SAFE_MARGIN: Final[ByteSize] = parse_obj_as(ByteSize, "1GiB")
+_CPUS_SAFE_MARGIN: Final[float] = 0.1
 
 
 async def _update_project_node_resources_from_hardware_info(
@@ -263,28 +265,41 @@ def _by_type_name(ec2: EC2InstanceType) -> bool:
         selected_ec2_instance_type = next(
             iter(filter(_by_type_name, unordered_list_ec2_instance_types))
         )
+
         # now update the project node required resources
         # NOTE: we keep a safe margin with the RAM as the dask-sidecar "sees"
         # less memory than the machine theoretical amount
         project_nodes_repo = ProjectNodesRepo(project_uuid=project_id)
         node = await project_nodes_repo.get(connection, node_id=node_id)
         node_resources = parse_obj_as(ServiceResourcesDict, node.required_resources)
-        assert DEFAULT_SINGLE_SERVICE_NAME in node_resources  # nosec
-        image_resources: ImageResources = node_resources[DEFAULT_SINGLE_SERVICE_NAME]
-        image_resources.resources["CPU"].set_value(
-            float(selected_ec2_instance_type.cpus)
-        )
-        image_resources.resources["RAM"].set_value(
-            selected_ec2_instance_type.ram - _RAM_SAFE_MARGIN
-        )
-
-        await project_nodes_repo.update(
-            connection,
-            node_id=node_id,
-            required_resources=ServiceResourcesDictHelpers.create_jsonable(
-                node_resources
-            ),
+        if DEFAULT_SINGLE_SERVICE_NAME in node_resources:
+            image_resources: ImageResources = node_resources[
+                DEFAULT_SINGLE_SERVICE_NAME
+            ]
+            image_resources.resources["CPU"].set_value(
+                float(selected_ec2_instance_type.cpus) - _CPUS_SAFE_MARGIN
+            )
+            image_resources.resources["RAM"].set_value(
+                selected_ec2_instance_type.ram - _RAM_SAFE_MARGIN
+            )
+
+            await project_nodes_repo.update(
+                connection,
+                node_id=node_id,
+                required_resources=ServiceResourcesDictHelpers.create_jsonable(
+                    node_resources
+                ),
+            )
+        else:
+            _logger.warning(
+                "Services resource override not implemented yet for multi-container services!!!"
+            )
+    except StopIteration as exc:
+        msg = (
+            f"invalid EC2 type name selected {set(hardware_info.aws_ec2_instances)}."
+            " TIP: adjust product configuration"
         )
+        raise ConfigurationError(msg) from exc
     except (RemoteMethodNotRegisteredError, RPCServerError) as exc:
         raise ClustersKeeperNotAvailableError from exc
 
@@ -357,14 +372,14 @@ async def generate_tasks_list_from_project(
             node_version=node.version,
         )
         assert rabbitmq_rpc_client  # nosec
-        # await _update_project_node_resources_from_hardware_info(
-        #     connection,
-        #     is_wallet=is_wallet,
-        #     project_id=project.uuid,
-        #     node_id=NodeID(node_id),
-        #     hardware_info=hardware_info,
-        #     rabbitmq_rpc_client=rabbitmq_rpc_client,
-        # )
+        await _update_project_node_resources_from_hardware_info(
+            connection,
+            is_wallet=is_wallet,
+            project_id=project.uuid,
+            node_id=NodeID(node_id),
+            hardware_info=hardware_info,
+            rabbitmq_rpc_client=rabbitmq_rpc_client,
+        )
 
         image = await _generate_task_image(
             catalog_client=catalog_client,

services/director-v2/src/simcore_service_director_v2/modules/resource_usage_tracker_client.py

@@ -107,7 +107,7 @@ async def get_default_pricing_and_hardware_info(
                     service_pricing_plan_get.pricing_plan_id,
                     unit.pricing_unit_id,
                     unit.current_cost_per_unit_id,
-                    unit.specific_info["aws_ec2_instances"],
+                    unit.specific_info.aws_ec2_instances,
                 )
         raise PricingPlanUnitNotFoundError(
             "Default pricing plan and unit does not exist"

services/director-v2/tests/unit/test_modules_dynamic_sidecar_scheduler.py

@@ -276,6 +276,7 @@ async def test_scheduler_add_remove(
     assert scheduler_data.service_name not in scheduler._scheduler._to_observe
 
 
+@pytest.mark.flaky(max_runs=3)
 async def test_scheduler_removes_partially_started_services(
     disabled_scheduler_background_task: None,
     manually_trigger_scheduler: Callable[[], Awaitable[None]],