🎨 Introduce grouping multiple jobs in task manager (Multiport simulation use case) (🗃️)

api/specs/web-server/_computations.py

@@ -4,6 +4,9 @@
 from fastapi import APIRouter, Depends, status
 from fastapi_pagination import Page
 from models_library.api_schemas_webserver.computations import (
+    ComputationCollectionRunListQueryParams,
+    ComputationCollectionRunPathParams,
+    ComputationCollectionRunTaskListQueryParams,
     ComputationGet,
     ComputationPathParams,
     ComputationRunIterationsLatestListQueryParams,
@@ -95,3 +98,22 @@ async def list_computations_latest_iteration_tasks(
     _query: Annotated[as_query(ComputationTaskListQueryParams), Depends()],
     _path: Annotated[ComputationTaskPathParams, Depends()],
 ): ...
+
+
+@router.get(
+    "/computation-collection-runs",
+    response_model=Page[ComputationTaskRestGet],
+)
+async def list_computation_collection_runs(
+    _query: Annotated[as_query(ComputationCollectionRunListQueryParams), Depends()],
+): ...
+
+
+@router.get(
+    "/computation-collection-runs/{collection_run_id}/tasks",
+    response_model=Page[ComputationTaskRestGet],
+)
+async def list_computation_collection_run_tasks(
+    _query: Annotated[as_query(ComputationCollectionRunTaskListQueryParams), Depends()],
+    _path: Annotated[ComputationCollectionRunPathParams, Depends()],
+): ...

packages/models-library/src/models_library/api_schemas_directorv2/comp_runs.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from typing import Any, NamedTuple
 
+from models_library.computations import CollectionRunID
 from models_library.services_types import ServiceRunID
 from pydantic import (
     AnyUrl,
@@ -63,6 +64,55 @@ class ComputationRunRpcGetPage(NamedTuple):
     total: PositiveInt
 
 
+class ComputationCollectionRunRpcGet(BaseModel):
+    collection_run_id: CollectionRunID
+    project_ids: list[str]
+    state: RunningState
+    info: dict[str, Any]
+    submitted_at: datetime
+    started_at: datetime | None
+    ended_at: datetime | None
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "collection_run_id": "12e0c8b2-bad6-40fb-9948-8dec4f65d4d9",
+                    "project_ids": ["beb16d18-d57d-44aa-a638-9727fa4a72ef"],
+                    "state": "SUCCESS",
+                    "info": {
+                        "wallet_id": 9866,
+                        "user_email": "[email protected]",
+                        "wallet_name": "test",
+                        "product_name": "osparc",
+                        "project_name": "test",
+                        "project_metadata": {
+                            "parent_node_id": "12e0c8b2-bad6-40fb-9948-8dec4f65d4d9",
+                            "parent_node_name": "UJyfwFVYySnPCaLuQIaz",
+                            "parent_project_id": "beb16d18-d57d-44aa-a638-9727fa4a72ef",
+                            "parent_project_name": "qTjDmYPxeqAWfCKCQCYF",
+                            "root_parent_node_id": "37176e84-d977-4993-bc49-d76fcfc6e625",
+                            "root_parent_node_name": "UEXExIZVPeFzGRmMglPr",
+                            "root_parent_project_id": "beb16d18-d57d-44aa-a638-9727fa4a72ef",
+                            "root_parent_project_name": "FuDpjjFIyeNTWRUWCuKo",
+                        },
+                        "node_id_names_map": {},
+                        "simcore_user_agent": "agent",
+                    },
+                    "submitted_at": "2023-01-11 13:11:47.293595",
+                    "started_at": "2023-01-11 13:11:47.293595",
+                    "ended_at": "2023-01-11 13:11:47.293595",
+                }
+            ]
+        }
+    )
+
+
+class ComputationCollectionRunRpcGetPage(NamedTuple):
+    items: list[ComputationCollectionRunRpcGet]
+    total: PositiveInt
+
+
 class ComputationTaskRpcGet(BaseModel):
     project_uuid: ProjectID
     node_id: NodeID
