Merge remote-tracking branch 'upstream/master' into pr-osparc-migrate-dy-scheduler-part-1

Author: Andrei Neagu

packages/aws-library/tests/test_ec2_client.py

@@ -419,6 +419,63 @@ async def test_stop_start_instances(
                 assert getattr(s, f.name) == getattr(c, f.name)
 
 
+async def test_start_instances_with_insufficient_instance_capacity(
+    simcore_ec2_api: SimcoreEC2API,
+    ec2_client: EC2Client,
+    faker: Faker,
+    ec2_instance_config: EC2InstanceConfig,
+    mocker: MockerFixture,
+):
+    # we have nothing running now in ec2
+    await _assert_no_instances_in_ec2(ec2_client)
+    # create some instance
+    _NUM_INSTANCES = 10
+    num_instances = faker.pyint(min_value=1, max_value=_NUM_INSTANCES)
+    created_instances = await simcore_ec2_api.launch_instances(
+        ec2_instance_config,
+        min_number_of_instances=num_instances,
+        number_of_instances=num_instances,
+    )
+    await _assert_instances_in_ec2(
+        ec2_client,
+        expected_num_reservations=1,
+        expected_num_instances=num_instances,
+        expected_instance_type=ec2_instance_config.type,
+        expected_tags=ec2_instance_config.tags,
+        expected_state="running",
+    )
+    # stop the instances
+    await simcore_ec2_api.stop_instances(created_instances)
+    await _assert_instances_in_ec2(
+        ec2_client,
+        expected_num_reservations=1,
+        expected_num_instances=num_instances,
+        expected_instance_type=ec2_instance_config.type,
+        expected_tags=ec2_instance_config.tags,
+        expected_state="stopped",
+    )
+
+    # Mock the EC2 client to simulate InsufficientInstanceCapacity on first subnet
+    async def mock_start_instances(*args, **kwargs) -> Any:
+        # no more machines, simulate insufficient capacity
+        error_response: dict[str, Any] = {
+            "Error": {
+                "Code": "InsufficientInstanceCapacity",
+                "Message": "An error occurred (InsufficientInstanceCapacity) when calling the StartInstances operation (reached max retries: 4): Insufficient capacity.",
+            },
+        }
+        raise botocore.exceptions.ClientError(error_response, "StartInstances")  # type: ignore
+
+    # Apply the mock
+    mocker.patch.object(
+        simcore_ec2_api.client, "start_instances", side_effect=mock_start_instances
+    )
+
+    # start the instances now
+    with pytest.raises(EC2InsufficientCapacityError):
+        await simcore_ec2_api.start_instances(created_instances)
+
+
 async def test_terminate_instance(
     simcore_ec2_api: SimcoreEC2API,
     ec2_client: EC2Client,
@@ -717,7 +774,9 @@ async def mock_run_instances(*args, **kwargs) -> Any:
     mocker.patch.object(
         simcore_ec2_api.client, "run_instances", side_effect=mock_run_instances
     )
-    with pytest.raises(EC2InsufficientCapacityError) as exc_info:
+    with pytest.raises(
+        EC2InsufficientCapacityError, match=fake_ec2_instance_type.name
+    ) as exc_info:
         await simcore_ec2_api.launch_instances(
             ec2_instance_config,
             min_number_of_instances=1,

packages/postgres-database/src/simcore_postgres_database/migration/versions/06eafd25d004_add_state_type_unknown.py

@@ -0,0 +1,53 @@
+"""add state type unknown
+
+Revision ID: 06eafd25d004
+Revises: ec4f62595e0c
+Create Date: 2025-09-01 12:25:25.617790+00:00
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "06eafd25d004"
+down_revision = "ec4f62595e0c"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute("ALTER TYPE statetype ADD VALUE 'UNKNOWN'")
+
+
+def downgrade() -> None:
+    # NOTE: PostgreSQL doesn't support removing enum values directly
+    # This downgrades only ensure that StateType.UNKNOWN is not used
+    #
+
+    # Find all tables and columns that use statetype enum
+    result = op.get_bind().execute(
+        sa.DDL(
+            """
+        SELECT t.table_name, c.column_name, c.column_default
+        FROM information_schema.columns c
+        JOIN information_schema.tables t ON c.table_name = t.table_name
+        WHERE c.udt_name = 'statetype'
+        AND t.table_schema = 'public'
+    """
+        )
+    )
+
+    tables_columns = result.fetchall()
+
+    # Update UNKNOWN states to FAILED in all affected tables
+    for table_name, column_name, _ in tables_columns:
+        op.execute(
+            sa.DDL(
+                f"""
+            UPDATE {table_name}
+            SET {column_name} = 'FAILED'
+            WHERE {column_name} = 'UNKNOWN'
+        """
+            )
+        )

packages/postgres-database/src/simcore_postgres_database/models/comp_pipeline.py

@@ -1,6 +1,5 @@
-""" Computational Pipeline Table
+"""Computational Pipeline Table"""
 
-"""
 import enum
 import uuid
 
@@ -24,6 +23,7 @@ class StateType(enum.Enum):
     ABORTED = "ABORTED"
     WAITING_FOR_RESOURCES = "WAITING_FOR_RESOURCES"
     WAITING_FOR_CLUSTER = "WAITING_FOR_CLUSTER"
+    UNKNOWN = "UNKNOWN"
 
 
 def _new_uuid():

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

@@ -15,7 +15,11 @@
     EC2Tags,
     Resources,
 )
-from aws_library.ec2._errors import EC2AccessError, EC2TooManyInstancesError
+from aws_library.ec2._errors import (
+    EC2AccessError,
+    EC2InsufficientCapacityError,
+    EC2TooManyInstancesError,
+)
 from fastapi import FastAPI
 from models_library.generated_models.docker_rest_api import Node
 from models_library.rabbitmq_messages import ProgressType
@@ -421,10 +425,46 @@ async def _activate_drained_nodes(
     )
 
 
+def _de_assign_tasks_from_warm_buffer_ec2s(
+    cluster: Cluster, instances_to_start: list[EC2InstanceData]
+) -> tuple[Cluster, list]:
+    # de-assign tasks from the warm buffer instances that could not be started
+    deassigned_tasks = list(
+        itertools.chain.from_iterable(
+            i.assigned_tasks
+            for i in cluster.warm_buffer_ec2s
+            if i.ec2_instance in instances_to_start
+        )
+    )
+    # upgrade the cluster
+    return (
+        dataclasses.replace(
+            cluster,
+            warm_buffer_ec2s=[
+                (
+                    dataclasses.replace(i, assigned_tasks=[])
+                    if i.ec2_instance in instances_to_start
+                    else i
+                )
+                for i in cluster.warm_buffer_ec2s
+            ],
+        ),
+        deassigned_tasks,
+    )
+
+
 async def _try_start_warm_buffer_instances(
     app: FastAPI, cluster: Cluster, auto_scaling_mode: AutoscalingProvider
-) -> Cluster:
-    """starts warm buffer if there are assigned tasks, or if a hot buffer of the same type is needed"""
+) -> tuple[Cluster, list]:
+    """
+    starts warm buffer if there are assigned tasks, or if a hot buffer of the same type is needed
+
+    Returns:
+        A tuple containing:
+            - The updated cluster instance after attempting to start warm buffer instances.
+            - In case warm buffer could not be started, a list of de-assigned tasks (tasks whose resource requirements cannot be fulfilled by warm buffers anymore).
+
+    """
 
     app_settings = get_application_settings(app)
     assert app_settings.AUTOSCALING_EC2_INSTANCES  # nosec
@@ -466,26 +506,34 @@ async def _try_start_warm_buffer_instances(
         ]
 
     if not instances_to_start:
-        return cluster
+        return cluster, []
 
     with log_context(
-        _logger, logging.INFO, f"start {len(instances_to_start)} warm buffer machines"
+        _logger,
+        logging.INFO,
+        f"start {len(instances_to_start)} warm buffer machines '{[i.id for i in instances_to_start]}'",
     ):
         try:
             started_instances = await get_ec2_client(app).start_instances(
                 instances_to_start
             )
-        except EC2AccessError:
+        except EC2InsufficientCapacityError:
+            # NOTE: this warning is only raised if none of the instances could be started due to InsufficientCapacity
             _logger.warning(
-                "Could not start warm buffer instances! "
-                "TIP: This can happen in case of Insufficient "
-                "Capacity on AWS AvailabilityZone(s) where the warm buffers were originally created. "
-                "Until https://github.com/ITISFoundation/osparc-simcore/issues/8273 is fixed this "
-                "will prevent fulfilling this instance type need.",
-                exc_info=True,
+                "Could not start warm buffer instances: %s due to Insufficient Capacity in the current AWS Availability Zone! "
+                "The warm buffer assigned tasks will be moved to new instances if possible.",
+                [i.id for i in instances_to_start],
             )
-            # we need to re-assign the tasks assigned to the warm buffer instances
-            return cluster
+            return _de_assign_tasks_from_warm_buffer_ec2s(cluster, instances_to_start)
+
+        except EC2AccessError:
+            _logger.exception(
+                "Could not start warm buffer instances %s! TIP: This needs to be analysed!"
+                "The warm buffer assigned tasks will be moved to new instances if possible.",
+                [i.id for i in instances_to_start],
+            )
+            return _de_assign_tasks_from_warm_buffer_ec2s(cluster, instances_to_start)
+
         # NOTE: first start the instance and then set the tags in case the instance cannot start (e.g. InsufficientInstanceCapacity)
         await get_ec2_client(app).set_instances_tags(
             started_instances,
@@ -495,15 +543,18 @@ async def _try_start_warm_buffer_instances(
         )
     started_instance_ids = [i.id for i in started_instances]
 
-    return dataclasses.replace(
-        cluster,
-        warm_buffer_ec2s=[
-            i
-            for i in cluster.warm_buffer_ec2s
-            if i.ec2_instance.id not in started_instance_ids
-        ],
-        pending_ec2s=cluster.pending_ec2s
-        + [NonAssociatedInstance(ec2_instance=i) for i in started_instances],
+    return (
+        dataclasses.replace(
+            cluster,
+            warm_buffer_ec2s=[
+                i
+                for i in cluster.warm_buffer_ec2s
+                if i.ec2_instance.id not in started_instance_ids
+            ],
+            pending_ec2s=cluster.pending_ec2s
+            + [NonAssociatedInstance(ec2_instance=i) for i in started_instances],
+        ),
+        [],
     )
 
 
@@ -1243,7 +1294,11 @@ async def _autoscale_cluster(
     cluster = await _activate_drained_nodes(app, cluster)
 
     # 3. start warm buffer instances to cover the remaining tasks
-    cluster = await _try_start_warm_buffer_instances(app, cluster, auto_scaling_mode)
+    cluster, de_assigned_tasks = await _try_start_warm_buffer_instances(
+        app, cluster, auto_scaling_mode
+    )
+    # 3.1 if some tasks were de-assigned, we need to add them to the still pending tasks
+    still_pending_tasks.extend(de_assigned_tasks)
 
     # 4. scale down unused instances
     cluster = await _scale_down_unused_cluster_instances(

services/autoscaling/tests/unit/test_modules_cluster_scaling_dynamic.py

@@ -2454,9 +2454,6 @@ async def _raise_insufficient_capacity_error(*args: Any, **kwargs: Any) -> None:
     )
 
 
-@pytest.mark.xfail(
-    reason="bug described in https://github.com/ITISFoundation/osparc-simcore/issues/8273"
-)
 @pytest.mark.parametrize(
     # NOTE: only the main test test_cluster_scaling_up_and_down is run with all options
     "with_docker_join_drained",
@@ -2495,7 +2492,7 @@ async def test_fresh_instance_is_launched_if_warm_buffers_cannot_start_due_to_in
         InstanceTypeType,
         next(iter(app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES)),
     )
-    await create_buffer_machines(1, warm_buffer_instance_type, "stopped", None)
+    await create_buffer_machines(3, warm_buffer_instance_type, "stopped", None)
 
     # create several tasks that needs more power
     scale_up_params = _ScaleUpParams(

services/clusters-keeper/src/simcore_service_clusters_keeper/utils/clusters.py

@@ -102,7 +102,7 @@ def _convert_to_env_dict(entries: dict[str, Any]) -> str:
         f"WORKERS_EC2_INSTANCES_KEY_NAME={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_KEY_NAME}",
         f"WORKERS_EC2_INSTANCES_MAX_INSTANCES={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_MAX_INSTANCES}",
         f"WORKERS_EC2_INSTANCES_SECURITY_GROUP_IDS={_convert_to_env_list(app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_SECURITY_GROUP_IDS)}",
-        f"WORKERS_EC2_INSTANCES_SUBNET_IDS={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_SUBNET_IDS}",
+        f"WORKERS_EC2_INSTANCES_SUBNET_IDS={_convert_to_env_list(app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_SUBNET_IDS)}",
         f"WORKERS_EC2_INSTANCES_TIME_BEFORE_DRAINING={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_TIME_BEFORE_DRAINING}",
         f"WORKERS_EC2_INSTANCES_TIME_BEFORE_TERMINATION={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_TIME_BEFORE_TERMINATION}",
         f"AUTOSCALING_RABBITMQ={_convert_to_env_dict(model_dump_with_secrets(app_settings.CLUSTERS_KEEPER_PRIMARY_EC2_INSTANCES.PRIMARY_EC2_INSTANCES_RABBIT, show_secrets=True)) if app_settings.CLUSTERS_KEEPER_PRIMARY_EC2_INSTANCES.PRIMARY_EC2_INSTANCES_RABBIT else 'null'}",

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

@@ -95,11 +95,12 @@ def _handle_foreign_key_violation(
 
 def _resolve_grouped_state(states: list[RunningState]) -> RunningState:
     # If any state is not a final state, return STARTED
+
     final_states = {
         RunningState.FAILED,
         RunningState.ABORTED,
         RunningState.SUCCESS,
-        RunningState.UNKNOWN,
+        RunningState.UNKNOWN,  # NOTE: this is NOT a final state, but happens when tasks are missing
     }
     if any(state not in final_states for state in states):
         return RunningState.STARTED
@@ -399,7 +400,6 @@ async def list_all_collection_run_ids_for_user_currently_running_computations(
         product_name: str,
         user_id: UserID,
     ) -> list[CollectionRunID]:
-
         list_query = (
             sa.select(
                 comp_runs.c.collection_run_id,
@@ -493,17 +493,17 @@ async def list_group_by_collection_run_id(
             total_count = await conn.scalar(count_query)
             items = []
             async for row in await conn.stream(list_query):
-                db_states = [DB_TO_RUNNING_STATE[s] for s in row["states"]]
+                db_states = [DB_TO_RUNNING_STATE[s] for s in row.states]
                 resolved_state = _resolve_grouped_state(db_states)
                 items.append(
                     ComputationCollectionRunRpcGet(
-                        collection_run_id=row["collection_run_id"],
-                        project_ids=row["project_ids"],
+                        collection_run_id=row.collection_run_id,
+                        project_ids=row.project_ids,
                         state=resolved_state,
-                        info={} if row["info"] is None else row["info"],
-                        submitted_at=row["submitted_at"],
-                        started_at=row["started_at"],
-                        ended_at=row["ended_at"],
+                        info={} if row.info is None else row.info,
+                        submitted_at=row.submitted_at,
+                        started_at=row.started_at,
+                        ended_at=row.ended_at,
                     )
                 )
             return cast(int, total_count), items

services/director-v2/src/simcore_service_director_v2/utils/computations.py

@@ -65,6 +65,7 @@
         RunningState.PENDING,
         RunningState.NOT_STARTED,
         RunningState.WAITING_FOR_CLUSTER,
+        RunningState.WAITING_FOR_RESOURCES,
     ): RunningState.STARTED,
 }
 

services/director-v2/src/simcore_service_director_v2/utils/db.py

@@ -16,11 +16,10 @@
     StateType.ABORTED: RunningState.ABORTED,
     StateType.WAITING_FOR_RESOURCES: RunningState.WAITING_FOR_RESOURCES,
     StateType.WAITING_FOR_CLUSTER: RunningState.WAITING_FOR_CLUSTER,
+    StateType.UNKNOWN: RunningState.UNKNOWN,
 }
 
-RUNNING_STATE_TO_DB = {v: k for k, v in DB_TO_RUNNING_STATE.items()} | {
-    RunningState.UNKNOWN: StateType.FAILED
-}
+RUNNING_STATE_TO_DB = {v: k for k, v in DB_TO_RUNNING_STATE.items()}
 
 _logger = logging.getLogger(__name__)