Computational services can have an interactive service as input (#1825)

* removed config.DEBUG which caused issues

* Added some logging
- in case the services failed to start
- in debug mode

* sidecar properly unzipes input if zipfile

* changed the recrate into a reset status

* removing input only if zip

* log collection is now protected by a lock

* trying to fix logs recovery

* adding comments

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

services/director/src/simcore_service_director/producer.py

@@ -264,9 +264,8 @@ async def _create_docker_service_params(
             docker_params["labels"]["port"] = docker_params["labels"][
                 f"traefik.http.services.{service_name}.loadbalancer.server.port"
             ] = str(param["value"])
-        elif (
-            param["type"] == "EndpointSpec"
-        ):  # REST-API compatible
+        # REST-API compatible
+        elif param["type"] == "EndpointSpec":
             if "Ports" in param["value"]:
                 if (
                     isinstance(param["value"]["Ports"], list)
@@ -803,7 +802,11 @@ async def start_service(
         node_details = containers_meta_data[0]
         if config.MONITORING_ENABLED:
             service_started(
-                app, user_id, service_key, service_tag, "DYNAMIC",
+                app,
+                user_id,
+                service_key,
+                service_tag,
+                "DYNAMIC",
             )
         # we return only the info of the main service
         return node_details

services/sidecar/src/simcore_service_sidecar/executor.py

@@ -5,6 +5,8 @@
 import time
 from pathlib import Path
 from typing import Dict, Optional
+import zipfile
+import os
 
 import aiopg
 import attr
@@ -164,6 +166,12 @@ async def _process_task_input(self, port: node_ports.Port, input_ports: Dict):
                     final_path,
                     path,
                 )
+                # check if the file is a zip, in that case extract all
+                if zipfile.is_zipfile(final_path):
+                    with zipfile.ZipFile(final_path, "r") as zip_obj:
+                        zip_obj.extractall(final_path.parents[0])
+                    # finally remove the zip archive
+                    os.remove(final_path)
             else:
                 input_ports[port.key] = port_value
         else:

services/sidecar/tests/integration/test_sidecar.py

@@ -5,16 +5,17 @@
 import asyncio
 import inspect
 import json
+from collections import deque
 from pathlib import Path
 from typing import Any, Dict, List, Tuple
 from uuid import uuid4
 
 import aio_pika
 import pytest
 import sqlalchemy as sa
+from simcore_sdk.models.pipeline_models import ComputationalPipeline, ComputationalTask
 from yarl import URL
 
-from simcore_sdk.models.pipeline_models import ComputationalPipeline, ComputationalTask
 from simcore_service_sidecar import config, utils
 
 SIMCORE_S3_ID = 0
@@ -80,9 +81,29 @@ def sidecar_config(
     config.RABBIT_CONFIG = rabbit_config
 
 
-def _assert_incoming_data_logs(
+class LockedCollector:
+    __slots__ = ("_lock", "_list")
+
+    def __init__(self):
+        self._lock = asyncio.Lock()
+        self._list = deque()
+
+    async def is_empty(self):
+        async with self._lock:
+            return len(self._list) == 0
+
+    async def append(self, item):
+        async with self._lock:
+            self._list.append(item)
+
+    async def as_list(self) -> List:
+        async with self._lock:
+            return list(self._list)
+
+
+async def _assert_incoming_data_logs(
     tasks: List[str],
-    incoming_data: List[Dict[str, str]],
+    incoming_data: LockedCollector,
     user_id: int,
     project_id: str,
     service_repo: str,
@@ -94,7 +115,7 @@ def _assert_incoming_data_logs(
     tasks_logs = {task: [] for task in tasks}
     progress_logs = {task: [] for task in tasks}
     instrumentation_messages = {task: [] for task in tasks}
-    for message in incoming_data:
+    for message in await incoming_data.as_list():
         if "metrics" in message:
             # instrumentation message
             instrumentation_messages[message["service_uuid"]].append(message)
@@ -338,13 +359,14 @@ async def test_run_services(
     :param osparc_service: Fixture defined in pytest-simcore.docker_registry. Uses parameters service_repo, service_tag
     :type osparc_service: Dict[str, str]
     """
-    incoming_data = []
+    incoming_data = LockedCollector()
 
     async def rabbit_message_handler(message: aio_pika.IncomingMessage):
-        data = json.loads(message.body)
-        incoming_data.append(data)
+        async with message.process():
+            data = json.loads(message.body)
+            await incoming_data.append(data)
 
-    await rabbit_queue.consume(rabbit_message_handler, exclusive=True, no_ack=True)
+    await rabbit_queue.consume(rabbit_message_handler, exclusive=True)
 
     job_id = 1
 
@@ -353,7 +375,7 @@ async def rabbit_message_handler(message: aio_pika.IncomingMessage):
     # runs None first
     next_task_nodes = await cli.run_sidecar(job_id, user_id, pipeline.project_id, None)
     await asyncio.sleep(5)
-    assert not incoming_data
+    assert await incoming_data.is_empty()
     assert next_task_nodes
     assert len(next_task_nodes) == 1
     assert next_task_nodes[0] == next(iter(pipeline_cfg))
@@ -370,7 +392,7 @@ async def rabbit_message_handler(message: aio_pika.IncomingMessage):
         dag.extend(pipeline_cfg[key]["next"])
     assert next_task_nodes == dag
     await asyncio.sleep(5)  # wait a little bit for logs to come in
-    _assert_incoming_data_logs(
+    await _assert_incoming_data_logs(
         list(pipeline_cfg.keys()),
         incoming_data,
         user_id,

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

@@ -262,6 +262,8 @@ async def _set_adjacency_in_pipeline_db(
 async def _set_tasks_in_tasks_db(
     db_engine: Engine, project_id: str, tasks: Dict[str, Dict], replace_pipeline=True
 ):
+    """The replace_pipeline is missleading, it should be interpreted
+    as the "RUN" button was pressed on the UI."""
     # pylint: disable=no-value-for-parameter
 
     async def _task_already_exists(
@@ -328,14 +330,24 @@ async def _update_task(
             result = await conn.execute(query)
             tasks_rows = await result.fetchall()
 
-            # prune database from invalid tasks
+            # no longer prune database from invalid tasks
+            # mark comp tasks with job_id == NULL and set status to 0
+            # effectively marking a rest of the pipeline without loosing
+            # inputs from comp services
             for task_row in tasks_rows:
+                # for some reason the outputs are not present in the 
+                # tasks outputs. copy them over (will be used below)
+                tasks[task_row.node_id]["outputs"] = task_row.outputs
                 if not task_row.node_id in tasks:
-                    query = comp_tasks.delete().where(
-                        and_(
-                            comp_tasks.c.project_id == project_id,
-                            comp_tasks.c.node_id == task_row.node_id,
+                    query = (
+                        comp_tasks.update()
+                        .where(
+                            and_(
+                                comp_tasks.c.project_id == project_id,
+                                comp_tasks.c.node_id == task_row.node_id,
+                            )
                         )
+                        .values(job_id=None, state=UNKNOWN)
                     )
                     await conn.execute(query)
 
@@ -405,7 +417,7 @@ async def update_pipeline_db(
     project_data: Dict,
     replace_pipeline: bool = True,
 ) -> None:
-    """ Updates entries in comp_pipeline and comp_task pg tables for a given project
+    """Updates entries in comp_pipeline and comp_task pg tables for a given project
 
     :param replace_pipeline: Fully replaces instead of partial updates of existing entries, defaults to True
     """

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

@@ -125,6 +125,8 @@ async def start_project_interactive_services(
     running_services = await director_api.get_running_interactive_services(
         request.app, user_id, project["uuid"]
     )
+    log.debug("Running services %s", running_services)
+
     running_service_uuids = [x["service_uuid"] for x in running_services]
     # now start them if needed
     project_needed_services = {
@@ -133,6 +135,7 @@ async def start_project_interactive_services(
         if _is_node_dynamic(service["key"])
         and service_uuid not in running_service_uuids
     }
+    log.debug("Services to start %s", project_needed_services)
 
     start_service_tasks = [
         director_api.start_service(
@@ -145,7 +148,18 @@ async def start_project_interactive_services(
         )
         for service_uuid, service in project_needed_services.items()
     ]
-    await logged_gather(*start_service_tasks, reraise=True)
+
+    result = await logged_gather(*start_service_tasks, reraise=True)
+    log.debug("Services start result %s", result)
+    for entry in result:
+        # if the status is present in the results fo the start_service
+        # it means that the API call failed
+        # also it is enforced that the status is different from 200 OK
+        if "status" not in entry:
+            continue
+
+        if entry["status"] != 200:
+            log.error("Error while starting dynamic service %s", entry)
 
 
 async def delete_project(request: web.Request, project_uuid: str, user_id: int) -> None: