Fix/updated pipeline (#983)

Unnecessary updates of the pipeline db due to project updates would stop execution of pipeline.

- FIX: Updates pipeline ONLY when project's DAG topology has changed
- FIX: right access to maintenance service
- NEW for dev: adds .vscode-template/launch.json template to remote debugging of webserver and storage services

Author: pcrespov

.vscode-template/launch.json

@@ -0,0 +1,34 @@
+{
+  // Use IntelliSense to learn about possible attributes.
+  // Hover to view descriptions of existing attributes.
+  // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+  "version": "0.2.0",
+  "configurations": [
+    {
+      "name": "Python: Remote Attach webserver",
+      "type": "python",
+      "request": "attach",
+      "port": 3001,
+      "host": "127.0.0.1",
+      "pathMappings": [
+        {
+          "localRoot": "${workspaceFolder}/services/web/server",
+          "remoteRoot": "/devel/services/web/server"
+        }
+      ]
+    },
+    {
+      "name": "Python: Remote Attach storage",
+      "type": "python",
+      "request": "attach",
+      "port": 3003,
+      "host": "127.0.0.1",
+      "pathMappings": [
+        {
+          "localRoot": "${workspaceFolder}/services/storage",
+          "remoteRoot": "/devel/services/storage"
+        }
+      ]
+    }
+  ]
+}

.vscode-template/pytest.ini

@@ -25,7 +25,7 @@ RUN pip --no-cache install --upgrade pip wheel setuptools && \
 COPY --chown=jovyan:users services/maintenance/work/ $HOME/work
 
 WORKDIR $HOME/work
-RUN jupyter trust notebooks/*
+RUN jupyter trust notebooks/*; mkdir data
 
 
 USER root

services/maintenance/Dockerfile

@@ -1 +1,2 @@
 docker
+aiopg[sa]

services/maintenance/requirements.txt

@@ -9,19 +9,21 @@
 import sqlalchemy as sa
 from aiohttp import web, web_exceptions
 from aiopg.sa import Engine
+from sqlalchemy import and_
+
 from servicelib.application_keys import APP_DB_ENGINE_KEY
 from simcore_director_sdk.rest import ApiException
 from simcore_postgres_database.webserver_models import (comp_pipeline,
                                                         comp_tasks)
-from sqlalchemy import and_
 
 from .director import director_sdk
 
 log = logging.getLogger(__file__)
 logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
 
 
-async def _get_node_details(node_key:str, node_version:str, app: web.Application)->dict:
+
+async def _get_node_details(node_key:str, node_version:str, app: web.Application) -> Dict:
     if "file-picker" in node_key:
         # create a fake file-picker schema here!!
         fake_node_details = {"inputs":{},
@@ -59,7 +61,7 @@ async def _get_node_details(node_key:str, node_version:str, app: web.Application
         log.exception("Error could not find service %s:%s", node_key, node_version)
         raise web_exceptions.HTTPNotFound(reason=str(err))
 
-async def _build_adjacency_list(node_uuid:str, node_schema:dict, node_inputs:dict, pipeline_data:dict, dag_adjacency_list:dict, app: web.Application)->dict: # pylint: disable=too-many-arguments
+async def _build_adjacency_list(node_uuid:str, node_schema:Dict, node_inputs:Dict, pipeline_data:Dict, dag_adjacency_list:Dict, app: web.Application)->Dict: # pylint: disable=too-many-arguments
     if node_inputs is None or node_schema is None:
         return dag_adjacency_list
 
@@ -89,7 +91,7 @@ async def _build_adjacency_list(node_uuid:str, node_schema:dict, node_inputs:dic
                     dag_adjacency_list[input_node_uuid].append(node_uuid)
     return dag_adjacency_list
 
-async def _parse_pipeline(pipeline_data:dict, app: web.Application): # pylint: disable=R0912
+async def _parse_pipeline(pipeline_data:Dict, app: web.Application): # pylint: disable=R0912
     dag_adjacency_list = dict()
     tasks = dict()
 
@@ -139,29 +141,31 @@ async def _parse_pipeline(pipeline_data:dict, app: web.Application): # pylint: d
     return dag_adjacency_list, tasks
 
 async def _set_adjacency_in_pipeline_db(db_engine: Engine, project_id: str, dag_adjacency_list: Dict):
+    query = sa.select([comp_pipeline]).\
+                where(comp_pipeline.c.project_id==project_id)
+
     async with db_engine.acquire() as conn:
-        query = sa.select([comp_pipeline]).\
-                    where(comp_pipeline.c.project_id==project_id)
         result = await conn.execute(query)
         pipeline = await result.first()
 
-        if pipeline is None:
-            # pylint: disable=no-value-for-parameter
-            # let's create one then
-            query = comp_pipeline.insert().\
-                    values(project_id=project_id,
-                            dag_adjacency_list=dag_adjacency_list,
-                            state=0)
-
-            log.debug("Pipeline object created")
-        else:
-            # let's modify it
-            log.debug("Pipeline object found")
-            #pylint: disable=no-value-for-parameter
-            query = comp_pipeline.update().\
-                        where(comp_pipeline.c.project_id == project_id).\
-                        values(state=0,
-                                dag_adjacency_list=dag_adjacency_list)
+    if pipeline is None:
+        # pylint: disable=no-value-for-parameter
+        # let's create one then
+        query = comp_pipeline.insert().\
+                values(project_id=project_id,
+                        dag_adjacency_list=dag_adjacency_list,
+                        state=0)
+        log.debug("Pipeline object created")
+    else:
+        # let's modify it
+        log.debug("Pipeline object found")
+        #pylint: disable=no-value-for-parameter
+        query = comp_pipeline.update().\
+                    where(comp_pipeline.c.project_id == project_id).\
+                    values(state=0,
+                            dag_adjacency_list=dag_adjacency_list)
+
+    async with db_engine.acquire() as conn:
         await conn.execute(query)
 
 async def _set_tasks_in_tasks_db(db_engine: Engine, project_id: str, tasks: Dict):
@@ -219,10 +223,11 @@ async def _set_tasks_in_tasks_db(db_engine: Engine, project_id: str, tasks: Dict
 async def update_pipeline_db(app: web.Application, project_id, pipeline_data):
     db_engine = app[APP_DB_ENGINE_KEY]
 
-    log.debug("Client calls update_pipeline with project id: %s, pipeline data %s", project_id, pipeline_data)
+    log.info("Pipeline has been updated for project %s", project_id)
+    log.debug("Updating pipeline: %s", pformat(pipeline_data))
     dag_adjacency_list, tasks = await _parse_pipeline(pipeline_data, app)
 
-    log.debug("Pipeline parsed:\nlist: %s\ntasks: %s", str(dag_adjacency_list), str(tasks))
+    log.debug("Pipeline parsed:\nlist: %s\ntasks: %s", pformat(dag_adjacency_list), pformat(tasks))
     await _set_adjacency_in_pipeline_db(db_engine, project_id, dag_adjacency_list)
     await _set_tasks_in_tasks_db(db_engine, project_id, tasks)
 

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

@@ -33,15 +33,15 @@
 DB_EXCLUSIVE_COLUMNS = ["type", "id", "published"]
 
 # TODO: check here how schema to model db works!?
-def _convert_to_db_names(project_data: Dict) -> Dict:
+def _convert_to_db_names(project_document_data: Dict) -> Dict:
     converted_args = {}
-    for key, value in project_data.items():
+    for key, value in project_document_data.items():
         converted_args[ChangeCase.camel_to_snake(key)] = value
     return converted_args
 
-def _convert_to_schema_names(project_db_data: Mapping) -> Dict:
+def _convert_to_schema_names(project_database_data: Mapping) -> Dict:
     converted_args = {}
-    for key, value in project_db_data.items():
+    for key, value in project_database_data.items():
         if key in DB_EXCLUSIVE_COLUMNS:
             continue
         converted_value = value
@@ -272,6 +272,18 @@ async def get_template_project(self, project_uuid: str, *, only_published=False)
 
         return template_prj
 
+    async def get_project_workbench(self, project_uuid: str):
+        async with self.engine.acquire() as conn:
+            query = select([projects.c.workbench]).where(
+                    projects.c.uuid == project_uuid
+                    )
+            result = await conn.execute(query)
+            row = await result.first()
+            if row:
+                return row[projects.c.workbench]
+        return {}
+
+
     async def update_user_project(self, project_data: Dict, user_id: str, project_uuid: str):
         """ updates a project from a user
 

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

@@ -12,6 +12,7 @@
 from ..login.decorators import RQT_USERID_KEY, login_required
 from ..security_api import check_permission
 from ..storage_api import delete_data_folders_of_project
+from .projects_utils import is_graph_equal
 from .projects_api import validate_project
 from .projects_db import APP_PROJECT_DBAPI
 from .projects_exceptions import (ProjectInvalidRightsError,
@@ -79,7 +80,7 @@ async def create_projects(request: web.Request):
         # update metadata (uuid, timestamps, ownership) and save
         await db.add_project(project, user_id, force_as_template=as_template is not None)
 
-        # Every change in projects workbench needs to be reflected in the pipeline db
+        # This is a new project and every new graph needs to be reflected in the pipeline db
         await update_pipeline_db(request.app, project["uuid"], project["workbench"])
 
     except ValidationError:
@@ -186,10 +187,12 @@ async def replace_project(request: web.Request):
     try:
         validate_project(request.app, new_project)
 
+        previous_workbench = await db.get_project_workbench(project_uuid)
         await db.update_user_project(new_project, user_id, project_uuid)
 
-        # Every change in projects workbench needs to be reflected in the pipeline db
-        await update_pipeline_db(request.app, project_uuid, new_project["workbench"])
+        if not is_graph_equal(new_project["workbench"], previous_workbench):
+            # Every change in the pipeline workflow needs to be reflected in the pipeline db
+            await update_pipeline_db(request.app, project_uuid, new_project["workbench"])
 
     except ValidationError:
         raise web.HTTPBadRequest

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

@@ -82,3 +82,29 @@ def _normalize_value(s):
         inputs.update(new_inputs)
 
     return project
+
+
+def is_graph_equal(lhs_workbench: Dict, rhs_workbench: Dict) -> bool:
+    """ Checks whether both workbench contain the same graph
+
+        Two graphs are the same when the same topology (i.e. nodes and edges)
+        and the ports at each node have same values/connections
+    """
+    try:
+        assert set(rhs_workbench.keys()) == set(lhs_workbench.keys())
+        for node_id, node in rhs_workbench.items():
+            # same nodes
+            assert all(node.get(k) == lhs_workbench[node_id].get(k)
+                for k in ['key', 'version']
+            )
+
+            # same connectivity (edges)
+            assert set(node.get('inputNodes')) == set(lhs_workbench[node_id].get('inputNodes'))
+
+            # same input values
+            for port_id, port in node.get("inputs", {}).items():
+                assert port == lhs_workbench[node_id].get("inputs", {}).get(port_id)
+
+    except (AssertionError, TypeError, AttributeError):
+        return False
+    return True

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

@@ -144,7 +144,8 @@ async def access_study(request: web.Request) -> web.Response:
 
     template_project = await get_public_project(request.app, project_id)
     if not template_project:
-        raise web.HTTPNotFound(reason=f"Invalid public study [{project_id}]")
+        raise web.HTTPNotFound(reason=f"Requested study ({project_id}) has not been published.\
+             Please contact the data curators for more information.")
 
     user = None
     is_anonymous_user = await is_anonymous(request)