♻️✨adding nodeports support to dynamic-sidecar

.github/workflows/ci-testing-deploy.yml

@@ -1919,7 +1919,7 @@ jobs:
           path: codeclimate.${{ github.job }}_coverage.json
 
   integration-test-director-v2-02:
-    timeout-minutes: 20 # if this timeout gets too small, then split the tests
+    timeout-minutes: 30 # if this timeout gets too small, then split the tests
     name: "[int] director-v2 02"
     needs: [build-test-images]
     runs-on: ${{ matrix.os }}

api/specs/webserver/openapi-projects.yaml

@@ -395,6 +395,55 @@ paths:
         default:
           $ref: "./openapi.yaml#/components/responses/DefaultErrorResponse"
 
+  /projects/{project_id}/nodes/{node_id}/retrieve:
+    parameters:
+      - name: project_id
+        in: path
+        required: true
+        schema:
+          type: string
+      - name: node_id
+        in: path
+        required: true
+        schema:
+          type: string
+
+    post:
+      tags:
+        - project
+      description: Triggers service retrieve
+      operationId: post_retrieve
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                port_keys:
+                  description: list of por keys to be retrieved
+                  type: array
+                  items:
+                    type: string
+      responses:
+        "200":
+          description: Returns the amount of transferred bytes when pulling data via nodeports
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  data:
+                    type: object
+                    description: response payload
+                    properties:
+                      size_bytes:
+                        type: integer
+                        description: amount of transferred bytes
+
+        default:
+          $ref: "#/components/responses/DefaultErrorResponse"
+
   /projects/{study_uuid}/tags/{tag_id}:
     parameters:
       - name: tag_id

api/specs/webserver/openapi.yaml

@@ -200,6 +200,9 @@ paths:
   /projects/{project_id}/nodes/{node_id}:
     $ref: "./openapi-projects.yaml#/paths/~1projects~1{project_id}~1nodes~1{node_id}"
 
+  /projects/{project_id}/nodes/{node_id}:retrieve:
+    $ref: "./openapi-projects.yaml#/paths/~1projects~1{project_id}~1nodes~1{node_id}~1retrieve"
+
   /nodes/{nodeInstanceUUID}/outputUi/{outputKey}:
     $ref: "./openapi-node-v0.0.1.yaml#/paths/~1nodes~1{nodeInstanceUUID}~1outputUi~1{outputKey}"
 

ci/github/integration-testing/director-v2.bash

@@ -23,6 +23,7 @@ test() {
   echo "testing in services/director-v2/tests/integration/$1"
   pytest --cov=simcore_service_director_v2 --durations=10 --cov-append \
     --color=yes --cov-report=term-missing --cov-report=xml --cov-config=.coveragerc \
+    -vvv -s --log-cli-level=DEBUG \
     -v -m "not travis" "services/director-v2/tests/integration/$1" --log-level=DEBUG
 }
 

packages/pytest-simcore/src/pytest_simcore/docker_registry.py

@@ -230,52 +230,54 @@ def jupyter_service(docker_registry: str, node_meta_schema: Dict) -> Dict[str, s
     )
 
 
-DY_STATIC_FILE_SERVER_VERSION = "1.0.5"
+@pytest.fixture(scope="session", params=["2.0.2"])
+def dy_static_file_server_version(request):
+    return request.param
 
 
 @pytest.fixture(scope="session")
 def dy_static_file_server_service(
-    docker_registry: str, node_meta_schema: Dict
+    docker_registry: str, node_meta_schema: Dict, dy_static_file_server_version: str
 ) -> Dict[str, str]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server
     """
     return _pull_push_service(
         "itisfoundation/dy-static-file-server",
-        DY_STATIC_FILE_SERVER_VERSION,
+        dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )
 
 
 @pytest.fixture(scope="session")
 def dy_static_file_server_dynamic_sidecar_service(
-    docker_registry: str, node_meta_schema: Dict
+    docker_registry: str, node_meta_schema: Dict, dy_static_file_server_version: str
 ) -> Dict[str, str]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server-dynamic-sidecar
     """
     return _pull_push_service(
         "itisfoundation/dy-static-file-server-dynamic-sidecar",
-        DY_STATIC_FILE_SERVER_VERSION,
+        dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )
 
 
 @pytest.fixture(scope="session")
 def dy_static_file_server_dynamic_sidecar_compose_spec_service(
-    docker_registry: str, node_meta_schema: Dict
+    docker_registry: str, node_meta_schema: Dict, dy_static_file_server_version: str
 ) -> Dict[str, str]:
     """
     Adds the below service in docker registry
     itisfoundation/dy-static-file-server-dynamic-sidecar-compose-spec
     """
     return _pull_push_service(
         "itisfoundation/dy-static-file-server-dynamic-sidecar-compose-spec",
-        DY_STATIC_FILE_SERVER_VERSION,
+        dy_static_file_server_version,
         docker_registry,
         node_meta_schema,
     )

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

@@ -2,12 +2,25 @@
 import logging
 from collections import deque
 from functools import wraps
-from typing import Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Callable, Deque, Dict, List, Optional
 
 import attr
 
 logger = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    Queue = asyncio.Queue
+else:
+
+    class FakeGenericMeta(type):
+        def __getitem__(self, item):
+            return self
+
+    class Queue(
+        asyncio.Queue, metaclass=FakeGenericMeta
+    ):  # pylint: disable=function-redefined
+        pass
+
 
 @attr.s(auto_attribs=True)
 class Context:
@@ -30,7 +43,9 @@ async def stop_sequential_workers() -> None:
     logger.info("All run_sequentially_in_context pending workers stopped")
 
 
-def run_sequentially_in_context(target_args: List[str] = None):
+def run_sequentially_in_context(
+    target_args: List[str] = None,
+) -> Callable[[Any], Any]:
     """All request to function with same calling context will be run sequentially.
 
     Example:
@@ -68,15 +83,17 @@ async def func(param1, param2, param3):
     """
     target_args = [] if target_args is None else target_args
 
-    def internal(decorated_function):
-        def get_context(args, kwargs: Dict) -> Context:
+    def internal(
+        decorated_function: Callable[[Any], Optional[Any]]
+    ) -> Callable[[Any], Optional[Any]]:
+        def get_context(args: Any, kwargs: Dict[Any, Any]) -> Context:
             arg_names = decorated_function.__code__.co_varnames[
                 : decorated_function.__code__.co_argcount
             ]
             search_args = dict(zip(arg_names, args))
             search_args.update(kwargs)
 
-            key_parts = deque()
+            key_parts: Deque[str] = deque()
             for arg in target_args:
                 sub_args = arg.split(".")
                 main_arg = sub_args[0]
@@ -108,13 +125,13 @@ def get_context(args, kwargs: Dict) -> Context:
             return _sequential_jobs_contexts[key]
 
         @wraps(decorated_function)
-        async def wrapper(*args, **kwargs):
+        async def wrapper(*args: Any, **kwargs: Any) -> Any:
             context: Context = get_context(args, kwargs)
 
             if not context.initialized:
                 context.initialized = True
 
-                async def worker(in_q: asyncio.Queue, out_q: asyncio.Queue):
+                async def worker(in_q: Queue, out_q: Queue) -> None:
                     while True:
                         awaitable = await in_q.get()
                         in_q.task_done()
@@ -137,7 +154,7 @@ async def worker(in_q: asyncio.Queue, out_q: asyncio.Queue):
                     worker(context.in_queue, context.out_queue)
                 )
 
-            await context.in_queue.put(decorated_function(*args, **kwargs))
+            await context.in_queue.put(decorated_function(*args, **kwargs))  # type: ignore
 
             wrapped_result = await context.out_queue.get()
             if isinstance(wrapped_result, Exception):

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

@@ -1,5 +1,7 @@
+import asyncio
 from concurrent.futures import ProcessPoolExecutor
 from contextlib import contextmanager
+from typing import Any, Callable
 
 # only gets created on use and is guaranteed to be the s
 # ame for the entire lifetime of the application
@@ -35,3 +37,10 @@ def non_blocking_process_pool_executor(**kwargs) -> ProcessPoolExecutor:
         # FIXME: uncomment below line when the issue is fixed
         # executor.shutdown(wait=False)
         pass
+
+
+async def async_on_threadpool(callable_function: Callable, *args: Any) -> Any:
+    """Ensures blocking operation runs on shared thread pool"""
+    return await asyncio.get_event_loop().run_in_executor(
+        None, callable_function, *args
+    )

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

@@ -7,7 +7,6 @@
 import asyncio
 import logging
 import os
-
 from pathlib import Path
 from typing import Any, Awaitable, Coroutine, List, Optional, Union
 
@@ -79,8 +78,11 @@ def log_exception_callback(fut: asyncio.Future):
 
 # // tasks
 async def logged_gather(
-    *tasks, reraise: bool = True, log: logging.Logger = logger, max_concurrency: int = 0
-) -> List[Any]:
+    *tasks: Awaitable[Any],
+    reraise: bool = True,
+    log: logging.Logger = logger,
+    max_concurrency: int = 0,
+) -> List[Optional[Any]]:
     """
         Thin wrapper around asyncio.gather that allows excuting ALL tasks concurently until the end
         even if any of them fail. Finally, all errors are logged and the first raised (if reraise=True)
@@ -91,18 +93,15 @@ async def logged_gather(
         use directly asyncio.gather(*tasks, return_exceptions=True).
 
     :param reraise: reraises first exception (in order the tasks were passed) concurrent tasks, defaults to True
-    :type reraise: bool, optional
     :param log: passing the logger gives a chance to identify the origin of the gather call, defaults to current submodule's logger
-    :type log: logging.Logger, optional
     :return: list of tasks results and errors e.g. [1, 2, ValueError("task3 went wrong"), 33, "foo"]
-    :rtype: List[Any]
     """
 
     wrapped_tasks = tasks
     if max_concurrency > 0:
         semaphore = asyncio.Semaphore(max_concurrency)
 
-        async def sem_task(task):
+        async def sem_task(task: Awaitable[Any]) -> Any:
             async with semaphore:
                 return await task
 

packages/service-library/tests/test_pools.py

@@ -1,8 +1,7 @@
 from asyncio import BaseEventLoop
 from concurrent.futures import ProcessPoolExecutor
 
-
-from servicelib.pools import non_blocking_process_pool_executor
+from servicelib.pools import async_on_threadpool, non_blocking_process_pool_executor
 
 
 def return_int_one() -> int:
@@ -32,3 +31,7 @@ async def test_different_pool_instances() -> None:
         max_workers=1
     ) as first, non_blocking_process_pool_executor() as second:
         assert first != second
+
+
+async def test_run_on_thread_pool() -> None:
+    assert await async_on_threadpool(return_int_one) == 1

packages/simcore-sdk/requirements/_test.in

@@ -18,6 +18,7 @@ pytest-mock
 pytest-runner
 pytest-sugar
 pytest-xdist
+pytest-lazy-fixture
 
 # mockups/fixtures
 alembic

packages/simcore-sdk/requirements/_test.txt

@@ -131,6 +131,7 @@ pytest==6.2.5
     #   pytest-forked
     #   pytest-icdiff
     #   pytest-instafail
+    #   pytest-lazy-fixture
     #   pytest-mock
     #   pytest-sugar
     #   pytest-xdist
@@ -144,6 +145,8 @@ pytest-icdiff==0.5
     # via -r requirements/_test.in
 pytest-instafail==0.4.2
     # via -r requirements/_test.in
+pytest-lazy-fixture==0.6.3
+    # via -r requirements/_test.in
 pytest-mock==3.6.1
     # via -r requirements/_test.in
 pytest-runner==5.3.1

packages/simcore-sdk/src/simcore_sdk/node_data/data_manager.py

@@ -66,36 +66,50 @@ async def push(
         return await _push_file(user_id, project_id, node_uuid, archive_file_path, None)
 
 
-async def _pull_file(user_id: int, project_id: str, node_uuid: str, file_path: Path):
+async def _pull_file(
+    user_id: int,
+    project_id: str,
+    node_uuid: str,
+    file_path: Path,
+    save_to: Optional[Path] = None,
+):
+    destination_path = file_path if save_to is None else save_to
     s3_object = _create_s3_object(project_id, node_uuid, file_path)
     log.info("pulling data from %s to %s...", s3_object, file_path)
     downloaded_file = await filemanager.download_file_from_s3(
         user_id=user_id,
         store_id="0",
         s3_object=s3_object,
-        local_folder=file_path.parent,
+        local_folder=destination_path.parent,
     )
-    if downloaded_file != file_path:
-        if file_path.exists():
-            file_path.unlink()
-        move(f"{downloaded_file}", file_path)
-    log.info("%s successfuly pulled", file_path)
+    if downloaded_file != destination_path:
+        destination_path.unlink(missing_ok=True)
+        move(f"{downloaded_file}", destination_path)
+    log.info("%s successfuly pulled", destination_path)
 
 
 def _get_archive_name(path: Path) -> str:
     return f"{path.stem}.zip"
 
 
-async def pull(user_id: int, project_id: str, node_uuid: str, file_or_folder: Path):
+async def pull(
+    user_id: int,
+    project_id: str,
+    node_uuid: str,
+    file_or_folder: Path,
+    save_to: Optional[Path] = None,
+):
     if file_or_folder.is_file():
-        return await _pull_file(user_id, project_id, node_uuid, file_or_folder)
+        return await _pull_file(user_id, project_id, node_uuid, file_or_folder, save_to)
     # we have a folder, so we need somewhere to extract it to
     with TemporaryDirectory() as tmp_dir_name:
         archive_file = Path(tmp_dir_name) / _get_archive_name(file_or_folder)
         await _pull_file(user_id, project_id, node_uuid, archive_file)
         log.info("extracting data from %s", archive_file)
+
+        destination_folder = file_or_folder if save_to is None else save_to
         await unarchive_dir(
-            archive_to_extract=archive_file, destination_folder=file_or_folder
+            archive_to_extract=archive_file, destination_folder=destination_folder
         )
         log.info("extraction completed")