@@ -100,3 +150,42 @@ class ComputationTaskRpcGet(BaseModel):
 class ComputationTaskRpcGetPage(NamedTuple):
     items: list[ComputationTaskRpcGet]
     total: PositiveInt
+
+
+class ComputationCollectionRunTaskRpcGet(BaseModel):
+    project_uuid: ProjectID
+    node_id: NodeID
+    state: RunningState
+    progress: float
+    image: dict[str, Any]
+    started_at: datetime | None
+    ended_at: datetime | None
+    log_download_link: AnyUrl | None
+    service_run_id: ServiceRunID
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "project_uuid": "beb16d18-d57d-44aa-a638-9727fa4a72ef",
+                    "node_id": "12e0c8b2-bad6-40fb-9948-8dec4f65d4d9",
+                    "state": "SUCCESS",
+                    "progress": 0.0,
+                    "image": {
+                        "name": "simcore/services/comp/ti-solutions-optimizer",
+                        "tag": "1.0.19",
+                        "node_requirements": {"CPU": 8.0, "RAM": 25769803776},
+                    },
+                    "started_at": "2023-01-11 13:11:47.293595",
+                    "ended_at": "2023-01-11 13:11:47.293595",
+                    "log_download_link": "https://example.com/logs",
+                    "service_run_id": "comp_1_12e0c8b2-bad6-40fb-9948-8dec4f65d4d9_1",
+                }
+            ]
+        }
+    )
+
+
+class ComputationCollectionRunTaskRpcGetPage(NamedTuple):
+    items: list[ComputationCollectionRunTaskRpcGet]
+    total: PositiveInt

packages/models-library/src/models_library/api_schemas_directorv2/computations.py

@@ -1,5 +1,6 @@
 from typing import Annotated, Any, TypeAlias
 
+from models_library.computations import CollectionRunID
 from pydantic import (
     AnyHttpUrl,
     AnyUrl,
@@ -72,6 +73,7 @@ class ComputationCreate(BaseModel):
             description="contains information about the wallet used to bill the running service"
         ),
     ] = None
+    collection_run_id: CollectionRunID
 
     @field_validator("product_name")
     @classmethod

packages/models-library/src/models_library/api_schemas_webserver/computations.py

@@ -8,17 +8,20 @@
     BaseModel,
     ConfigDict,
     Field,
+    field_validator,
 )
 
 from ..api_schemas_directorv2.computations import (
     ComputationGet as _DirectorV2ComputationGet,
 )
 from ..basic_types import IDStr
+from ..computations import CollectionRunID
 from ..projects import CommitID, ProjectID
 from ..projects_nodes_io import NodeID
 from ..projects_state import RunningState
 from ..rest_ordering import OrderBy, create_ordering_query_model_class
 from ..rest_pagination import PageQueryParameters
