🐛 Bugfix/clean storage issue (ITISFoundation#2421)

* fix unbound variable
* increase timeout for sync operation
* fix issue with removed pipeline

Author: sanderegg

packages/service-library/src/servicelib/monitoring.py

@@ -9,6 +9,7 @@
     - TODO: see https://github.com/claws/aioprometheus
 """
 
+import asyncio
 import logging
 import time
 
@@ -31,6 +32,7 @@ def middleware_factory(app_name):
     @web.middleware
     async def middleware_handler(request: web.Request, handler):
         # See https://prometheus.io/docs/concepts/metric_types
+        resp = None
         try:
             request["start_time"] = time.time()
             request.app["REQUEST_IN_PROGRESS"].labels(
@@ -43,10 +45,12 @@ async def middleware_handler(request: web.Request, handler):
             # Captures raised reponses (success/failures accounted with resp.status)
             resp = exc
             raise
-        except Exception as exc:  # pylint: disable=broad-except
+        except asyncio.CancelledError as exc:
+            # python 3.8 cancellederror is a subclass of BaseException and NOT Exception
+            resp = web.HTTPRequestTimeout(reason=str(exc))
+        except BaseException as exc:  # pylint: disable=broad-except
             # Prevents issue #1025.
             resp = web.HTTPInternalServerError(reason=str(exc))
-            resp_time = time.time() - request["start_time"]
 
             # NOTE: all access to API (i.e. and not other paths as /socket, /x, etc) shall return web.HTTPErrors since processed by error_middleware_factory
             log.exception(
@@ -55,9 +59,10 @@ async def middleware_handler(request: web.Request, handler):
                 request.remote,
                 request.method,
                 request.path,
-                resp_time,
+                time.time() - request["start_time"],
                 resp.status,
             )
+
         finally:
             # metrics on the same request
             resp_time = time.time() - request["start_time"]
@@ -69,9 +74,10 @@ async def middleware_handler(request: web.Request, handler):
                 app_name, request.path, request.method
             ).dec()
 
-            request.app["REQUEST_COUNT"].labels(
-                app_name, request.method, request.path, resp.status
-            ).inc()
+            if resp:
+                request.app["REQUEST_COUNT"].labels(
+                    app_name, request.method, request.path, resp.status
+                ).inc()
 
         return resp
 

services/director-v2/requirements/_base.in

@@ -4,9 +4,9 @@
 # NOTE: ALL version constraints MUST be commented
 --constraint ../../../requirements/constraints.txt
 
---requirements ../../../packages/models-library/requirements/_base.in
---requirements ../../../packages/settings-library/requirements/_base.in
---requirements ../../../packages/postgres-database/requirements/_base.in
+--requirement ../../../packages/models-library/requirements/_base.in
+--requirement ../../../packages/settings-library/requirements/_base.in
+--requirement ../../../packages/postgres-database/requirements/_base.in
 
 # fastapi and extensions
 fastapi[all]

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

@@ -32,6 +32,7 @@
 from ..models.schemas.constants import UserID
 from ..modules.celery import CeleryClient, CeleryTaskIn
 from ..utils.computations import get_pipeline_state_from_task_states
+from ..utils.exceptions import PipelineNotFoundError
 from .db.repositories import BaseRepository
 from .db.repositories.comp_pipelines import CompPipelinesRepository
 from .db.repositories.comp_runs import CompRunsRepository
@@ -204,24 +205,33 @@ async def _schedule_pipeline(
             user_id,
         )
 
-        pipeline_dag: nx.DiGraph = await self._get_pipeline_dag(project_id)
-        pipeline_tasks: Dict[str, CompTaskAtDB] = await self._get_pipeline_tasks(
-            project_id, pipeline_dag
-        )
+        pipeline_dag = nx.DiGraph()
+        pipeline_tasks: Dict[str, CompTaskAtDB] = {}
+        pipeline_result: RunningState = RunningState.UNKNOWN
+        try:
+            pipeline_dag = await self._get_pipeline_dag(project_id)
+            pipeline_tasks: Dict[str, CompTaskAtDB] = await self._get_pipeline_tasks(
+                project_id, pipeline_dag
+            )
 
-        # filter out the tasks with what were already completed
-        pipeline_dag.remove_nodes_from(
-            {
-                node_id
-                for node_id, t in pipeline_tasks.items()
-                if t.state in _COMPLETED_STATES
-            }
-        )
+            # filter out the tasks with what were already completed
+            pipeline_dag.remove_nodes_from(
+                {
+                    node_id
+                    for node_id, t in pipeline_tasks.items()
+                    if t.state in _COMPLETED_STATES
+                }
+            )
 
-        # update the current status of the run
-        pipeline_result: RunningState = await self._update_run_result(
-            user_id, project_id, iteration, pipeline_tasks
-        )
+            # update the current status of the run
+            pipeline_result = await self._update_run_result(
+                user_id, project_id, iteration, pipeline_tasks
+            )
+        except PipelineNotFoundError:
+            logger.warning(
+                "pipeline %s does not exist in comp_pipeline table, it will be removed from scheduler",
+                project_id,
+            )
 
         if not pipeline_dag.nodes():
             # there is nothing left, the run is completed, we're done here

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

@@ -6,7 +6,7 @@
 import urllib
 from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
-from aiohttp import web
+from aiohttp import ClientTimeout, web
 from servicelib.request_keys import RQT_USERID_KEY
 from servicelib.rest_responses import unwrap_envelope
 from servicelib.rest_utils import extract_and_validate
@@ -46,7 +46,7 @@ def _resolve_storage_url(request: web.Request) -> URL:
     return url
 
 
-async def _request_storage(request: web.Request, method: str):
+async def _request_storage(request: web.Request, method: str, **kwargs):
     await extract_and_validate(request)
 
     url = _resolve_storage_url(request)
@@ -57,7 +57,9 @@ async def _request_storage(request: web.Request, method: str):
         body = await request.json()
 
     session = get_client_session(request.app)
-    async with session.request(method.upper(), url, ssl=False, json=body) as resp:
+    async with session.request(
+        method.upper(), url, ssl=False, json=body, **kwargs
+    ) as resp:
         payload = await resp.json()
         return payload
 
@@ -155,7 +157,7 @@ async def delete_file(request: web.Request):
 @login_required
 @permission_required("storage.files.sync")
 async def synchronise_meta_data_table(request: web.Request):
-    payload = await _request_storage(request, "POST")
+    payload = await _request_storage(request, "POST", timeout=ClientTimeout(total=300))
     return payload