Merge branch 'master' into add-functions-locust-test

Author: bisgaard-itis

packages/dask-task-models-library/src/dask_task_models_library/container_tasks/io.py

@@ -195,7 +195,9 @@ def from_task_output(
             with suppress(json.JSONDecodeError):
                 # NOTE: The suppression here is ok, since if the data is empty,
                 # there will be a validation error anyway
-                data = json_loads(output_data_file.read_text())
+                loaded_data = json_loads(output_data_file.read_text())
+                # ignore what is not in the schema
+                data = {k: v for k, v in loaded_data.items() if k in schema}
 
         for output_key, output_params in schema.items():
             if isinstance(output_params, FilePortSchema):

packages/dask-task-models-library/tests/container_tasks/test_io.py

@@ -184,3 +184,39 @@ def test_objects_are_compatible_with_dask_requirements(model_cls, model_cls_exam
         model_instance = model_cls.model_validate(example)
         reloaded_instance = loads(dumps(model_instance))
         assert reloaded_instance == model_instance
+
+
+def test_create_task_output_from_task_ignores_additional_entries(
+    tmp_path: Path, faker: Faker
+):
+    task_output_schema = TaskOutputDataSchema.model_validate(
+        {
+            "some_output_1": {
+                "required": True,
+            },
+            "some_output_2": {
+                "required": True,
+            },
+        }
+    )
+    output_file = _create_fake_outputs(task_output_schema, tmp_path, False, faker)
+    assert output_file
+    # Add more data to the output file to simulate additional entries
+    file_path = tmp_path / output_file
+    data = json.loads(file_path.read_text())
+    # Ensure the file contains the expected keys first
+    for key in task_output_schema:
+        assert key in data
+    # Add an extra key
+    data["extra_key"] = "extra_value"
+    file_path.write_text(json.dumps(data))
+
+    task_output_data = TaskOutputData.from_task_output(
+        schema=task_output_schema,
+        output_folder=tmp_path,
+        output_file_ext=output_file,
+    )
+    # Only keys defined in the schema should be present
+    assert set(task_output_data.keys()) == set(
+        task_output_schema.keys()
+    ), "Should only contain the expected keys"

scripts/maintenance/computational-clusters/pyproject.toml

@@ -6,8 +6,8 @@ dependencies = [
   "black",
   "boto3",
   # NOTE: these must be in sync with ospar
-  "cloudpickle==3.1.0",
-  "dask[distributed]==2024.12.0",
+  "cloudpickle",
+  "dask[distributed]",
   "mypy_boto3_ec2",
   "types-boto3",
   "parse",
@@ -20,7 +20,7 @@ dependencies = [
   "sqlalchemy[asyncio]",
   "sshtunnel",
   "ansible>=10.7.0",
-  "lz4==4.3.3",
+  "lz4",
 ]
 name = "autoscaled-monitor"
 version = "1.0.0"

scripts/maintenance/computational-clusters/uv.lock

@@ -1,4 +1,4 @@
-""" Defines the different exceptions that may arise in the director
+"""Defines the different exceptions that may arise in the director
 
 
 TODO: Exceptions should provide all info to create Error instances of the API model
@@ -88,13 +88,14 @@ def __init__(self, project_id: ProjectID, node_id: NodeID, **ctx: Any) -> None:
 
     def get_errors(self) -> list[ErrorDict]:
         # default implementation
+
         return [
             {
                 "loc": (
                     f"{self.project_id}",
                     f"{self.node_id}",
                 ),
-                "msg": f"{self.args[0]}",
+                "msg": f"{self}",
                 "type": self.code,
             },
         ]

services/director-v2/src/simcore_service_director_v2/core/errors.py

@@ -115,48 +115,54 @@ class SortedTasks:
     potentially_lost: list[CompTaskAtDB]
 
 
+_MAX_WAITING_TIME_FOR_UNKNOWN_TASKS: Final[datetime.timedelta] = datetime.timedelta(
+    seconds=30
+)
+
+
 async def _triage_changed_tasks(
-    changed_tasks_or_executing: list[tuple[_Previous, _Current]],
+    changed_tasks: list[tuple[_Previous, _Current]],
 ) -> SortedTasks:
     started_tasks = [
         current
-        for previous, current in changed_tasks_or_executing
+        for previous, current in changed_tasks
         if current.state in RUNNING_STATES
         or (
             previous.state in WAITING_FOR_START_STATES
             and current.state in COMPLETED_STATES
         )
     ]
 
-    # NOTE: some tasks can be both started and completed since we might have the time they were running
     completed_tasks = [
-        current
-        for _, current in changed_tasks_or_executing
-        if current.state in COMPLETED_STATES
+        current for _, current in changed_tasks if current.state in COMPLETED_STATES
     ]
 
     waiting_for_resources_tasks = [
         current
-        for previous, current in changed_tasks_or_executing
+        for previous, current in changed_tasks
         if current.state in WAITING_FOR_START_STATES
     ]
 
-    lost_or_momentarily_lost_tasks = [
+    lost_tasks = [
         current
-        for _, current in changed_tasks_or_executing
-        if current.state is RunningState.UNKNOWN
+        for previous, current in changed_tasks
+        if (current.state is RunningState.UNKNOWN)
+        and (
+            (arrow.utcnow().datetime - previous.modified)
+            > _MAX_WAITING_TIME_FOR_UNKNOWN_TASKS
+        )
     ]
-    if lost_or_momentarily_lost_tasks:
+    if lost_tasks:
         _logger.warning(
             "%s are currently in unknown state. TIP: If they are running in an external cluster and it is not yet ready, that might explain it. But inform @sanderegg nevertheless!",
-            [t.node_id for t in lost_or_momentarily_lost_tasks],
+            [t.node_id for t in lost_tasks],
         )
 
     return SortedTasks(
         started_tasks,
         completed_tasks,
         waiting_for_resources_tasks,
-        lost_or_momentarily_lost_tasks,
+        lost_tasks,
     )
 
 
@@ -500,7 +506,7 @@ async def _update_states_from_comp_backend(
         # PENDING -> WAITING_FOR_RESOURCES (workers creation or missing) -> PENDING -> STARTED (worker started processing the task) -> SUCCESS/FAILED
         # or ABORTED (user cancelled) or UNKNOWN (lost task - it might be transient, be careful with this one)
         sorted_tasks = await _triage_changed_tasks(tasks_with_changed_states)
-
+        _logger.debug("found the following %s tasks with changed states", sorted_tasks)
         # now process the tasks
         if sorted_tasks.started:
             # NOTE: the dask-scheduler cannot differentiate between tasks that are effectively computing and

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_base.py

@@ -23,9 +23,6 @@
 from models_library.users import UserID
 from servicelib.common_headers import UNDEFINED_DEFAULT_SIMCORE_USER_AGENT_VALUE
 from servicelib.logging_utils import log_catch
-from simcore_service_director_v2.modules.comp_scheduler._utils import (
-    WAITING_FOR_START_STATES,
-)
 
 from ...core.errors import (
     ComputationalBackendNotConnectedError,
@@ -52,6 +49,9 @@
 )
 from ..db.repositories.comp_tasks import CompTasksRepository
 from ._scheduler_base import BaseCompScheduler
+from ._utils import (
+    WAITING_FOR_START_STATES,
+)
 
 _logger = logging.getLogger(__name__)
 

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_dask.py

@@ -517,7 +517,7 @@ async def get_task_result(self, job_id: str) -> TaskOutputData:
         except KeyError as exc:
             raise ComputationalBackendTaskNotFoundError(job_id=job_id) from exc
         except distributed.TimeoutError as exc:
-            raise ComputationalBackendTaskResultsNotReadyError from exc
+            raise ComputationalBackendTaskResultsNotReadyError(job_id=job_id) from exc
 
     async def release_task_result(self, job_id: str) -> None:
         _logger.debug("releasing results for %s", f"{job_id=}")

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

@@ -34,6 +34,7 @@
     NodeportsException,
     S3InvalidPathError,
     StorageInvalidCall,
+    UnboundPortError,
 )
 from simcore_sdk.node_ports_v2 import FileLinkType, Port, links, port_utils
 from simcore_sdk.node_ports_v2.links import ItemValue as _NPItemValue
@@ -147,6 +148,20 @@ async def parse_output_data(
             await (await ports.outputs)[port_key].set_value(value_to_transfer)
         except ValidationError as err:
             ports_errors.extend(_get_port_validation_errors(port_key, err))
+        except UnboundPortError as err:
+            ports_errors.extend(
+                [
+                    {
+                        "loc": (
+                            f"{project_id}",
+                            f"{node_id}",
+                            f"{port_key}",
+                        ),
+                        "msg": str(err),
+                        "type": "unbound_port",
+                    }
+                ]
+            )
 
     if ports_errors:
         raise PortsValidationError(

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

@@ -1316,6 +1316,14 @@ async def test_handling_of_disconnected_scheduler_dask(
     )
 
 
+@pytest.fixture
+def with_disabled_unknown_max_time(mocker: MockerFixture) -> None:
+    mocker.patch(
+        "simcore_service_director_v2.modules.comp_scheduler._scheduler_base._MAX_WAITING_TIME_FOR_UNKNOWN_TASKS",
+        new=datetime.timedelta(0),
+    )
+
+
 @dataclass(frozen=True, kw_only=True)
 class RebootState:
     dask_task_status: RunningState
@@ -1397,6 +1405,7 @@ class RebootState:
 async def test_handling_scheduled_tasks_after_director_reboots(
     with_disabled_auto_scheduling: mock.Mock,
     with_disabled_scheduler_publisher: mock.Mock,
+    with_disabled_unknown_max_time: None,
     mocked_dask_client: mock.MagicMock,
     sqlalchemy_async_engine: AsyncEngine,
     running_project: RunningProject,