🐛 Is3240/fix pull inputs sidecar (ITISFoundation#3280)

Author: pcrespov

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/events.py

@@ -583,8 +583,6 @@ async def _remove_containers_save_state_and_outputs() -> None:
                 scheduler_data.dynamic_sidecar.service_removal_state.can_save
                 and scheduler_data.dynamic_sidecar.were_containers_created
             ):
-                dynamic_sidecar_client = get_dynamic_sidecar_client(app)
-
                 logger.info(
                     "Calling into dynamic-sidecar to save state and pushing data to nodeports"
                 )

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/_routing.py

@@ -1,18 +1,29 @@
-# module acting as root for all routes
+""" Module to collect, tag and prefix all routes under 'main_router'
+
+Setup and register all routes here form different modules
+"""
 
 from fastapi import APIRouter
 
 from .._meta import API_VTAG
-from .containers import containers_router
-from .containers_extension import containers_router_extension
-from .containers_tasks import containers_router_tasks
-from .health import health_router
+from . import containers, containers_extension, containers_long_running_tasks, health
 
-# setup and register all routes here form different modules
 main_router = APIRouter()
-main_router.include_router(health_router)
-main_router.include_router(containers_router, prefix=f"/{API_VTAG}")
-main_router.include_router(containers_router_extension, prefix=f"/{API_VTAG}")
-main_router.include_router(containers_router_tasks, prefix=f"/{API_VTAG}")
+main_router.include_router(health.router)
+main_router.include_router(
+    containers.router,
+    tags=["containers"],
+    prefix=f"/{API_VTAG}",
+)
+main_router.include_router(
+    containers_extension.router,
+    tags=["containers"],
+    prefix=f"/{API_VTAG}",
+)
+main_router.include_router(
+    containers_long_running_tasks.router,
+    tags=["containers"],
+    prefix=f"/{API_VTAG}",
+)
 
 __all__: tuple[str, ...] = ("main_router",)

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/containers.py

@@ -26,10 +26,10 @@ def _raise_if_container_is_missing(
         raise HTTPException(status.HTTP_404_NOT_FOUND, detail=message)
 
 
-containers_router = APIRouter(tags=["containers"])
+router = APIRouter()
 
 
-@containers_router.get(
+@router.get(
     "/containers",
     responses={
         status.HTTP_500_INTERNAL_SERVER_ERROR: {"description": "Errors in container"}
@@ -74,7 +74,20 @@ def _format_result(container_inspect: dict[str, Any]) -> dict[str, Any]:
         return results
 
 
-@containers_router.get(
+# Some of the operations and sub-resources on containers are implemented as long-running tasks.
+# Handlers for these operations can be found in:
+#
+# POST /containers                       : SEE containers_long_running_tasks::create_service_containers_task
+# POST /containers:down                  : SEE containers_long_running_tasks::runs_docker_compose_down_task
+# POST /containers/state:restore         : SEE containers_long_running_tasks::state_restore_task
+# POST /containers/state:save            : SEE containers_long_running_tasks::state_save_task
+# POST /containers/ports/inputs:pull     : SEE containers_long_running_tasks::ports_inputs_pull_task
+# POST /containers/ports/outputs:pull    : SEE containers_long_running_tasks::ports_outputs_pull_task
+# POST /containers/ports/outputs:push    : SEE containers_long_running_tasks::ports_outputs_push_task
+#
+
+
+@router.get(
     "/containers/{id}/logs",
     responses={
         status.HTTP_404_NOT_FOUND: {
@@ -120,7 +133,7 @@ async def get_container_logs(
         return container_logs
 
 
-@containers_router.get(
+@router.get(
     "/containers/name",
     responses={
         status.HTTP_404_NOT_FOUND: {
@@ -186,7 +199,7 @@ async def get_containers_name(
     return f"{container_name}"
 
 
-@containers_router.get(
+@router.get(
     "/containers/{id}",
     responses={
         status.HTTP_404_NOT_FOUND: {"description": "Container does not exist"},

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/containers_extension.py

@@ -45,15 +45,10 @@ async def send_message(rabbitmq: RabbitMQ, message: str) -> None:
 #
 # HANDLERS ------------------
 #
-# - ANE: importing the `containers_router` router from .containers
-# and generating the openapi spec, will not add the below entrypoints
-# we need to create a new one in order for all the APIs to be
-# detected as before
-#
-containers_router_extension = APIRouter(tags=["containers"])
+router = APIRouter()
 
 
-@containers_router_extension.patch(
+@router.patch(
     "/containers/directory-watcher",
     summary="Enable/disable directory-watcher event propagation",
     response_class=Response,
@@ -69,7 +64,7 @@ async def toggle_directory_watcher(
         directory_watcher.disable_directory_watcher(app)
 
 
-@containers_router_extension.post(
+@router.post(
     "/containers/ports/outputs/dirs",
     summary=(
         "Creates the output directories declared by the docker images's labels. "
@@ -91,7 +86,7 @@ async def create_output_dirs(
             dir_to_create.mkdir(parents=True, exist_ok=True)
 
 
-@containers_router_extension.post(
+@router.post(
     "/containers/{id}/networks:attach",
     summary="attach container to a network, if not already attached",
     response_class=Response,
@@ -132,7 +127,7 @@ async def attach_container_to_network(
         )
 
 
-@containers_router_extension.post(
+@router.post(
     "/containers/{id}/networks:detach",
     summary="detach container from a network, if not already detached",
     response_class=Response,

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/containers_tasks.py renamed to services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/containers_long_running_tasks.py

@@ -38,14 +38,10 @@
 )
 
 logger = logging.getLogger(__name__)
+router = APIRouter()
 
-containers_router_tasks = APIRouter(tags=["containers"])
 
-
-# HANDLERS
-
-
-@containers_router_tasks.post(
+@router.post(
     "/containers",
     summary=dedent(
         """
@@ -98,7 +94,7 @@ async def create_service_containers_task(  # pylint: disable=too-many-arguments
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers:down",
     summary="Remove the previously started containers",
     status_code=status.HTTP_202_ACCEPTED,
@@ -133,7 +129,7 @@ async def runs_docker_compose_down_task(
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers/state:restore",
     summary="Restores the state of the dynamic service",
     status_code=status.HTTP_202_ACCEPTED,
@@ -168,7 +164,7 @@ async def state_restore_task(
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers/state:save",
     summary="Stores the state of the dynamic service",
     status_code=status.HTTP_202_ACCEPTED,
@@ -203,7 +199,7 @@ async def state_save_task(
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers/ports/inputs:pull",
     summary="Pull input ports data",
     status_code=status.HTTP_202_ACCEPTED,
@@ -238,7 +234,7 @@ async def ports_inputs_pull_task(
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers/ports/outputs:pull",
     summary="Pull output ports data",
     status_code=status.HTTP_202_ACCEPTED,
@@ -273,7 +269,7 @@ async def ports_outputs_pull_task(
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers/ports/outputs:push",
     summary="Push output ports data",
     status_code=status.HTTP_202_ACCEPTED,
@@ -308,7 +304,7 @@ async def ports_outputs_push_task(
         raise HTTPException(status.HTTP_409_CONFLICT, detail=f"{e}") from e
 
 
-@containers_router_tasks.post(
+@router.post(
     "/containers:restart",
     summary="Restarts previously started containers",
     status_code=status.HTTP_202_ACCEPTED,

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/health.py

@@ -3,10 +3,10 @@
 from ..models.schemas.application_health import ApplicationHealth
 from ._dependencies import get_application_health
 
-health_router = APIRouter()
+router = APIRouter()
 
 
-@health_router.get(
+@router.get(
     "/health",
     response_model=ApplicationHealth,
     responses={
@@ -22,6 +22,3 @@ async def health_endpoint(
         )
 
     return application_health
-
-
-__all__: tuple[str, ...] = ("health_router",)

services/dynamic-sidecar/tests/unit/test_api_containers_tasks.py renamed to services/dynamic-sidecar/tests/unit/test_api_containers_long_running_tasks.py

@@ -37,7 +37,7 @@
 from servicelib.fastapi.long_running_tasks.client import setup as client_setup
 from simcore_sdk.node_ports_common.exceptions import NodeNotFound
 from simcore_service_dynamic_sidecar._meta import API_VTAG
-from simcore_service_dynamic_sidecar.api import containers_tasks
+from simcore_service_dynamic_sidecar.api import containers_long_running_tasks
 from simcore_service_dynamic_sidecar.models.shared_store import SharedStore
 
 FAST_STATUS_POLL: Final[float] = 0.1
@@ -71,12 +71,14 @@ async def _just_log_task(*args, **kwargs) -> None:
     # searching by name since all start with _task
     tasks_names = [
         x[0]
-        for x in getmembers(containers_tasks, isfunction)
+        for x in getmembers(containers_long_running_tasks, isfunction)
         if x[0].startswith("task")
     ]
 
     for task_name in tasks_names:
-        mocker.patch.object(containers_tasks, task_name, new=_just_log_task)
+        mocker.patch.object(
+            containers_long_running_tasks, task_name, new=_just_log_task
+        )
 
     yield None
 

services/web/server/src/simcore_service_webserver/computation_comp_tasks_listening_task.py

@@ -7,7 +7,7 @@
 import logging
 from contextlib import suppress
 from pprint import pformat
-from typing import Dict, List, Optional
+from typing import Optional
 
 from aiohttp import web
 from aiopg.sa import Engine
@@ -44,7 +44,7 @@ async def _update_project_state(
     project_uuid: str,
     node_uuid: str,
     new_state: RunningState,
-    node_errors: Optional[List[ErrorDict]],
+    node_errors: Optional[list[ErrorDict]],
 ) -> None:
     project = await projects_api.update_project_node_state(
         app, user_id, project_uuid, node_uuid, new_state
@@ -73,7 +73,7 @@ async def listen(app: web.Application, db_engine: Engine):
                 "received update from database: %s", pformat(notification.payload)
             )
             # get the data and the info on what changed
-            payload: Dict = json.loads(notification.payload)
+            payload: dict = json.loads(notification.payload)
 
             # FIXME: all this should move to rabbitMQ instead of this
             task_data = payload.get("data", {})
@@ -100,6 +100,7 @@ async def listen(app: web.Application, db_engine: Engine):
                 if any(f in task_changes for f in ["outputs", "run_hash"]):
                     new_outputs = task_data.get("outputs", {})
                     new_run_hash = task_data.get("run_hash", None)
+
                     await update_node_outputs(
                         app,
                         the_project_owner,

services/web/server/src/simcore_service_webserver/projects/projects_api.py

@@ -58,7 +58,12 @@
 from ..users_api import UserRole, get_user_name, get_user_role
 from ..users_exceptions import UserNotFoundError
 from . import _delete
-from .project_lock import UserNameDict, get_project_locked_state, lock_project
+from .project_lock import (
+    UserNameDict,
+    get_project_locked_state,
+    is_project_locked,
+    lock_project,
+)
 from .projects_db import APP_PROJECT_DBAPI, ProjectDBAPI
 from .projects_exceptions import NodeNotFoundError, ProjectLockError
 from .projects_utils import extract_dns_without_default_port
@@ -387,11 +392,23 @@ async def is_node_id_present_in_any_project_workbench(
     return await db.node_id_exists(node_id)
 
 
-async def trigger_connected_service_retrieve(
+async def _trigger_connected_service_retrieve(
     app: web.Application, project: dict, updated_node_uuid: str, changed_keys: list[str]
 ) -> None:
+    project_id = project["uuid"]
+    if await is_project_locked(app, project_id):
+        # NOTE: we log warn since this function is fire&forget and raise an exception would not be anybody to handle it
+        log.warning(
+            "Skipping service retrieval because project with %s is currently locked."
+            "Operation triggered by %s",
+            f"{project_id=}",
+            f"{changed_keys=}",
+        )
+        return
+
     workbench = project["workbench"]
     nodes_keys_to_update: dict[str, list[str]] = defaultdict(list)
+
     # find the nodes that need to retrieve data
     for node_uuid, node in workbench.items():
         # check this node is dynamic
@@ -425,7 +442,7 @@ async def post_trigger_connected_service_retrieve(
     app: web.Application, **kwargs
 ) -> None:
     await fire_and_forget_task(
-        trigger_connected_service_retrieve(app, **kwargs),
+        _trigger_connected_service_retrieve(app, **kwargs),
         task_suffix_name="trigger_connected_service_retrieve",
         fire_and_forget_tasks_collection=app[APP_FIRE_AND_FORGET_TASKS_KEY],
     )

services/web/server/src/simcore_service_webserver/projects/projects_nodes_utils.py

@@ -84,7 +84,7 @@ async def update_node_outputs(
         else list(ui_changed_keys | set(keys_changed))
     )
 
-    # notify
+    # fire&forget to notify connected nodes to retrieve its inputs **if necessary**
     await projects_api.post_trigger_connected_service_retrieve(
         app=app, project=project, updated_node_uuid=node_uuid, changed_keys=keys
     )