Merge branch 'add-project-grouping-for-task-manager' into add-filter-on-active-jobs

Author: matusdrobuliak66

services/director-v2/src/simcore_service_director_v2/api/routes/computations.py

@@ -224,6 +224,11 @@ async def _try_start_pipeline(
         wallet_id = computation.wallet_info.wallet_id
         wallet_name = computation.wallet_info.wallet_name
 
+    if computation.collection_run_id is None:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Project {computation.project_id} has no collection run ID",
+        )
     await run_new_pipeline(
         app,
         user_id=computation.user_id,

services/director-v2/tests/integration/01/test_computation_api.py

@@ -170,7 +170,6 @@ def fake_workbench_computational_pipeline_details_not_started(
                 "project_id": "16e60a5d-834e-4267-b44d-3af49171bf21",
                 "product_name": "not a product",
                 "product_api_base_url": "http://invalid",
-                "collection_run_id": "16e60a5d-834e-4267-b44d-3af49171bf21",
             },
             status.HTTP_404_NOT_FOUND,
         ),
@@ -997,7 +996,6 @@ async def test_pipeline_with_control_loop_made_of_dynamic_services_is_allowed(
             "start_pipeline": False,
             "product_name": osparc_product_name,
             "product_api_base_url": osparc_product_api_base_url,
-            "collection_run_id": str(uuid.uuid4()),
         },
     )
     assert (
@@ -1084,7 +1082,6 @@ async def test_pipeline_with_cycle_containing_a_computational_service_is_forbidd
             "start_pipeline": False,
             "product_name": osparc_product_name,
             "product_api_base_url": osparc_product_api_base_url,
-            "collection_run_id": str(uuid.uuid4()),
         },
     )
     assert (

services/director-v2/tests/integration/conftest.py

@@ -100,7 +100,9 @@ async def _creator(
                 "start_pipeline": start_pipeline,
                 "product_name": product_name,
                 "product_api_base_url": product_api_base_url,
-                "collection_run_id": str(uuid.uuid4()),
+                "collection_run_id": (
+                    str(uuid.uuid4()) if start_pipeline is True else None
+                ),
                 **kwargs,
             },
         )

services/director-v2/tests/unit/with_dbs/comp_scheduler/test_api_route_computations.py

@@ -387,15 +387,13 @@ async def test_computation_create_validators(
         product_name=product_name,
         product_api_base_url=product_api_base_url,
         use_on_demand_clusters=True,
-        collection_run_id=fake_collection_run_id,
     )
     ComputationCreate(
         user_id=user["id"],
         project_id=proj.uuid,
         product_name=product_name,
         product_api_base_url=product_api_base_url,
         use_on_demand_clusters=False,
-        collection_run_id=fake_collection_run_id,
     )
 
 
@@ -422,7 +420,6 @@ async def test_create_computation(
                 project_id=proj.uuid,
                 product_name=product_name,
                 product_api_base_url=product_api_base_url,
-                collection_run_id=fake_collection_run_id,
             )
         ),
     )
@@ -540,7 +537,6 @@ async def test_create_computation_with_wallet(
                 product_name=product_name,
                 product_api_base_url=product_api_base_url,
                 wallet_info=wallet_info,
-                collection_run_id=fake_collection_run_id,
             )
         ),
     )
@@ -644,7 +640,6 @@ async def test_create_computation_with_wallet_with_invalid_pricing_unit_name_rai
                 product_name=product_name,
                 product_api_base_url=product_api_base_url,
                 wallet_info=wallet_info,
-                collection_run_id=fake_collection_run_id,
             )
         ),
     )
@@ -686,7 +681,6 @@ async def test_create_computation_with_wallet_with_no_clusters_keeper_raises_503
                 product_name=product_name,
                 product_api_base_url=product_api_base_url,
                 wallet_info=wallet_info,
-                collection_run_id=fake_collection_run_id,
             )
         ),
     )
