Bugfix/clean storage meta data table (ITISFoundation#2415)

* new entrypoint to trigger synchronisation of the table with S3

Author: sanderegg

api/specs/storage/openapi.yaml

@@ -97,6 +97,32 @@ paths:
         default:
           $ref: "#/components/responses/DefaultErrorResponse"
 
+  /locations/{location_id}:sync:
+    post:
+      summary: Manually triggers the synchronisation of the file meta data table in the database
+      operationId: synchronise_meta_data_table
+      parameters:
+        - name: location_id
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: dry_run
+          in: query
+          required: false
+          schema:
+            type: boolean
+            default: true
+      responses:
+        "200":
+          description: An object containing added, changed and removed paths
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/TableSynchronisationEnveloped"
+        default:
+          $ref: "#/components/responses/DefaultErrorResponse"
+
   /locations/{location_id}/datasets:
     get:
       summary: Lists all dataset's metadata
@@ -628,6 +654,28 @@ components:
           key1: value1
           key2: value2
 
+    TableSynchronisationEnveloped:
+      type: object
+      required:
+        - data
+        - error
+      properties:
+        data:
+          $ref: "#/components/schemas/TableSynchronisation"
+        error:
+          nullable: true
+          default: null
+
+    TableSynchronisation:
+      type: object
+      required:
+        - removed
+      properties:
+        removed:
+          type: array
+          items:
+            type: string
+
     FileLocationArrayEnveloped:
       type: object
       required:

api/specs/webserver/components/schemas/locations.yaml

@@ -4,7 +4,7 @@ FileLocationEnveloped:
     - data
   properties:
     data:
-      $ref: '#/FileLocation'
+      $ref: "#/FileLocation"
     error:
       nullable: true
       default: null
@@ -17,10 +17,32 @@ FileLocation:
     id:
       type: number
   example:
-    filename: 'simcore.s3'
+    filename: "simcore.s3"
     id: 0
 
 FileLocationArray:
   type: array
   items:
-    $ref: '#/FileLocation'
+    $ref: "#/FileLocation"
+
+TableSynchronisationEnveloped:
+  type: object
+  required:
+    - data
+    - error
+  properties:
+    data:
+      $ref: "#/TableSynchronisation"
+    error:
+      nullable: true
+      default: null
+
+TableSynchronisation:
+  type: object
+  required:
+    - removed
+  properties:
+    removed:
+      type: array
+      items:
+        type: string

api/specs/webserver/openapi-storage.yaml

@@ -15,6 +15,34 @@ paths:
         default:
           $ref: "#/components/responses/DefaultErrorResponse"
 
+  /storage/locations/{location_id}:sync:
+    post:
+      summary: Manually triggers the synchronisation of the file meta data table in the database
+      tags:
+        - storage
+      operationId: synchronise_meta_data_table
+      parameters:
+        - name: location_id
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: dry_run
+          in: query
+          required: false
+          schema:
+            type: boolean
+            default: true
+      responses:
+        "200":
+          description: An object containing added, changed and removed paths
+          content:
+            application/json:
+              schema:
+                $ref: "./components/schemas/locations.yaml#/TableSynchronisationEnveloped"
+        default:
+          $ref: "#/components/responses/DefaultErrorResponse"
+
   /storage/locations/{location_id}/datasets:
     get:
       summary: Get datasets metadata

api/specs/webserver/openapi.yaml

@@ -139,6 +139,9 @@ paths:
   /storage/locations:
     $ref: "./openapi-storage.yaml#/paths/~1storage~1locations"
 
+  /storage/locations/{location_id}:sync:
+    $ref: "./openapi-storage.yaml#/paths/~1storage~1locations~1{location_id}:sync"
+
   /storage/locations/{location_id}/files/metadata:
     $ref: "./openapi-storage.yaml#/paths/~1storage~1locations~1{location_id}~1files~1metadata"
 

services/storage/src/simcore_service_storage/api/v0/openapi.yaml

@@ -95,6 +95,31 @@ paths:
                 $ref: '#/components/schemas/FileLocationArrayEnveloped'
         default:
           $ref: '#/components/responses/DefaultErrorResponse'
+  '/locations/{location_id}:sync':
+    post:
+      summary: Manually triggers the synchronisation of the file meta data table in the database
+      operationId: synchronise_meta_data_table
+      parameters:
+        - name: location_id
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: dry_run
+          in: query
+          required: false
+          schema:
+            type: boolean
+            default: true
+      responses:
+        '200':
+          description: 'An object containing added, changed and removed paths'
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/TableSynchronisationEnveloped'
+        default:
+          $ref: '#/components/responses/DefaultErrorResponse'
   '/locations/{location_id}/datasets':
     get:
       summary: Lists all dataset's metadata
@@ -597,6 +622,26 @@ components:
         body_value:
           key1: value1
           key2: value2
+    TableSynchronisationEnveloped:
+      type: object
+      required:
+        - data
+        - error
+      properties:
+        data:
+          $ref: '#/components/schemas/TableSynchronisation'
+        error:
+          nullable: true
+          default: null
+    TableSynchronisation:
+      type: object
+      required:
+        - removed
+      properties:
+        removed:
+          type: array
+          items:
+            type: string
     FileLocationArrayEnveloped:
       type: object
       required:

services/storage/src/simcore_service_storage/dsm.py

@@ -1009,3 +1009,42 @@ async def create_soft_link(
             result = await conn.execute(stmt)
             link = to_meta_data_extended(await result.first())
             return link
+
+    async def synchronise_meta_data_table(
+        self, location: str, dry_run: bool
+    ) -> Dict[str, Any]:
+        sync_results = {"removed": []}
+        if location == SIMCORE_S3_STR:
+            # NOTE: only valid for Simcore, since datcore data is not in the database table
+            # let's get all the files in the table
+            logger.warning(
+                "synchronisation of database/s3 storage started, this will take some time..."
+            )
+            async with self.engine.acquire() as conn, self._create_client_context() as s3_client:
+                number_of_rows_in_db = await conn.scalar(file_meta_data.count())
+                logger.warning(
+                    "total number of entries to check %d",
+                    number_of_rows_in_db,
+                )
+
+                async for row in conn.execute(file_meta_data.select()):
+                    s3_key = row.object_name  # type: ignore
+
+                    # now check if the file exists in S3
+                    try:
+                        await s3_client.get_object(
+                            Bucket=self.simcore_bucket_name,
+                            Key=s3_key,
+                        )
+                    except s3_client.exceptions.NoSuchKey:
+                        # this file does not exist
+                        sync_results["removed"].append(s3_key)
+
+                if not dry_run:
+                    await conn.execute(
+                        file_meta_data.delete().where(
+                            file_meta_data.c.object_name.in_(sync_results["removed"])
+                        )
+                    )
+
+        return sync_results

services/storage/src/simcore_service_storage/handlers.py

@@ -1,7 +1,7 @@
 import json
 import logging
 from contextlib import contextmanager
-from typing import Dict
+from typing import Any, Dict
 
 import attr
 from aiohttp import web
@@ -219,6 +219,24 @@ async def get_file_metadata(request: web.Request):
         }
 
 
+@routes.post(f"/{api_vtag}/locations/{{location_id}}:sync")  # type: ignore
+async def synchronise_meta_data_table(request: web.Request) -> Dict[str, Any]:
+    params, query, *_ = await extract_and_validate(request)
+    assert query["dry_run"] is not None  # nosec
+    assert params["location_id"]  # nosec
+
+    with handle_storage_errors():
+        location_id = params["location_id"]
+        dry_run = query["dry_run"]
+        dsm = await _prepare_storage_manager(params, query, request)
+        location = dsm.location_from_id(location_id)
+        data_changed: Dict[str, Any] = await dsm.synchronise_meta_data_table(
+            location, dry_run
+        )
+
+    return {"error": None, "data": data_changed}
+
+
 # DISABLED: @routes.patch(f"/{api_vtag}/locations/{{location_id}}/files/{{fileId}}/metadata") # type: ignore
 async def update_file_meta_data(request: web.Request):
     params, query, body = await extract_and_validate(request)

services/storage/tests/conftest.py

@@ -13,7 +13,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from random import randrange
-from typing import Dict, Iterator, Tuple
+from typing import Any, Dict, Iterator, Tuple
 
 import dotenv
 import pytest
@@ -188,7 +188,7 @@ async def postgres_engine(loop, postgres_service_url):
 
 
 @pytest.fixture(scope="session")
-def minio_service(docker_services, docker_ip):
+def minio_service(docker_services, docker_ip) -> Dict[str, Any]:
 
     # Build URL to service listening on random port.
     url = "http://%s:%d/" % (
@@ -215,7 +215,7 @@ def minio_service(docker_services, docker_ip):
 
 
 @pytest.fixture(scope="module")
-def s3_client(minio_service):
+def s3_client(minio_service: Dict[str, Any]) -> MinioClientWrapper:
 
     s3_client = MinioClientWrapper(
         endpoint=minio_service["endpoint"],

services/storage/tests/unit/test_dsm.py

@@ -16,12 +16,15 @@
 import uuid
 from pathlib import Path
 from shutil import copyfile
+from typing import Any, Dict, Tuple
 
 import attr
 import pytest
 import tests.utils
 from simcore_service_storage.constants import DATCORE_STR, SIMCORE_S3_ID, SIMCORE_S3_STR
+from simcore_service_storage.dsm import DataStorageManager
 from simcore_service_storage.models import FileMetaData
+from simcore_service_storage.s3wrapper.s3_client import MinioClientWrapper
 from tests.utils import BUCKET_NAME, USER_ID, has_datcore_tokens
 
 
@@ -592,6 +595,46 @@ async def test_dsm_list_datasets_s3(dsm_fixture, dsm_mockup_complete_db):
     assert any("Kember" in d.display_name for d in datasets)
 
 
+async def test_sync_table_meta_data(
+    dsm_fixture: DataStorageManager,
+    dsm_mockup_complete_db: Tuple[Dict[str, str], Dict[str, str]],
+    s3_client: MinioClientWrapper,
+):
+    dsm_fixture.has_project_db = True
+
+    expected_removed_files = []
+    # the list should be empty on start
+    list_changes: Dict[str, Any] = await dsm_fixture.synchronise_meta_data_table(
+        location=SIMCORE_S3_STR, dry_run=True
+    )
+    assert "removed" in list_changes
+    assert list_changes["removed"] == expected_removed_files
+
+    # now remove the files
+    for file_entry in dsm_mockup_complete_db:
+        s3_key = f"{file_entry['project_id']}/{file_entry['node_id']}/{file_entry['filename']}"
+        s3_client.remove_objects(BUCKET_NAME, [s3_key])
+        expected_removed_files.append(s3_key)
+
+        # the list should now contain the removed entries
+        list_changes: Dict[str, Any] = await dsm_fixture.synchronise_meta_data_table(
+            location=SIMCORE_S3_STR, dry_run=True
+        )
+        assert "removed" in list_changes
+        assert list_changes["removed"] == expected_removed_files
+
+    # now effectively call the function should really remove the files
+    list_changes: Dict[str, Any] = await dsm_fixture.synchronise_meta_data_table(
+        location=SIMCORE_S3_STR, dry_run=False
+    )
+    # listing again will show an empty list again
+    list_changes: Dict[str, Any] = await dsm_fixture.synchronise_meta_data_table(
+        location=SIMCORE_S3_STR, dry_run=True
+    )
+    assert "removed" in list_changes
+    assert list_changes["removed"] == []
+
+
 @pytest.mark.skipif(not has_datcore_tokens(), reason="no datcore tokens")
 async def test_dsm_list_datasets_datcore(dsm_fixture, datcore_structured_testbucket):
     datasets = await dsm_fixture.list_datasets(user_id=USER_ID, location=DATCORE_STR)

services/storage/tests/utils.py

@@ -19,7 +19,7 @@
 log = logging.getLogger(__name__)
 
 
-DATABASE = "aio_login_tests"
+DATABASE = "test"
 USER = "admin"
 PASS = "admin"