improvements

Author: matusdrobuliak66

services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks_setup.py

@@ -40,7 +40,7 @@ async def _startup() -> None:
                     get_redis_lock_client(app),
                     task["task_func"],
                     task_period=timedelta(seconds=60),  # 1 minute
-                    retry_after=timedelta(seconds=60),  # 5 minutes
+                    retry_after=timedelta(seconds=300),  # 5 minutes
                     task_name=task["name"],
                     app=app,
                 )

services/efs-guardian/src/simcore_service_efs_guardian/services/efs_manager.py

@@ -55,6 +55,18 @@ async def create_project_specific_data_dir(
         )  # This gives rwx permissions to user and group, and nothing to others
         return _dir_path
 
+    async def check_project_node_data_directory_exits(
+        self, project_id: ProjectID, node_id: NodeID
+    ) -> bool:
+        _dir_path = (
+            self._efs_mounted_path
+            / self._project_specific_data_base_directory
+            / f"{project_id}"
+            / f"{node_id}"
+        )
+
+        return _dir_path.exists()
+
     async def get_project_node_data_size(
         self, project_id: ProjectID, node_id: NodeID
     ) -> ByteSize:

services/efs-guardian/src/simcore_service_efs_guardian/services/efs_manager_utils.py

@@ -1,52 +1,48 @@
 import asyncio
+import logging
 
 from pydantic import ByteSize
 
+_logger = logging.getLogger(__name__)
+
 
 async def get_size_bash_async(path) -> ByteSize:
-    try:
-        # Create the subprocess
-        process = await asyncio.create_subprocess_exec(
-            "du",
-            "-sb",
-            path,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-
-        # Wait for the subprocess to complete
-        stdout, stderr = await process.communicate()
-
-        if process.returncode == 0:
-            # Parse the output
-            size = ByteSize(stdout.decode().split()[0])
-            return size
-        else:
-            print(f"Error: {stderr.decode()}")
-            raise ValueError
-    except Exception as e:
-        raise e
+    # Create the subprocess
+    command = ["du", "-sb", path]
+    process = await asyncio.create_subprocess_exec(
+        *command,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+
+    # Wait for the subprocess to complete
+    stdout, stderr = await process.communicate()
+
+    if process.returncode == 0:
+        # Parse the output
+        size = ByteSize(stdout.decode().split()[0])
+        return size
+    else:
+        msg = f"Command {' '.join(command)} failed with error code {process.returncode}: {stderr.decode()}"
+        _logger.error(msg)
+        raise RuntimeError(msg)
 
 
 async def remove_write_permissions_bash_async(path) -> None:
-    try:
-        # Create the subprocess
-        process = await asyncio.create_subprocess_exec(
-            "chmod",
-            "-R",
-            "a-w",
-            path,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-
-        # Wait for the subprocess to complete
-        stdout, stderr = await process.communicate()
-
-        if process.returncode == 0:
-            return
-        else:
-            print(f"Error: {stderr.decode()}")
-            raise ValueError
-    except Exception as e:
-        raise e
+    # Create the subprocess
+    command = ["chmod", "-R", "a-w", path]
+    process = await asyncio.create_subprocess_exec(
+        *command,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+
+    # Wait for the subprocess to complete
+    _, stderr = await process.communicate()
+
+    if process.returncode == 0:
+        return
+    else:
+        msg = f"Command {' '.join(command)} failed with error code {process.returncode}: {stderr.decode()}"
+        _logger.error(msg)
+        raise RuntimeError(msg)

services/efs-guardian/src/simcore_service_efs_guardian/services/process_messages.py

@@ -17,23 +17,46 @@ async def process_dynamic_service_running_message(app: FastAPI, data: bytes) ->
         DynamicServiceRunningMessage, data  # type: ignore[arg-type]
     )
     _logger.debug(
-        "Process dynamic service running msg, project ID: %s node ID: %s",
+        "Process dynamic service running msg, project ID: %s node ID: %s, current user: %s",
         rabbit_message.project_id,
         rabbit_message.node_id,
+        rabbit_message.user_id,
     )
 
     settings = get_application_settings(app)
     efs_manager: EfsManager = app.state.efs_manager
+
+    dir_exists = await efs_manager.check_project_node_data_directory_exits(
+        rabbit_message.project_id, node_id=rabbit_message.node_id
+    )
+    if dir_exists is False:
+        _logger.debug(
+            "Directory doesn't exists in EFS, project ID: %s node ID: %s, current user: %s",
+            rabbit_message.project_id,
+            rabbit_message.node_id,
+            rabbit_message.user_id,
+        )
+        return True
+
     size = await efs_manager.get_project_node_data_size(
         rabbit_message.project_id, node_id=rabbit_message.node_id
     )
+    _logger.debug(
+        "Current directory size: %s, project ID: %s node ID: %s, current user: %s",
+        size,
+        rabbit_message.project_id,
+        rabbit_message.node_id,
+        rabbit_message.user_id,
+    )
 
     if size > settings.EFS_DEFAULT_USER_SERVICE_SIZE_BYTES:
         _logger.warning(
-            "Removing write permissions inside of EFS starts for project ID: %s, node ID: %s, current user: %s",
+            "Removing write permissions inside of EFS starts for project ID: %s, node ID: %s, current user: %s, size: %s, upper limit: %s",
             rabbit_message.project_id,
             rabbit_message.node_id,
             rabbit_message.user_id,
+            size,
+            settings.EFS_DEFAULT_USER_SERVICE_SIZE_BYTES,
         )
         redis = get_redis_lock_client(app)
         async with redis.lock_context(

services/efs-guardian/tests/unit/test_efs_manager.py

@@ -104,6 +104,13 @@ async def test_remove_write_access_rights(
 
     efs_manager: EfsManager = app.state.efs_manager
 
+    assert (
+        await efs_manager.check_project_node_data_directory_exits(
+            project_id=_project_id, node_id=_node_id
+        )
+        is False
+    )
+
     with patch(
         "simcore_service_efs_guardian.services.efs_manager.os.chown"
     ) as mocked_chown:
@@ -113,6 +120,13 @@ async def test_remove_write_access_rights(
             storage_directory_name=_storage_directory_name,
         )
 
+    assert (
+        await efs_manager.check_project_node_data_directory_exits(
+            project_id=_project_id, node_id=_node_id
+        )
+        is True
+    )
+
     size_before = await efs_manager.get_project_node_data_size(
         project_id=_project_id, node_id=_node_id
     )