Bugfix/storage lazy update of files (ITISFoundation#2636)

Author: sanderegg

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

@@ -18,7 +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
+        assert isinstance(key, str)  # nosec
         return self.__root__[key]
 
     def __iter__(self) -> Iterator[PortKey]:

scripts/maintenance/check_consistency_data.py

@@ -1,15 +1,6 @@
 #! /usr/bin/env python3
 
-"""Script to check consistency of the file storage backend in oSparc.
 
-
-
-    1. From an osparc database, go over all projects, get the project IDs and Node IDs
-    2. From the same database, now get all the files listed like projectID/nodeID from 1.
-    3. We get a list of files that are needed for the current projects
-    4. connect to the S3 backend, check that these files exist
-    5. generate a report with: project uuid, owner, files missing in S3
-"""
 import asyncio
 import csv
 import logging
@@ -25,7 +16,10 @@
 import aiopg
 import typer
 from dateutil import parser
-from tenacity import after_log, retry, stop_after_attempt, wait_random
+from tenacity import retry
+from tenacity.after import after_log
+from tenacity.stop import stop_after_attempt
+from tenacity.wait import wait_random
 
 log = logging.getLogger(__name__)
 
@@ -38,8 +32,8 @@ async def managed_docker_compose(
     compose_file = Path.cwd() / "consistency" / "docker-compose.yml"
     try:
         subprocess.run(
-            f"docker-compose --file {compose_file} up --detach",
-            shell=True,
+            ["docker-compose", "--file", compose_file, "up", "--detach"],
+            shell=False,
             check=True,
             cwd=compose_file.parent,
             env={**os.environ, **{"POSTGRES_DATA_VOLUME": postgres_volume_name}},
@@ -65,8 +59,8 @@ async def postgres_responsive():
         yield
     finally:
         subprocess.run(
-            f"docker-compose --file {compose_file} down",
-            shell=True,
+            ["docker-compose", "--file", compose_file, "down"],
+            shell=False,
             check=True,
             cwd=compose_file.parent,
         )
@@ -122,6 +116,23 @@ async def _get_files_from_project_nodes(
             }
 
 
+async def _get_all_invalid_files_from_file_meta_data(
+    pool,
+) -> Set[Tuple[str, int, datetime]]:
+    async with pool.acquire() as conn:
+        async with conn.cursor() as cursor:
+            await cursor.execute(
+                'SELECT file_uuid, file_size, last_modified FROM "file_meta_data" '
+                "WHERE file_meta_data.file_size < 1 OR file_meta_data.entity_tag IS NULL"
+            )
+            # here we got all the files for that project uuid/node_ids combination
+            file_rows = await cursor.fetchall()
+            return {
+                (file_uuid, file_size, parser.parse(last_modified or "2000-01-01"))
+                for file_uuid, file_size, last_modified in file_rows
+            }
+
+
 POWER_LABELS = {0: "B", 1: "KiB", 2: "MiB", 3: "GiB"}
 LABELS_POWER = {v: k for k, v in POWER_LABELS.items()}
 
@@ -214,7 +225,7 @@ async def main_async(
     s3_bucket: str,
 ):
 
-    # ---------------------- GET FILE ENTRIES FROM DB ---------------------------------------------------------------------
+    # ---------------------- GET FILE ENTRIES FROM DB PROKECT TABLE -------------------------------------------------------------
     async with managed_docker_compose(
         postgres_volume_name, postgres_username, postgres_password
     ):
@@ -231,6 +242,9 @@ async def main_async(
                     for project_uuid, prj_data in project_nodes.items()
                 ]
             )
+            all_invalid_files_in_file_meta_data = (
+                await _get_all_invalid_files_from_file_meta_data(pool)
+            )
     db_file_entries: Set[Tuple[str, int, datetime]] = set().union(
         *all_sets_of_file_entries
     )
@@ -243,6 +257,20 @@ async def main_async(
         fg=typer.colors.YELLOW,
     )
 
+    if all_invalid_files_in_file_meta_data:
+        db_file_meta_data_invalid_entries_path = (
+            Path.cwd() / f"{s3_endpoint}_db_file_meta_data_invalid_entries.csv"
+        )
+        write_file(
+            db_file_meta_data_invalid_entries_path,
+            all_invalid_files_in_file_meta_data,
+            ["file_uuid", "size", "last modified"],
+        )
+        typer.secho(
+            f"processed {len(all_invalid_files_in_file_meta_data)} INVALID file entries, saved in {db_file_meta_data_invalid_entries_path}",
+            fg=typer.colors.YELLOW,
+        )
+
     # ---------------------- GET FILE ENTRIES FROM S3 ---------------------------------------------------------------------
     # let's proceed with S3 backend: files are saved in BUCKET_NAME/projectID/nodeID/fileA.ext
     # Rationale: Similarly we list here all the files in each of the projects. And it goes faster to list them recursively.
@@ -336,6 +364,21 @@ def main(
     s3_secret: str,
     s3_bucket: str,
 ):
+    """Script to check consistency of the file storage backend in oSparc.
+
+    requirements:
+    - local docker volume containing a database from a deployment (see make import-db-from-docker-volume in /packages/postgres-database)
+
+    1. From an osparc database, go over all projects, get the project IDs and Node IDs
+
+    2. From the same database, now get all the files listed like projectID/nodeID from 1.
+
+    3. We get a list of files that are needed for the current projects
+
+    4. connect to the S3 backend, check that these files exist
+
+    5. generate a report with: project uuid, owner, files missing in S3"""
+
     asyncio.run(
         main_async(
             postgres_volume_name,

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

@@ -118,7 +118,7 @@ async def push_file_to_remote(
                     None, zfp.write, src_path, src_path.name
                 )
             logger.debug("%s created.", archive_file_path)
-            assert archive_file_path.exists()  # no sec
+            assert archive_file_path.exists()  # nosec
             file_to_upload = archive_file_path
             await log_publishing_cb(
                 f"Compression of '{src_path.name}' to '{archive_file_path.name}' complete."

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/dask_scheduler.py

@@ -89,7 +89,7 @@ async def _on_task_completed(self, event: TaskStateEvent) -> None:
             event.job_id
         )
 
-        assert event.state in COMPLETED_STATES  # no sec
+        assert event.state in COMPLETED_STATES  # nosec
 
         logger.info(
             "task %s completed with state: %s",
@@ -98,7 +98,7 @@ async def _on_task_completed(self, event: TaskStateEvent) -> None:
         )
         if event.state == RunningState.SUCCESS:
             # we need to parse the results
-            assert event.msg  # no sec
+            assert event.msg  # nosec
             await parse_output_data(
                 self.db_engine,
                 event.job_id,

services/director-v2/src/simcore_service_director_v2/utils/dask.py

@@ -225,7 +225,7 @@ def done_dask_callback(
         # remove the future from the dict to remove any handle to the future, so the worker can free the memory
         task_to_future_map.pop(job_id)
         logger.debug("dispatching callback to finish task '%s'", job_id)
-        assert event_data  # no sec
+        assert event_data  # nosec
         try:
             asyncio.run_coroutine_threadsafe(user_callback(event_data), main_loop)
         except Exception:  # pylint: disable=broad-except

services/storage/src/simcore_service_storage/dsm.py

@@ -10,8 +10,9 @@
 import tempfile
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, Final, List, Optional, Tuple, Union
 
 import aiobotocore
 import attr
@@ -61,7 +62,11 @@
 )
 from .s3wrapper.s3_client import MinioClientWrapper
 from .settings import Settings
-from .utils import download_to_file_or_raise
+from .utils import download_to_file_or_raise, is_file_entry_valid, to_meta_data_extended
+
+_MINUTE: Final[int] = 60
+_HOUR: Final[int] = 60 * _MINUTE
+
 
 logger = logging.getLogger(__name__)
 
@@ -96,17 +101,7 @@ async def _cleanup_context(app: web.Application):
     app.cleanup_ctx.append(_cleanup_context)
 
 
-def to_meta_data_extended(row: RowProxy) -> FileMetaDataEx:
-    assert row
-    meta = FileMetaData(**dict(row))  # type: ignore
-    meta_extended = FileMetaDataEx(
-        fmd=meta,
-        parent_id=str(Path(meta.object_name).parent),
-    )  # type: ignore
-    return meta_extended
-
-
-@attr.s(auto_attribs=True)
+@dataclass
 class DatCoreApiToken:
     api_token: Optional[str] = None
     api_secret: Optional[str] = None
@@ -115,7 +110,7 @@ def to_tuple(self):
         return (self.api_token, self.api_secret)
 
 
-@attr.s(auto_attribs=True)
+@dataclass
 class DataStorageManager:  # pylint: disable=too-many-public-methods
     """Data storage manager
 
@@ -154,12 +149,12 @@ class DataStorageManager:  # pylint: disable=too-many-public-methods
     pool: ThreadPoolExecutor
     simcore_bucket_name: str
     has_project_db: bool
-    session: AioSession = attr.Factory(aiobotocore.get_session)
-    datcore_tokens: Dict[str, DatCoreApiToken] = attr.Factory(dict)
+    session: AioSession = field(default_factory=aiobotocore.get_session)
+    datcore_tokens: Dict[str, DatCoreApiToken] = field(default_factory=dict)
     app: Optional[web.Application] = None
 
     def _create_aiobotocore_client_context(self) -> ClientCreatorContext:
-        assert hasattr(self.session, "create_client")
+        assert hasattr(self.session, "create_client")  # nosec
         # pylint: disable=no-member
 
         # SEE API in https://botocore.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html
@@ -217,22 +212,28 @@ async def list_files(
                 accesible_projects_ids = await get_readable_project_ids(
                     conn, int(user_id)
                 )
-                has_read_access = (
+                where_statement = (
                     file_meta_data.c.user_id == user_id
                 ) | file_meta_data.c.project_id.in_(accesible_projects_ids)
-
-                query = sa.select([file_meta_data]).where(has_read_access)
+                if uuid_filter:
+                    where_statement &= file_meta_data.c.file_uuid.ilike(
+                        f"%{uuid_filter}%"
+                    )
+                query = sa.select([file_meta_data]).where(where_statement)
 
                 async for row in conn.execute(query):
                     dex = to_meta_data_extended(row)
-                    if dex.fmd.entity_tag is None:
+                    if not is_file_entry_valid(dex.fmd):
+                        # NOTE: the file is not updated with the information from S3 backend.
+                        # 1. Either the file exists, but was never updated in the database
+                        # 2. Or the file does not exist or was never completed, and the file_meta_data entry is old and faulty
                         # we need to update from S3 here since the database is not up-to-date
-                        dex = await self.update_database_from_storage(
+                        dex = await self.try_update_database_from_storage(
                             dex.fmd.file_uuid,
                             dex.fmd.bucket_name,
                             dex.fmd.object_name,
                         )
-                    if dex and dex.fmd.entity_tag:
+                    if dex:
                         data.append(dex)
 
             if self.has_project_db:
@@ -288,6 +289,9 @@ async def list_files(
 
         elif location == DATCORE_STR:
             api_token, api_secret = self._get_datcore_tokens(user_id)
+            assert self.app  # nosec
+            assert api_secret  # nosec
+            assert api_token  # nosec
             return await datcore_adapter.list_all_datasets_files_metadatas(
                 self.app, api_token, api_secret
             )
@@ -330,6 +334,9 @@ async def list_files_dataset(
         elif location == DATCORE_STR:
             api_token, api_secret = self._get_datcore_tokens(user_id)
             # lists all the files inside the dataset
+            assert self.app  # nosec
+            assert api_secret  # nosec
+            assert api_token  # nosec
             return await datcore_adapter.list_all_files_metadatas_in_dataset(
                 self.app, api_token, api_secret, dataset_id
             )
@@ -368,6 +375,9 @@ async def list_datasets(self, user_id: str, location: str) -> List[DatasetMetaDa
 
         elif location == DATCORE_STR:
             api_token, api_secret = self._get_datcore_tokens(user_id)
+            assert self.app  # nosec
+            assert api_secret  # nosec
+            assert api_token  # nosec
             return await datcore_adapter.list_datasets(self.app, api_token, api_secret)
 
         return data
@@ -391,13 +401,14 @@ async def list_file(
                     if not row:
                         return None
                     file_metadata = to_meta_data_extended(row)
-                    if file_metadata.fmd.entity_tag is None:
-                        # we need to update from S3 here since the database is not up-to-date
-                        file_metadata = await self.update_database_from_storage(
-                            file_metadata.fmd.file_uuid,
-                            file_metadata.fmd.bucket_name,
-                            file_metadata.fmd.object_name,
-                        )
+                    if is_file_entry_valid(file_metadata.fmd):
+                        return file_metadata
+                    # we need to update from S3 here since the database is not up-to-date
+                    file_metadata = await self.try_update_database_from_storage(
+                        file_metadata.fmd.file_uuid,
+                        file_metadata.fmd.bucket_name,
+                        file_metadata.fmd.object_name,
+                    )
                     return file_metadata
                 # FIXME: returns None in both cases: file does not exist or use has no access
                 logger.debug("User %s cannot read file %s", user_id, file_uuid)
@@ -423,7 +434,7 @@ async def upload_file_to_datcore(
         # api_token, api_secret = self._get_datcore_tokens(user_id)
         # await dcw.upload_file_to_id(destination_id, local_file_path)
 
-    async def update_database_from_storage(
+    async def try_update_database_from_storage(
         self,
         file_uuid: str,
         bucket_name: str,
@@ -469,7 +480,7 @@ async def update_database_from_storage(
             return None
 
     @retry(
-        stop=stop_after_delay(3600),
+        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)
@@ -479,7 +490,7 @@ async def update_database_from_storage(
     async def auto_update_database_from_storage_task(
         self, file_uuid: str, bucket_name: str, object_name: str
     ):
-        return await self.update_database_from_storage(
+        return await self.try_update_database_from_storage(
             file_uuid, bucket_name, object_name, silence_exception=True
         )
 
@@ -573,6 +584,9 @@ async def download_link_s3(self, file_uuid: str, user_id: int) -> str:
 
     async def download_link_datcore(self, user_id: str, file_id: str) -> URL:
         api_token, api_secret = self._get_datcore_tokens(user_id)
+        assert self.app  # nosec
+        assert api_secret  # nosec
+        assert api_token  # nosec
         return await datcore_adapter.get_file_download_presigned_link(
             self.app, api_token, api_secret, file_id
         )
@@ -928,6 +942,9 @@ async def delete_file(self, user_id: str, location: str, file_uuid: str):
         elif location == DATCORE_STR:
             # FIXME: review return inconsistencies
             api_token, api_secret = self._get_datcore_tokens(user_id)
+            assert self.app  # nosec
+            assert api_secret  # nosec
+            assert api_token  # nosec
             await datcore_adapter.delete_file(
                 self.app, api_token, api_secret, file_uuid
             )