+from ..utils.common_validators import null_or_none_str_to_none_validator
 from ._base import (
     InputSchemaWithoutCamelCase,
     OutputSchema,
@@ -153,3 +156,56 @@ class ComputationTaskRestGet(OutputSchema):
     log_download_link: AnyUrl | None
     node_name: str
     osparc_credits: Decimal | None
+
+
+### Computation Collection Run
+
+
+class ComputationCollectionRunListQueryParams(
+    PageQueryParameters,
+):
+    filter_only_running: bool = Field(
+        default=False,
+        description="If true, only running collection runs are returned",
+    )
+    filter_by_root_project_id: ProjectID | None = Field(
+        default=None,
+    )
+
+    _null_or_none_to_none = field_validator("filter_by_root_project_id", mode="before")(
+        null_or_none_str_to_none_validator
+    )
+
+
+class ComputationCollectionRunRestGet(OutputSchema):
+    collection_run_id: CollectionRunID
+    project_ids: list[str]
+    state: RunningState
+    info: dict[str, Any]
+    submitted_at: datetime
+    started_at: datetime | None
+    ended_at: datetime | None
+    name: str
+
+
+class ComputationCollectionRunPathParams(BaseModel):
+    collection_run_id: CollectionRunID
+    model_config = ConfigDict(populate_by_name=True, extra="forbid")
+
+
+class ComputationCollectionRunTaskListQueryParams(
+    PageQueryParameters,
+): ...
+
+
+class ComputationCollectionRunTaskRestGet(OutputSchema):
+    project_uuid: ProjectID
+    node_id: NodeID
+    state: RunningState
+    progress: float
+    image: dict[str, Any]
+    started_at: datetime | None
+    ended_at: datetime | None
+    log_download_link: AnyUrl | None
+    osparc_credits: Decimal | None
+    name: str

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

@@ -1,6 +1,7 @@
 from datetime import datetime
 from decimal import Decimal
-from typing import Any
+from typing import Any, TypeAlias
+from uuid import UUID
 
 from pydantic import AnyUrl, BaseModel
 
@@ -36,3 +37,34 @@ class ComputationRunWithAttributes(BaseModel):
     # Attributes added by the webserver
     root_project_name: str
     project_custom_metadata: dict[str, Any]
+
+
+CollectionRunID: TypeAlias = UUID
+
+
+class ComputationCollectionRunWithAttributes(BaseModel):
+    collection_run_id: CollectionRunID
+    project_ids: list[str]
+    state: RunningState
+    info: dict[str, Any]
+    submitted_at: datetime
+    started_at: datetime | None
+    ended_at: datetime | None
+
+    # Attributes added by the webserver
+    name: str  # Either root project name or collection name if provided by the client on start
+
+
+class ComputationCollectionRunTaskWithAttributes(BaseModel):
+    project_uuid: ProjectID
+    node_id: NodeID
+    state: RunningState
+    progress: float
+    image: dict[str, Any]
+    started_at: datetime | None
+    ended_at: datetime | None
+    log_download_link: AnyUrl | None
+
+    # Attributes added by the webserver
+    name: str  # Either node name or job name if provided by the client on start
+    osparc_credits: Decimal | None

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

@@ -24,16 +24,24 @@ class RunningState(str, Enum):
     SEE StateType for task state
     """
 
-    UNKNOWN = "UNKNOWN"
-    PUBLISHED = "PUBLISHED"
-    NOT_STARTED = "NOT_STARTED"
-    PENDING = "PENDING"
-    WAITING_FOR_RESOURCES = "WAITING_FOR_RESOURCES"
-    STARTED = "STARTED"
-    SUCCESS = "SUCCESS"
-    FAILED = "FAILED"
-    ABORTED = "ABORTED"
-    WAITING_FOR_CLUSTER = "WAITING_FOR_CLUSTER"
+    UNKNOWN = "UNKNOWN"  # Error state: the task disappeared from the system (Dask backend lost the task and its state is unknown)
+    NOT_STARTED = "NOT_STARTED"  # Default state when the project is created
+
+    PUBLISHED = (
+        "PUBLISHED"  # The project is awaiting the compute scheduler to pick up the task
+    )
+
+    PENDING = "PENDING"  # The compute scheduler has taken over, and the task is waiting for a worker to pick it up
+
+    WAITING_FOR_CLUSTER = "WAITING_FOR_CLUSTER"  # No cluster is available to run the task; waiting for one to become available
+    WAITING_FOR_RESOURCES = "WAITING_FOR_RESOURCES"  # No worker is available to run the task; waiting for one to become available
+    # PENDING -> WAITING_FOR_CLUSTER -> PENDING -> WAITING_FOR_RESOURCES -> PENDING -> STARTED
+
+    STARTED = "STARTED"  # A worker has picked up the task and is executing it
+
+    SUCCESS = "SUCCESS"  # Task finished successfully
+    FAILED = "FAILED"  # Task finished with an error
+    ABORTED = "ABORTED"  # Task was aborted before completion
 
     @staticmethod
     def list_running_states() -> list["RunningState"]: