🐛 Dask scheduler: Fix issue when value and/or size is not set in file download progress (ITISFoundation#2627)

* add hostname as name so it's easier to find which dask-sidecar

Author: sanderegg

packages/pytest-simcore/src/pytest_simcore/helpers/utils_docker.py

@@ -179,4 +179,5 @@ def save_docker_infos(destination_path: Path):
                 cont.logs(timestamps=True, stdout=True, stderr=True).decode(), width=200
             ),
         )
-    print("\n\twrote docker log files in ", destination_path)
+    if all_containers:
+        print("\n\twrote docker log files in ", destination_path)

packages/simcore-sdk/src/simcore_sdk/node_ports_common/filemanager.py

@@ -318,3 +318,14 @@ async def get_file_metadata(
         raise exceptions.StorageInvalidCall(f"The file '{s3_object}' cannot be found")
     log.debug("Result for metadata s3_object=%s, result=%s", s3_object, result)
     return (f"{result.get('location_id', '')}", result.get("entity_tag", ""))
+
+
+async def delete_file(
+    user_id: int,
+    store_id: str,
+    s3_object: str,
+    client_session: Optional[ClientSession] = None,
+) -> None:
+    async with ClientSessionContextManager(client_session) as session:
+        log.debug("Will delete file for s3_object=%s", s3_object)
+        await storage_client.delete_file(session, s3_object, store_id, user_id)

packages/simcore-sdk/src/simcore_sdk/node_ports_common/storage_client.py

@@ -149,3 +149,26 @@ async def get_file_metadata(
         if file_metadata_enveloped.data is None:
             raise exceptions.S3InvalidPathError(file_id)
         return file_metadata_enveloped.data.dict(by_alias=True)
+
+
+@handle_client_exception
+async def delete_file(
+    session: ClientSession, file_id: str, location_id: str, user_id: UserID
+) -> None:
+    if (
+        not isinstance(file_id, str)
+        or not isinstance(location_id, str)
+        or not isinstance(user_id, int)
+    ):
+        raise exceptions.StorageInvalidCall(
+            f"invalid call: user_id '{user_id}', location_id '{location_id}', file_id '{file_id}' are invalid",
+        )
+    if file_id is None or location_id is None or user_id is None:
+        raise exceptions.StorageInvalidCall(
+            f"invalid call: user_id '{user_id}', location_id '{location_id}', file_id '{file_id}' are not allowed to be empty",
+        )
+    async with session.delete(
+        f"{_base_url()}/locations/{location_id}/files/{quote(file_id, safe='')}",
+        params={"user_id": f"{user_id}"},
+    ) as response:
+        response.raise_for_status()

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/port_utils.py

@@ -91,6 +91,28 @@ async def get_upload_link_from_storage(
     return parse_obj_as(AnyUrl, f"{link}")
 
 
+async def target_link_exists(
+    user_id: int, project_id: str, node_id: str, file_name: str
+) -> bool:
+    log.debug(
+        "checking if target of link to file from storage for %s exists", file_name
+    )
+    s3_object = data_items_utils.encode_file_id(Path(file_name), project_id, node_id)
+    return await filemanager.entry_exists(
+        user_id=user_id, store_id="0", s3_object=s3_object
+    )
+
+
+async def delete_target_link(
+    user_id: int, project_id: str, node_id: str, file_name: str
+) -> None:
+    log.debug("deleting target of link to file from storage for %s", file_name)
+    s3_object = data_items_utils.encode_file_id(Path(file_name), project_id, node_id)
+    return await filemanager.delete_file(
+        user_id=user_id, store_id="0", s3_object=s3_object
+    )
+
+
 async def pull_file_from_store(
     user_id: int,
     key: str,

packages/simcore-sdk/src/simcore_sdk/node_ports_v2/ports_mapping.py

@@ -18,6 +18,7 @@ def __getitem__(self, key: Union[int, PortKey]) -> Port:
                 key = list(self.__root__.keys())[key]
         if not key in self.__root__:
             raise UnboundPortError(key)
+        assert isinstance(key, str)  # no sec
         return self.__root__[key]
 
     def __iter__(self) -> Iterator[PortKey]:

packages/simcore-sdk/tests/integration/test_node_ports_common_filemanager.py

@@ -5,7 +5,7 @@
 
 import filecmp
 from pathlib import Path
-from typing import Callable
+from typing import Any, Awaitable, Callable, Optional
 from uuid import uuid4
 
 import np_helpers
@@ -125,7 +125,9 @@ async def test_errors_upon_invalid_file_identifiers(
         await filemanager.download_file_from_s3(
             user_id=user_id,
             store_id=store,
-            s3_object=np_helpers.file_uuid("invisible.txt", project_id, f"{uuid4()}"),
+            s3_object=np_helpers.file_uuid(
+                Path("invisible.txt"), project_id, f"{uuid4()}"
+            ),
             local_folder=download_folder,
         )
 
@@ -201,27 +203,70 @@ async def test_valid_metadata(
     assert is_metadata_present is True
 
 
+@pytest.mark.parametrize(
+    "fct",
+    [filemanager.entry_exists, filemanager.delete_file, filemanager.get_file_metadata],
+)
 async def test_invalid_call_raises_exception(
     tmpdir: Path,
     bucket: str,
     filemanager_cfg: None,
     user_id: int,
     create_valid_file_uuid: Callable[[Path], str],
     s3_simcore_location: str,
+    fct: Callable[[int, str, str, Optional[Any]], Awaitable],
 ):
     file_path = Path(tmpdir) / "test.test"
     file_id = create_valid_file_uuid(file_path)
     assert file_path.exists() is False
 
     with pytest.raises(exceptions.StorageInvalidCall):
-        await filemanager.entry_exists(
+        await fct(
             user_id=None, store_id=s3_simcore_location, s3_object=file_id  # type: ignore
         )
     with pytest.raises(exceptions.StorageInvalidCall):
-        await filemanager.entry_exists(
-            user_id=user_id, store_id=None, s3_object=file_id  # type: ignore
-        )
+        await fct(user_id=user_id, store_id=None, s3_object=file_id)  # type: ignore
     with pytest.raises(exceptions.StorageInvalidCall):
-        await filemanager.entry_exists(
+        await fct(
             user_id=user_id, store_id=s3_simcore_location, s3_object=None  # type: ignore
         )
+
+
+async def test_delete_File(
+    tmpdir: Path,
+    bucket: str,
+    filemanager_cfg: None,
+    user_id: int,
+    create_valid_file_uuid: Callable[[Path], str],
+    s3_simcore_location: str,
+):
+    file_path = Path(tmpdir) / "test.test"
+    file_path.write_text("I am a test file")
+    assert file_path.exists()
+
+    file_id = create_valid_file_uuid(file_path)
+    store_id, e_tag = await filemanager.upload_file(
+        user_id=user_id,
+        store_id=s3_simcore_location,
+        s3_object=file_id,
+        local_file_path=file_path,
+    )
+    assert store_id == s3_simcore_location
+    assert e_tag
+
+    is_metadata_present = await filemanager.entry_exists(
+        user_id=user_id, store_id=store_id, s3_object=file_id
+    )
+    assert is_metadata_present is True
+
+    await filemanager.delete_file(
+        user_id=user_id, store_id=s3_simcore_location, s3_object=file_id
+    )
+
+    # check that it disappeared
+    assert (
+        await filemanager.entry_exists(
+            user_id=user_id, store_id=store_id, s3_object=file_id
+        )
+        == False
+    )

services/dask-sidecar/Dockerfile

@@ -125,9 +125,9 @@ COPY --chown=scu:scu services/dask-sidecar/docker services/dask-sidecar/docker
 # make sure dask worker is started as ``dask-worker --dashboard-address 8787``.
 # Otherwise the worker will take random ports to serve the /health entrypoint.
 HEALTHCHECK \
-  --interval=60s \
-  --timeout=60s \
-  --start-period=10s \
+  --interval=30s \
+  --timeout=15s \
+  --start-period=5s \
   --retries=3 \
   CMD ["curl", "-Lf", "http://127.0.0.1:8787/health"]
 

services/dask-sidecar/docker/boot.sh

@@ -111,19 +111,23 @@ else
       --preload simcore_service_dask_sidecar.tasks \
       --reconnect \
       --no-nanny \
+      --nprocs 1 \
       --nthreads "$num_cpus" \
       --dashboard-address 8787 \
       --memory-limit "$ram" \
-      --resources "$resources"
+      --resources "$resources" \
+      --name "$(hostname)"
   else
     exec dask-worker "${DASK_SCHEDULER_ADDRESS}" \
       --local-directory /tmp/dask-sidecar \
       --preload simcore_service_dask_sidecar.tasks \
       --reconnect \
       --no-nanny \
+      --nprocs 1 \
       --nthreads "$num_cpus" \
       --dashboard-address 8787 \
       --memory-limit "$ram" \
-      --resources "$resources"
+      --resources "$resources" \
+      --name "$(hostname)"
   fi
 fi

services/dask-sidecar/src/simcore_service_dask_sidecar/computational_sidecar/core.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from pprint import pformat
 from types import TracebackType
-from typing import Any, Awaitable, Dict, List, Optional, Type
+from typing import Any, Dict, List, Optional, Type
 from uuid import uuid4
 
 from aiodocker import Docker
@@ -160,7 +160,7 @@ async def _publish_sidecar_state(
     async def run(self, command: List[str]) -> TaskOutputData:
         await self._publish_sidecar_state(RunningState.STARTED)
         await self._publish_sidecar_log(
-            f"Starting task for {self.service_key}:{self.service_version}..."
+            f"Starting task for {self.service_key}:{self.service_version} on {socket.gethostname()}..."
         )
 
         settings = Settings.create_from_envs()
@@ -222,10 +222,6 @@ async def run(self, command: List[str]) -> TaskOutputData:
                             msg=f"error while running container '{container.id}' for '{self.service_key}:{self.service_version}'",
                         )
 
-                        await self._publish_sidecar_log(
-                            f"Error while running container: task FAILED with exit code '{container_data['State']['ExitCode']}'"
-                        )
-
                         raise ServiceRunError(
                             self.service_key,
                             self.service_version,
@@ -250,6 +246,9 @@ async def __aexit__(
         exc_type: Optional[Type[BaseException]],
         exc: Optional[BaseException],
         tb: Optional[TracebackType],
-    ) -> Awaitable[Optional[bool]]:
-        """NOTE: this is empty but this is intended. the ComputationSidecar
-        is meant to be used as context manager"""
+    ) -> None:
+        if exc:
+            await self._publish_sidecar_log(f"Task error:\n{exc}")
+            await self._publish_sidecar_log(
+                "There might be more information in the service log file"
+            )

services/dask-sidecar/src/simcore_service_dask_sidecar/file_utils.py

@@ -33,8 +33,8 @@ def _file_progress_cb(
     asyncio.run_coroutine_threadsafe(
         log_publishing_cb(
             f"{text_prefix}"
-            f" {100.0 * float(value)/float(size):.1f}%"
-            f" ({ByteSize(value).human_readable()} / {ByteSize(size).human_readable() if size != 0 else 'NaN'})"
+            f" {100.0 * float(value or 0)/float(size or 1):.1f}%"
+            f" ({ByteSize(value).human_readable() if value else 0} / {ByteSize(size).human_readable() if size else 'NaN'})"
         ),
         main_loop,
     )