@@ -702,13 +696,40 @@ async def test_start_computation_without_product_fails(
     create_registered_user: Callable[..., dict[str, Any]],
     create_project: Callable[..., Awaitable[ProjectAtDB]],
     async_client: httpx.AsyncClient,
+    fake_collection_run_id: CollectionRunID,
+):
+    user = create_registered_user()
+    proj = await create_project(user, workbench=fake_workbench_without_outputs)
+    create_computation_url = httpx.URL("/v2/computations")
+    response = await async_client.post(
+        create_computation_url,
+        json={
+            "user_id": f"{user['id']}",
+            "project_id": f"{proj.uuid}",
+            "start_pipeline": f"{True}",
+            "collection_run_id": f"{fake_collection_run_id}",
+        },
+    )
+    assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY, response.text
+
+
+async def test_start_computation_without_collection_run_id_fails(
+    minimal_configuration: None,
+    mocked_director_service_fcts: respx.MockRouter,
+    mocked_catalog_service_fcts: respx.MockRouter,
+    product_name: str,
+    fake_workbench_without_outputs: dict[str, Any],
+    create_registered_user: Callable[..., dict[str, Any]],
+    create_project: Callable[..., Awaitable[ProjectAtDB]],
+    async_client: httpx.AsyncClient,
 ):
     user = create_registered_user()
     proj = await create_project(user, workbench=fake_workbench_without_outputs)
     create_computation_url = httpx.URL("/v2/computations")
     response = await async_client.post(
         create_computation_url,
         json={
+            "product_name": product_name,
             "user_id": f"{user['id']}",
             "project_id": f"{proj.uuid}",
             "start_pipeline": f"{True}",

services/storage/src/simcore_service_storage/modules/s3.py

@@ -7,19 +7,42 @@
 from common_library.json_serialization import json_dumps
 from fastapi import FastAPI
 from pydantic import TypeAdapter
-from servicelib.logging_utils import log_context
+from tenacity import retry, wait_random_exponential
 from tenacity.asyncio import AsyncRetrying
 from tenacity.before_sleep import before_sleep_log
 from tenacity.wait import wait_fixed
 from types_aiobotocore_s3.literals import BucketLocationConstraintType
 
 from ..constants import RETRY_WAIT_SECS
-from ..core.settings import get_application_settings
+from ..core.settings import ApplicationSettings, get_application_settings
 from ..exceptions.errors import ConfigurationError
 
 _logger = logging.getLogger(__name__)
 
 
+@retry(
+    wait=wait_random_exponential(),
+    before_sleep=before_sleep_log(_logger, logging.WARNING),
+    reraise=True,
+)
+async def _ensure_s3_bucket(
+    client: SimcoreS3API, settings: ApplicationSettings
+) -> None:
+    assert settings.STORAGE_S3  # nosec
+    if await client.bucket_exists(bucket=settings.STORAGE_S3.S3_BUCKET_NAME):
+        _logger.info(
+            "S3 bucket %s exists already, skipping creation",
+            settings.STORAGE_S3.S3_BUCKET_NAME,
+        )
+        return
+    await client.create_bucket(
+        bucket=settings.STORAGE_S3.S3_BUCKET_NAME,
+        region=TypeAdapter(
+            BucketLocationConstraintType | Literal["us-east-1"]
+        ).validate_python(settings.STORAGE_S3.S3_REGION),
+    )
+
+
 def setup_s3(app: FastAPI) -> None:
     async def _on_startup() -> None:
         app.state.s3_client = None
@@ -44,14 +67,7 @@ async def _on_startup() -> None:
                 assert client  # nosec
         app.state.s3_client = client
 
-        with log_context(_logger, logging.DEBUG, msg="setup.s3_bucket.cleanup_ctx"):
-            assert settings.STORAGE_S3  # nosec
-            await client.create_bucket(
-                bucket=settings.STORAGE_S3.S3_BUCKET_NAME,
-                region=TypeAdapter(
-                    BucketLocationConstraintType | Literal["us-east-1"]
-                ).validate_python(settings.STORAGE_S3.S3_REGION),
-            )
+        await _ensure_s3_bucket(client, settings)
 
     async def _on_shutdown() -> None:
         if app.state.s3_client: