add generic instances based on the provider

Author: sanderegg

services/autoscaling/src/simcore_service_autoscaling/modules/cluster_scaling/_auto_scaling_core.py

@@ -112,6 +112,9 @@ async def _analyze_current_cluster(
         state_names=["stopped"],
     )
 
+    for instance in itertools.chain(existing_ec2_instances, warm_buffer_ec2_instances):
+        auto_scaling_mode.add_instance_generic_resources(app, instance)
+
     attached_ec2s, pending_ec2s = associate_ec2_instances_with_nodes(
         docker_nodes, existing_ec2_instances
     )

services/autoscaling/src/simcore_service_autoscaling/modules/cluster_scaling/_provider_computational.py

@@ -182,3 +182,12 @@ async def is_instance_retired(
     async def try_retire_nodes(self, app: FastAPI) -> None:
         assert self  # nosec
         await dask.try_retire_nodes(_scheduler_url(app), _scheduler_auth(app))
+
+    def add_instance_generic_resources(
+        self, app: FastAPI, instance: EC2InstanceData
+    ) -> None:
+        assert self  # nosec
+        assert app  # nosec
+        app_settings = get_application_settings(app)
+        assert app_settings.AUTOSCALING_DASK  # nosec
+        dask.add_instance_generic_resources(app_settings.AUTOSCALING_DASK, instance)

services/autoscaling/src/simcore_service_autoscaling/modules/cluster_scaling/_provider_protocol.py

@@ -47,3 +47,7 @@ async def is_instance_retired(
     ) -> bool: ...
 
     async def try_retire_nodes(self, app: FastAPI) -> None: ...
+
+    def add_instance_generic_resources(
+        self, app: FastAPI, instance: EC2InstanceData
+    ) -> None: ...

services/autoscaling/src/simcore_service_autoscaling/modules/dask.py

@@ -12,6 +12,7 @@
 from dask_task_models_library.resource_constraints import (
     DASK_WORKER_THREAD_RESOURCE_NAME,
     DaskTaskResources,
+    create_ec2_resource_constraint_key,
 )
 from distributed.core import Status
 from models_library.clusters import ClusterAuthentication, TLSAuthentication
@@ -339,6 +340,9 @@ async def try_retire_nodes(
         )
 
 
+_LARGE_RESOURCE: Final[int] = 99999
+
+
 def add_instance_generic_resources(
     settings: DaskMonitoringSettings, instance: EC2InstanceData
 ) -> None:
@@ -351,3 +355,7 @@ def add_instance_generic_resources(
     instance.resources.generic_resources[DASK_WORKER_THREAD_RESOURCE_NAME] = (
         instance_threads
     )
+
+    instance.resources.generic_resources[
+        create_ec2_resource_constraint_key(instance.type)
+    ] = _LARGE_RESOURCE

services/autoscaling/tests/unit/test_modules_cluster_scaling_computational.py

@@ -638,7 +638,7 @@ async def test_cluster_scaling_up_and_down(  # noqa: PLR0915
     )
     mock_docker_tag_node.reset_mock()
     mock_docker_set_node_availability.assert_not_called()
-    mock_rabbitmq_post_message.assert_called_once()
+    assert mock_rabbitmq_post_message.call_count == 3
     mock_rabbitmq_post_message.reset_mock()
 
     # now we have 1 monitored node that needs to be mocked