🐛 Making storage logs more readable & faster decompression (ITISFoundation#3024)

* cancelled error is now captured and logged

* errors are more meaningful now

* pylint

* refactor

* better wording

* making file unzipping way faster

* docstrings

* pylint

* usigng create task

* improoved log messages

* pylint

* raised error to debug

* updated docstring

* making parameter mandatory

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

Author: GitHK

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

@@ -42,17 +42,42 @@ def _read_in_chunks(file_object, chunk_size=1024 * 8):
         yield data
 
 
+class _FastZipFileReader(zipfile.ZipFile):
+    """
+    Used to gain a speed boost of several orders of magnitude.
+
+    When opening archives the `_RealGetContents` is called 
+    generating the list of files contained in the zip archive.
+    This is done by the constructor.
+
+    If the archive contains a very large amount, the file scan operation
+    can take up to seconds. This was observed with 10000+ files.
+
+    When opening the zip file in the background worker the entire file 
+    list generation can be skipped because the `zipfile.ZipFile.open` 
+    is used passing `ZipInfo` object as file to decompress. 
+    Using a `ZipInfo` object does nto require to have the list of 
+    files contained in the archive.
+    """
+
+    def _RealGetContents(self):
+        """method disabled"""
+
+
 def _zipfile_single_file_extract_worker(
-    zip_file_path: Path, file_in_archive: str, destination_folder: Path, is_dir: bool
+    zip_file_path: Path,
+    file_in_archive: zipfile.ZipInfo,
+    destination_folder: Path,
+    is_dir: bool,
 ) -> Path:
     """Extracts file_in_archive from the archive zip_file_path -> destination_folder/file_in_archive
 
     Extracts in chunks to avoid memory pressure on zip/unzip
-    Retuns a path to extracted file or directory
+    returns: a path to extracted file or directory
     """
-    with zipfile.ZipFile(zip_file_path) as zf:
+    with _FastZipFileReader(zip_file_path) as zf:
         # assemble destination and ensure it exits
-        destination_path = destination_folder / file_in_archive
+        destination_path = destination_folder / file_in_archive.filename
 
         if is_dir:
             destination_path.mkdir(parents=True, exist_ok=True)
@@ -109,7 +134,7 @@ async def unarchive_dir(
                     pool,
                     _zipfile_single_file_extract_worker,
                     archive_to_extract,
-                    zip_entry.filename,
+                    zip_entry,
                     destination_folder,
                     zip_entry.is_dir(),
                 )

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

@@ -1,7 +1,7 @@
 import asyncio
 from concurrent.futures import ProcessPoolExecutor
 from contextlib import contextmanager
-from typing import Any, Callable
+from typing import Any, Callable, Iterator
 
 # only gets created on use and is guaranteed to be the s
 # ame for the entire lifetime of the application
@@ -21,7 +21,7 @@ def get_shared_process_pool_executor(**kwargs) -> ProcessPoolExecutor:
 
 
 @contextmanager
-def non_blocking_process_pool_executor(**kwargs) -> ProcessPoolExecutor:
+def non_blocking_process_pool_executor(**kwargs) -> Iterator[ProcessPoolExecutor]:
     """
     Avoids default context manger behavior which calls
     shutdown with wait=True an blocks.

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

@@ -69,8 +69,10 @@ def fire_and_forget_task(
     def log_exception_callback(fut: asyncio.Future):
         try:
             fut.result()
+        except asyncio.CancelledError:
+            logger.exception("%s spawned as fire&forget was cancelled", fut)
         except Exception:  # pylint: disable=broad-except
-            logger.exception("Error occured while running task!")
+            logger.exception("Error occurred while running task!")
 
     future.add_done_callback(log_exception_callback)
     return future

packages/service-library/tests/test_utils.py

@@ -4,9 +4,10 @@
 
 import asyncio
 from pathlib import Path
+from typing import Awaitable, Coroutine, Union, Type
 
 import pytest
-from servicelib.utils import logged_gather
+from servicelib.utils import logged_gather, fire_and_forget_task
 
 
 async def _value_error(uid, *, delay=1):
@@ -91,3 +92,45 @@ def print_tree(path: Path, level=0):
     print(f"{tab}{'+' if path.is_dir() else '-'} {path if level==0 else path.name}")
     for p in path.glob("*"):
         print_tree(p, level + 1)
+
+
+@pytest.fixture(params=[Awaitable, Coroutine])
+def future_type(request) -> Type[Union[asyncio.Future, Awaitable]]:
+    return request.param
+
+
+@pytest.fixture()
+async def future_to_test(
+    future_type: Type[Union[asyncio.Future, Awaitable]]
+) -> Union[asyncio.Future, Awaitable]:
+    async def _self_cancelling() -> None:
+        raise asyncio.CancelledError("manual cancellation")
+
+    if future_type == Awaitable:
+        return asyncio.create_task(_self_cancelling())
+    if future_type == Coroutine:
+        return _self_cancelling()
+
+    raise RuntimeError("not defined")
+
+
+async def test_fire_and_forget_cancellation_errors_raised_when_awaited(
+    future_type: Type[Union[asyncio.Future, Awaitable]],
+    future_to_test: Union[asyncio.Future, Awaitable],
+) -> None:
+    assert isinstance(future_to_test, future_type)
+
+    task = fire_and_forget_task(future_to_test)
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+
+async def test_fire_and_forget_cancellation_no_errors_raised(
+    future_type: Type[Union[asyncio.Future, Awaitable]],
+    future_to_test: Union[asyncio.Future, Awaitable],
+) -> None:
+    assert isinstance(future_to_test, future_type)
+
+    task = fire_and_forget_task(future_to_test)
+    await asyncio.sleep(0.1)
+    assert task.cancelled() is True

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

@@ -205,12 +205,14 @@ async def _get_networks_with_aliases_for_default_network(
                 project_id=project_id,
                 node_id=UUID(node_uuid),
                 messages=[
+                    # pylint:disable=anomalous-backslash-in-string
                     (
-                        f"Service with label '{node_content.label}' cannot "
-                        f"be connected to network {default_network}. If you "
-                        "whish for your services in this study to communicate "
-                        "provide a label containing the following characters "
-                        "'a-zA-Z0-9_-' and must be max 64 letters long."
+                        f"Service with label '{node_content.label}' cannot be "
+                        "identified on service network due to invalid name. "
+                        "To communicate over the network, please rename the "
+                        "service alphanumeric characters <64 characters, "
+                        "e.g. re.sub(r'\W+', '', SERVICE_NAME). "
+                        f"Network name is {default_network}"
                     )
                 ],
             )

services/storage/src/simcore_service_storage/dsm.py

@@ -31,8 +31,9 @@
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.sql.expression import literal_column
 from tenacity import retry
+from tenacity._asyncio import AsyncRetrying
 from tenacity.before_sleep import before_sleep_log
-from tenacity.retry import retry_if_exception_type, retry_if_result
+from tenacity.retry import retry_if_exception_type
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_exponential
 from yarl import URL
@@ -234,6 +235,7 @@ async def list_files(
                             dex.fmd.file_uuid,
                             dex.fmd.bucket_name,
                             dex.fmd.object_name,
+                            reraise_exceptions=False,
                         )
                     if dex:
                         data.append(dex)
@@ -410,6 +412,7 @@ async def list_file(
                         file_metadata.fmd.file_uuid,
                         file_metadata.fmd.bucket_name,
                         file_metadata.fmd.object_name,
+                        reraise_exceptions=False,
                     )
                     return file_metadata
                 # FIXME: returns None in both cases: file does not exist or use has no access
@@ -441,7 +444,8 @@ async def try_update_database_from_storage(
         file_uuid: str,
         bucket_name: str,
         object_name: str,
-        silence_exception: bool = False,
+        *,
+        reraise_exceptions: bool,
     ) -> Optional[FileMetaDataEx]:
         try:
             async with self._create_aiobotocore_client_context() as aioboto_client:
@@ -472,29 +476,23 @@ async def try_update_database_from_storage(
 
                     return to_meta_data_extended(row)
         except botocore.exceptions.ClientError:
-            if silence_exception:
-                logger.debug("Error happened while trying to access %s", file_uuid)
-            else:
-                logger.warning(
-                    "Error happened while trying to access %s", file_uuid, exc_info=True
-                )
-            # the file is not existing or some error happened
-            return None
+            logger.warning("Error happened while trying to access %s", file_uuid)
+            if reraise_exceptions:
+                raise
 
-    @retry(
-        stop=stop_after_delay(1 * _HOUR),
-        wait=wait_exponential(multiplier=0.1, exp_base=1.2, max=30),
-        retry=(
-            retry_if_exception_type() | retry_if_result(lambda result: result is None)
-        ),
-        before_sleep=before_sleep_log(logger, logging.INFO),
-    )
     async def auto_update_database_from_storage_task(
         self, file_uuid: str, bucket_name: str, object_name: str
     ) -> Optional[FileMetaDataEx]:
-        return await self.try_update_database_from_storage(
-            file_uuid, bucket_name, object_name, silence_exception=True
-        )
+        async for attempt in AsyncRetrying(
+            stop=stop_after_delay(1 * _HOUR),
+            wait=wait_exponential(multiplier=0.1, exp_base=1.2, max=30),
+            retry=(retry_if_exception_type()),
+            before_sleep=before_sleep_log(logger, logging.INFO),
+        ):
+            with attempt:
+                return await self.try_update_database_from_storage(
+                    file_uuid, bucket_name, object_name, reraise_exceptions=True
+                )
 
     async def update_metadata(
         self, file_uuid: str, user_id: int

services/storage/tests/unit/test_dsm.py

@@ -180,29 +180,29 @@ async def test_update_metadata_from_storage(
 
     assert (
         await dsm_fixture.try_update_database_from_storage(  # pylint: disable=protected-access
-            "some_fake_uuid", fmd.bucket_name, fmd.object_name
+            "some_fake_uuid", fmd.bucket_name, fmd.object_name, reraise_exceptions=False
         )
         is None
     )
 
     assert (
         await dsm_fixture.try_update_database_from_storage(  # pylint: disable=protected-access
-            fmd.file_uuid, "some_fake_bucket", fmd.object_name
+            fmd.file_uuid, "some_fake_bucket", fmd.object_name, reraise_exceptions=False
         )
         is None
     )
 
     assert (
         await dsm_fixture.try_update_database_from_storage(  # pylint: disable=protected-access
-            fmd.file_uuid, fmd.bucket_name, "some_fake_object"
+            fmd.file_uuid, fmd.bucket_name, "some_fake_object", reraise_exceptions=False
         )
         is None
     )
 
     file_metadata: Optional[
         FileMetaDataEx
     ] = await dsm_fixture.try_update_database_from_storage(  # pylint: disable=protected-access
-        fmd.file_uuid, fmd.bucket_name, fmd.object_name
+        fmd.file_uuid, fmd.bucket_name, fmd.object_name, reraise_exceptions=False
     )
     assert file_metadata is not None
     assert file_metadata.fmd.file_size == Path(tmp_file).stat().st_size