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

Author: matusdrobuliak66

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,12 @@ class ComputationCreate(BaseModel):
             description="contains information about the wallet used to bill the running service"
         ),
     ] = None
+    collection_run_id: Annotated[
+        CollectionRunID | None,
+        Field(
+            description="In case start_pipeline is True, this is the collection run id to which the comp run belongs."
+        ),
+    ] = None
 
     @field_validator("product_name")
     @classmethod
@@ -83,6 +90,20 @@ def _ensure_product_name_defined_if_computation_starts(
             raise ValueError(msg)
         return v
 
+    @field_validator("collection_run_id")
+    @classmethod
+    def _ensure_collection_run_id_dependency_on_start_pipeline(
+        cls, v, info: ValidationInfo
+    ):
+        start_pipeline = info.data.get("start_pipeline")
+        if start_pipeline and v is None:
+            msg = "collection_run_id must be provided when start_pipeline is True!"
+            raise ValueError(msg)
+        if not start_pipeline and v is not None:
+            msg = "collection_run_id must be None when start_pipeline is False!"
+            raise ValueError(msg)
+        return v
+
 
 class ComputationStop(BaseModel):
     user_id: UserID

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,54 @@ class ComputationTaskRestGet(OutputSchema):
     log_download_link: AnyUrl | None
     node_name: str
     osparc_credits: Decimal | None
+
+
+### Computation Collection Run
+
+
+class ComputationCollectionRunListQueryParams(
+    PageQueryParameters,
+):
+    filter_only_running: Annotated[
+        bool, Field(description="If true, only running collection runs are returned")
+    ] = False
+
+    filter_by_root_project_id: ProjectID | None = 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

@@ -22,18 +22,32 @@ class RunningState(str, Enum):
     """State of execution of a project's computational workflow
 
     SEE StateType for task state
+
+    # Computational backend states explained:
+    - UNKNOWN - The backend doesn't know about the task anymore, it has disappeared from the system or it was never created (eg. when we are asking for the task)
+    - NOT_STARTED - Default state when the task is created
+    - PUBLISHED - The task has been submitted to the computational backend (click on "Run" button in the UI)
+    - PENDING - Task has been transferred to the Dask scheduler and is waiting for a worker to pick it up (director-v2 --> Dask scheduler)
+       - But! it is also transition state (ex. PENDING -> WAITING_FOR_CLUSTER -> PENDING -> WAITING_FOR_RESOURCES -> PENDING -> STARTED)
+    - WAITING_FOR_CLUSTER - No cluster (Dask scheduler) is available to run the task; waiting for one to become available
+    - WAITING_FOR_RESOURCES - No worker (Dask worker) is available to run the task; waiting for one to become available
+    - STARTED - A worker has picked up the task and is executing it
+    - SUCCESS - Task finished successfully
+    - FAILED - Task finished with an error
+    - ABORTED - Task was aborted before completion
+
     """
 
     UNKNOWN = "UNKNOWN"
-    PUBLISHED = "PUBLISHED"
     NOT_STARTED = "NOT_STARTED"
+    PUBLISHED = "PUBLISHED"
     PENDING = "PENDING"
+    WAITING_FOR_CLUSTER = "WAITING_FOR_CLUSTER"
     WAITING_FOR_RESOURCES = "WAITING_FOR_RESOURCES"
     STARTED = "STARTED"
     SUCCESS = "SUCCESS"
     FAILED = "FAILED"
     ABORTED = "ABORTED"
-    WAITING_FOR_CLUSTER = "WAITING_FOR_CLUSTER"
 
     @staticmethod
     def list_running_states() -> list["RunningState"]: