🐛 workspace.zip disappears while uploading (ITISFoundation#3306)

sanderegg · web-flow · commit 9a8076d55b38 · 2022-09-02T11:23:12.000+02:00
diff --git a/packages/simcore-sdk/src/simcore_sdk/node_data/data_manager.py b/packages/simcore-sdk/src/simcore_sdk/node_data/data_manager.py
@@ -7,6 +7,7 @@
 from models_library.projects_nodes_io import StorageFileID
 from pydantic import parse_obj_as
 from servicelib.archiving_utils import archive_dir, unarchive_dir
+from servicelib.logging_utils import log_catch, log_context
 from settings_library.r_clone import RCloneSettings
 from simcore_sdk.node_ports_common.constants import SIMCORE_LOCATION
 
@@ -70,8 +71,9 @@ async def push(
             io_log_redirect_cb=io_log_redirect_cb,
         )
     # we have a folder, so we create a compressed file
-    with TemporaryDirectory() as tmp_dir_name:
-        log.info("compressing %s into %s...", file_or_folder.name, tmp_dir_name)
+    with log_catch(log), log_context(
+        log, logging.INFO, "pushing %s", file_or_folder
+    ), TemporaryDirectory() as tmp_dir_name:
         # compress the files
         archive_file_path = (
             Path(tmp_dir_name) / f"{rename_to or file_or_folder.stem}.zip"
diff --git a/services/director-v2/tests/integration/02/test_dynamic_sidecar_nodeports_integration.py b/services/director-v2/tests/integration/02/test_dynamic_sidecar_nodeports_integration.py
@@ -9,7 +9,6 @@
 from collections import namedtuple
 from itertools import tee
 from pathlib import Path
-from pprint import pformat
 from typing import Any, AsyncIterable, Callable, Iterable, Iterator, cast
 from uuid import uuid4
 
@@ -60,8 +59,11 @@
 )
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from starlette import status
+from tenacity._asyncio import AsyncRetrying
+from tenacity.retry import retry_if_exception_type
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_fixed
 from utils import (
-    SEPARATOR,
     assert_all_services_running,
     assert_retrieve_service,
     assert_services_reply_200,
@@ -680,7 +682,7 @@ async def _wait_for_dy_services_to_fully_stop(
     to_observe = (
         director_v2_client._transport.app.state.dynamic_sidecar_scheduler._to_observe
     )
-
+    # TODO: ANE please use tenacity
     for i in range(TIMEOUT_DETECT_DYNAMIC_SERVICES_STOPPED):
         print(
             f"Sleeping for {i+1}/{TIMEOUT_DETECT_DYNAMIC_SERVICES_STOPPED} "
@@ -727,27 +729,6 @@ def _relative_path(root_path: Path, full_path: Path) -> Path:
     }
 
 
-LINE_PARTS_TO_MATCH = [
-    (0, "INFO:simcore_service_dynamic_sidecar.modules.nodeports:Uploaded"),
-    (2, "bytes"),
-    (3, "in"),
-    (5, "seconds"),
-]
-
-
-def _is_matching_line_in_logs(logs: list[str]) -> bool:
-    for line in logs:
-        if LINE_PARTS_TO_MATCH[0][1] in line:
-            print("".join(logs))
-
-            line_parts = line.strip().split(" ")
-            for position, value in LINE_PARTS_TO_MATCH:
-                assert line_parts[position] == value
-
-            return True
-    return False
-
-
 async def _print_dynamic_sidecars_containers_logs_and_get_containers(
     dynamic_services_urls: dict[str, str]
 ) -> list[str]:
@@ -778,17 +759,9 @@ async def _print_dynamic_sidecars_containers_logs_and_get_containers(
     return containers_names
 
 
-async def _print_container_inspect(container_id: str) -> None:
-    async with aiodocker.Docker() as docker_client:
-        container = await docker_client.containers.get(container_id)
-        container_inspect = await container.show()
-        print(f"Container {container_id} inspect:\n{pformat(container_inspect)}")
-
-
-async def _print_all_docker_volumes() -> None:
-    async with aiodocker.Docker() as docker_client:
-        docker_volumes = await docker_client.volumes.list()
-        print(f"Detected volumes:\n{pformat(docker_volumes)}")
+_CONTROL_TESTMARK_DY_SIDECAR_NODEPORT_UPLOADED_MESSAGE = (
+    "TEST: test_nodeports_integration DO NOT REMOVE"
+)
 
 
 async def _assert_retrieve_completed(
@@ -806,50 +779,24 @@ async def _assert_retrieve_completed(
     # look at dynamic-sidecar's logs to be sure when nodeports
     # have been uploaded
     async with aiodocker.Docker() as docker_client:
-        container: DockerContainer = await docker_client.containers.get(container_id)
-
-        for i in range(TIMEOUT_OUTPUTS_UPLOAD_FINISH_DETECTED):
-            logs = await container.log(stdout=True, stderr=True)
-
-            if _is_matching_line_in_logs(logs):
-                break
-
-            if i == TIMEOUT_OUTPUTS_UPLOAD_FINISH_DETECTED - 1:
-                print(SEPARATOR)
-                print(f"Dumping information for service_uuid={service_uuid}")
-                print(SEPARATOR)
-
-                print("".join(logs))
-                print(SEPARATOR)
-
-                containers_names = (
-                    await _print_dynamic_sidecars_containers_logs_and_get_containers(
-                        dynamic_services_urls
-                    )
+        async for attempt in AsyncRetrying(
+            reraise=True,
+            retry=retry_if_exception_type(AssertionError),
+            stop=stop_after_delay(TIMEOUT_OUTPUTS_UPLOAD_FINISH_DETECTED),
+            wait=wait_fixed(0.5),
+        ):
+            with attempt:
+                print(
+                    f"--> checking container logs of {service_uuid=}, [attempt {attempt.retry_state.attempt_number}]..."
+                )
+                container: DockerContainer = await docker_client.containers.get(
+                    container_id
                 )
-                print(SEPARATOR)
-
-                # inspect dynamic-sidecar container
-                await _print_container_inspect(container_id=container_id)
-                print(SEPARATOR)
-
-                # inspect spawned container
-                for container_name in containers_names:
-                    await _print_container_inspect(container_id=container_name)
-                    print(SEPARATOR)
-
-                await _print_all_docker_volumes()
-                print(SEPARATOR)
-
-                assert False, "Timeout reached"
-
-            print(
-                f"Sleeping {i+1}/{TIMEOUT_OUTPUTS_UPLOAD_FINISH_DETECTED} "
-                f"before searching logs from {service_uuid} again"
-            )
-            await asyncio.sleep(1)
 
-        print(f"Nodeports outputs upload finish detected for {service_uuid}")
+                logs = " ".join(await container.log(stdout=True, stderr=True))
+                assert (
+                    _CONTROL_TESTMARK_DY_SIDECAR_NODEPORT_UPLOADED_MESSAGE in logs
+                ), "TIP: Message missing suggests that the data was never uploaded: look in services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/nodeports.py"
 
 
 # TESTS
@@ -966,21 +913,12 @@ async def test_nodeports_integration(
         dynamic_services_urls
     )
 
-    await _assert_retrieve_completed(
-        director_v2_client=async_client,
-        service_uuid=services_node_uuids.dy,
-        dynamic_services_urls=dynamic_services_urls,
-    )
-
-    # NOTE: Waits a bit for the DB to write the changes in
-    # comp_task for the upstream service.
-    await asyncio.sleep(2)
-
-    await _assert_retrieve_completed(
-        director_v2_client=async_client,
-        service_uuid=services_node_uuids.dy_compose_spec,
-        dynamic_services_urls=dynamic_services_urls,
-    )
+    for service_uuid in (services_node_uuids.dy, services_node_uuids.dy_compose_spec):
+        await _assert_retrieve_completed(
+            director_v2_client=async_client,
+            service_uuid=service_uuid,
+            dynamic_services_urls=dynamic_services_urls,
+        )
 
     # STEP 3
     # pull data via nodeports
diff --git a/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/application.py b/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/application.py
@@ -9,6 +9,7 @@
     get_common_oas_options,
     override_fastapi_openapi_method,
 )
+from servicelib.logging_utils import config_all_loggers
 from simcore_sdk.node_ports_common.exceptions import NodeNotFound
 
 from .._meta import API_VERSION, API_VTAG, PROJECT_NAME, SUMMARY, __version__
@@ -99,6 +100,7 @@ def setup_logger(settings: ApplicationSettings):
     # SEE https://github.com/ITISFoundation/osparc-simcore/issues/3148
     logging.basicConfig(level=settings.log_level)
     logging.root.setLevel(settings.log_level)
+    config_all_loggers()
 
 
 def create_base_app() -> FastAPI:
@@ -122,6 +124,7 @@ def create_base_app() -> FastAPI:
     long_running_tasks.server.setup(app)
 
     app.include_router(main_router)
+
     return app
 
 
diff --git a/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/nodeports.py b/services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/nodeports.py
@@ -3,14 +3,15 @@
 import os
 import shutil
 import sys
-import tempfile
 import time
 from collections import deque
+from contextlib import AsyncExitStack
 from enum import Enum
 from pathlib import Path
 from typing import Any, Coroutine, Optional, cast
 
 import magic
+from aiofiles.tempfile import TemporaryDirectory as AioTemporaryDirectory
 from models_library.projects import ProjectIDStr
 from models_library.projects_nodes import OutputsDict
 from models_library.projects_nodes_io import NodeIDStr
@@ -57,6 +58,11 @@ def _get_size_of_value(value: ItemConcreteValue) -> int:
     return sys.getsizeof(value)
 
 
+_CONTROL_TESTMARK_DY_SIDECAR_NODEPORT_UPLOADED_MESSAGE = (
+    "TEST: test_nodeports_integration DO NOT REMOVE"
+)
+
+
 @run_sequentially_in_context()
 async def upload_outputs(
     outputs_path: Path,
@@ -78,77 +84,69 @@ async def upload_outputs(
     )
 
     # let's gather the tasks
-    temp_paths: deque[Path] = deque()
-    ports_values: dict[str, ItemConcreteValue] = {}
+    ports_values: dict[str, Optional[ItemConcreteValue]] = {}
     archiving_tasks: deque[Coroutine[None, None, None]] = deque()
 
-    for port in (await PORTS.outputs).values():
-        logger.debug("Checking port %s", port.key)
-        if port_keys and port.key not in port_keys:
-            continue
-        logger.debug(
-            "uploading data to port '%s' with value '%s'...", port.key, port.value
-        )
-        if _FILE_TYPE_PREFIX in port.property_type:
-            src_folder = outputs_path / port.key
-            files_and_folders_list = list(src_folder.rglob("*"))
-            logger.debug("Discovered files to upload %s", files_and_folders_list)
-
-            if not files_and_folders_list:
-                ports_values[port.key] = None
-                continue
-
-            if len(files_and_folders_list) == 1 and (
-                files_and_folders_list[0].is_file()
-                or files_and_folders_list[0].is_symlink()
-            ):
-                # special case, direct upload
-                ports_values[port.key] = files_and_folders_list[0]
+    async with AsyncExitStack() as stack:
+        for port in (await PORTS.outputs).values():
+            logger.debug("Checking port %s", port.key)
+            if port_keys and port.key not in port_keys:
                 continue
-
-            # generic case let's create an archive
-            # only the filtered out files will be zipped
-            tmp_folder = Path(tempfile.mkdtemp())
-            tmp_file = tmp_folder / f"{src_folder.stem}.zip"
-            temp_paths.append(tmp_folder)
-
-            # when having multiple directories it is important to
-            # run the compression in parallel to guarantee better performance
-            archiving_tasks.append(
-                archive_dir(
-                    dir_to_compress=src_folder,
-                    destination=tmp_file,
-                    compress=False,
-                    store_relative_path=True,
-                )
+            logger.debug(
+                "uploading data to port '%s' with value '%s'...", port.key, port.value
             )
-            ports_values[port.key] = tmp_file
-        else:
-            data_file = outputs_path / _KEY_VALUE_FILE_NAME
-            if data_file.exists():
-                data = json.loads(data_file.read_text())
-                if port.key in data and data[port.key] is not None:
-                    ports_values[port.key] = data[port.key]
-                else:
-                    logger.debug("Port %s not found in %s", port.key, data)
+            if _FILE_TYPE_PREFIX in port.property_type:
+                src_folder = outputs_path / port.key
+                files_and_folders_list = list(src_folder.rglob("*"))
+                logger.debug("Discovered files to upload %s", files_and_folders_list)
+
+                if not files_and_folders_list:
+                    ports_values[port.key] = None
+                    continue
+
+                if len(files_and_folders_list) == 1 and (
+                    files_and_folders_list[0].is_file()
+                    or files_and_folders_list[0].is_symlink()
+                ):
+                    # special case, direct upload
+                    ports_values[port.key] = files_and_folders_list[0]
+                    continue
+
+                # generic case let's create an archive
+                # only the filtered out files will be zipped
+                tmp_folder = await stack.enter_async_context(AioTemporaryDirectory())
+                tmp_file = tmp_folder / f"{src_folder.stem}.zip"
+
+                # when having multiple directories it is important to
+                # run the compression in parallel to guarantee better performance
+                archiving_tasks.append(
+                    archive_dir(
+                        dir_to_compress=src_folder,
+                        destination=tmp_file,
+                        compress=False,
+                        store_relative_path=True,
+                    )
+                )
+                ports_values[port.key] = tmp_file
             else:
-                logger.debug("No file %s to fetch port values from", data_file)
+                data_file = outputs_path / _KEY_VALUE_FILE_NAME
+                if data_file.exists():
+                    data = json.loads(data_file.read_text())
+                    if port.key in data and data[port.key] is not None:
+                        ports_values[port.key] = data[port.key]
+                    else:
+                        logger.debug("Port %s not found in %s", port.key, data)
+                else:
+                    logger.debug("No file %s to fetch port values from", data_file)
 
-    try:
         if archiving_tasks:
             await logged_gather(*archiving_tasks)
         await PORTS.set_multiple(ports_values)
 
-        elapsed_time = time.perf_counter() - start_time
-        total_bytes = sum(_get_size_of_value(x) for x in ports_values.values())
-        logger.info("Uploaded %s bytes in %s seconds", total_bytes, elapsed_time)
-    finally:
-        # clean up possible compressed files
-        for file_path in temp_paths:
-            await async_on_threadpool(
-                # pylint: disable=cell-var-from-loop
-                lambda: shutil.rmtree(file_path.parent, ignore_errors=True)
-            )
+    elapsed_time = time.perf_counter() - start_time
+    total_bytes = sum(_get_size_of_value(x) for x in ports_values.values())
+    logger.info("Uploaded %s bytes in %s seconds", total_bytes, elapsed_time)
+    logger.debug(_CONTROL_TESTMARK_DY_SIDECAR_NODEPORT_UPLOADED_MESSAGE)
 
 
 async def dispatch_update_for_directory(
