226 bug scicat dataset is not reset in case of failure (#227)

Author: phwissmann

backend/api/.dockerignore

@@ -1,4 +1,4 @@
 .venv
-__pycache__
+**/__pycache__
 .pytest_cache
 .vscode

backend/archiver/.vscode/settings.json

@@ -2,6 +2,17 @@
   "python.testing.pytestArgs": [
     "./"
   ],
+  "python.testing.pytestPath": ".",
   "python.testing.unittestEnabled": false,
-  "python.testing.pytestEnabled": true
+  "python.testing.pytestEnabled": true,
+  "python.languageServer": "Pylance",
+  "cSpell.words": [
+    "datablock",
+    "datablocks",
+    "datasetlist",
+    "origdatablocks",
+    "scicat"
+  ],
+  "cSpell.enabled": false,
+  "cSpell.diagnosticLevel": "Hint",
 }

backend/archiver/config/variables.py

@@ -2,7 +2,7 @@
 from prefect.variables import Variable
 import os
 
-from archiver.utils.log import getLogger
+from utils.log import getLogger
 
 from pydantic_settings import (
     BaseSettings,

backend/archiver/flows/__main__.py

@@ -1,8 +1,8 @@
 from prefect import serve
 
-from .archive_datasets_flow import archive_datasets_flow
-from .retrieve_datasets_flow import retrieve_datasets_flow
-from .mock_flows import create_test_dataset_flow, end_to_end_test_flow
+from archive_datasets_flow import archive_datasets_flow
+from retrieve_datasets_flow import retrieve_datasets_flow
+from mock_flows import create_test_dataset_flow, end_to_end_test_flow
 
 
 if __name__ == "__main__":

backend/archiver/flows/archive_datasets_flow.py

@@ -16,35 +16,37 @@
     update_progress_artifact,
 )
 
-from archiver.config.variables import Variables
-from archiver.utils.datablocks import ArchiveInfo
+from config.variables import Variables
+from utils.datablocks import ArchiveInfo
 
-from .utils import StoragePaths, report_archival_error
+from .flow_utils import StoragePaths, report_archival_error
 from .task_utils import (
     generate_task_name_dataset,
     generate_flow_name_job_id,
     generate_subflow_run_name_job_id_dataset_id,
     generate_sleep_for_task_name
 )
-from archiver.scicat.scicat_interface import SciCatClient
-from archiver.scicat.scicat_tasks import (
+from scicat.scicat_interface import SciCatClient
+from scicat.scicat_tasks import (
     update_scicat_archival_job_status,
     update_scicat_archival_dataset_lifecycle,
     get_origdatablocks,
     register_datablocks,
     get_scicat_access_token,
     get_job_datasetlist,
+    reset_dataset
 )
-from archiver.scicat.scicat_tasks import (
+from scicat.scicat_tasks import (
     report_job_failure_system_error,
-    report_dataset_user_error,
+    report_dataset_user_error
 )
-from archiver.utils.datablocks import wait_for_free_space
-from archiver.utils.model import OrigDataBlock, DataBlock
-import archiver.utils.datablocks as datablocks_operations
-from archiver.config.concurrency_limits import ConcurrencyLimits
-from archiver.utils.s3_storage_interface import Bucket, get_s3_client
-from archiver.utils.log import getLogger
+
+from utils.datablocks import wait_for_free_space
+from utils.model import OrigDataBlock, DataBlock
+import utils.datablocks as datablocks_operations
+from config.concurrency_limits import ConcurrencyLimits
+from utils.s3_storage_interface import Bucket, get_s3_client
+from utils.log import getLogger
 
 
 def on_get_origdatablocks_error(dataset_id: str, task: Task, task_run: TaskRun, state: State):
@@ -108,28 +110,6 @@ def update_progress(p):
     return file_paths
 
 
-# @task(task_run_name=generate_task_name_dataset)
-# def create_datablocks(dataset_id: str, origDataBlocks: List[OrigDataBlock], file_paths: List[Path]) -> List[DataBlock]:
-#     """Prefect task to create datablocks.
-
-#     Args:
-#         dataset_id (str): dataset id
-#         origDataBlocks (List[OrigDataBlock]): List of OrigDataBlocks (Pydantic Model)
-
-#     Returns:
-#         List[DataBlock]: List of DataBlocks (Pydantic Model)
-#     """
-
-#     s3_client = get_s3_client()
-
-#     progress_artifact_id = create_progress_artifact(
-#         progress=0.0,
-#         description="Create datablocks from datafiles",
-#     )
-
-#     return datablocks_operations.create_datablocks(s3_client, dataset_id, origDataBlocks, file_paths, update_progress)
-
-
 @task(task_run_name=generate_task_name_dataset)
 def create_tarfiles(dataset_id: str) -> List[ArchiveInfo]:
     datablocks_scratch_folder = StoragePaths.scratch_archival_datablocks_folder(dataset_id)
@@ -388,22 +368,29 @@ def on_dataset_flow_failure(flow: Flow, flow_run: FlowRun, state: State):
         task_run=None,
         token=scicat_token,
     )
+    try:
+        reset_dataset(
+            dataset_id=flow_run.parameters["dataset_id"],
+            token=scicat_token
+        )
+    except Exception as e:
+        getLogger().error(f"failed to reset datablocks {e}")
     datablocks_operations.cleanup_lts_folder(flow_run.parameters["dataset_id"])
     datablocks_operations.cleanup_scratch(flow_run.parameters["dataset_id"])
     try:
         s3_client = get_s3_client()
         datablocks_operations.cleanup_s3_staging(s3_client, flow_run.parameters["dataset_id"])
-    except:
-        pass
+    except Exception as e:
+        getLogger().error(f"failed to cleanup staging {e}")
 
 
 def cleanup_dataset(flow: Flow, flow_run: FlowRun, state: State):
     try:
         s3_client = get_s3_client()
         datablocks_operations.cleanup_s3_landingzone(s3_client, flow_run.parameters["dataset_id"])
         datablocks_operations.cleanup_s3_staging(s3_client, flow_run.parameters["dataset_id"])
-    except:
-        pass
+    except Exception as e:
+        getLogger().error(f"failed to cleanup staging or landingzone {e}")
     datablocks_operations.cleanup_scratch(flow_run.parameters["dataset_id"])
 
 

backend/archiver/flows/utils.py backend/archiver/flows/flow_utils.pybackend/archiver/flows/utils.py renamed to backend/archiver/flows/flow_utils.py

@@ -4,11 +4,11 @@
 from prefect import State
 from prefect.client.schemas.objects import TaskRun
 
-from archiver.config.variables import Variables
-from archiver.scicat.scicat_tasks import (
+from config.variables import Variables
+from scicat.scicat_tasks import (
     report_dataset_system_error,
     report_dataset_user_error,
-    report_dataset_retrieval_error,
+    report_dataset_retrieval_error
 )
 
 
@@ -45,7 +45,7 @@ def report_archival_error(dataset_id: str, state: State, task_run: TaskRun, toke
 
 
 def report_retrieval_error(dataset_id: str, state: State, task_run: TaskRun, token: SecretStr):
-    """Report a retrieval error of a job of a dataset. Differntiates betwen "DatasetError" (User error, e.g. missing files)
+    """Report a retrieval error of a job of a dataset. Differentiates between "DatasetError" (User error, e.g. missing files)
     and SystemError (transient error).
 
     Args:

backend/archiver/flows/mock_flows.py

@@ -15,15 +15,15 @@
 import urllib
 import asyncio
 
-from archiver.config.variables import Variables
-from archiver.utils.datablocks import upload_objects_to_s3
-from archiver.utils.s3_storage_interface import Bucket, get_s3_client
-from archiver.utils.model import OrigDataBlock, DataFile, Dataset, DatasetLifecycle
-
-from archiver.utils.log import getLogger, log
-from archiver.flows.utils import StoragePaths
-from archiver.scicat.scicat_tasks import get_scicat_access_token
-from archiver.utils.model import DatasetListEntry, Job
+from config.variables import Variables
+from utils.datablocks import upload_objects_to_s3
+from utils.s3_storage_interface import Bucket, get_s3_client
+from utils.model import OrigDataBlock, DataFile, Dataset, DatasetLifecycle
+
+from utils.log import getLogger, log
+from .flow_utils import StoragePaths
+from scicat.scicat_tasks import get_scicat_access_token
+from utils.model import DatasetListEntry, Job
 from .task_utils import generate_task_name_dataset
 
 from prefect.flow_runs import wait_for_flow_run

backend/archiver/flows/retrieve_datasets_flow.py

@@ -10,28 +10,29 @@
 from prefect.client.schemas.filters import FlowRunFilter
 from prefect.flow_runs import wait_for_flow_run
 from prefect.context import get_run_context
-from archiver.utils.s3_storage_interface import get_s3_client
-from archiver.utils.s3_storage_interface import Bucket
+from utils.s3_storage_interface import get_s3_client
+from utils.s3_storage_interface import Bucket
 
 
 from .task_utils import generate_flow_name_dataset, generate_flow_name_job_id, generate_task_name_dataset, generate_task_name_datablock
-from .utils import report_retrieval_error
-from archiver.scicat.scicat_interface import SciCatClient
-from archiver.utils.model import DataBlock, JobResultObject
-from archiver.scicat.scicat_tasks import (
+from .flow_utils import report_retrieval_error
+from scicat.scicat_interface import SciCatClient
+from utils.model import DataBlock, JobResultObject
+from scicat.scicat_tasks import (
     update_scicat_retrieval_job_status,
     update_scicat_retrieval_dataset_lifecycle,
     get_scicat_access_token,
     get_job_datasetlist,
     create_job_result_object_task,
 )
-from archiver.scicat.scicat_tasks import (
+from scicat.scicat_tasks import (
     report_job_failure_system_error,
     report_dataset_user_error,
     get_datablocks,
 )
-from archiver.config.concurrency_limits import ConcurrencyLimits
-import archiver.utils.datablocks as datablocks_operations
+from config.concurrency_limits import ConcurrencyLimits
+from config.variables import Variables
+import utils.datablocks as datablocks_operations
 
 
 def on_get_datablocks_error(dataset_id: str, task: Task, task_run: TaskRun, state: State):
@@ -42,8 +43,7 @@ def on_get_datablocks_error(dataset_id: str, task: Task, task_run: TaskRun, stat
 @task(
     task_run_name=generate_task_name_datablock,
     tags=[ConcurrencyLimits().LTS_READ_TAG],
-      retries=5,
-      retry_delay_seconds=[60, 120, 240, 480, 960]
+    retry_delay_seconds=[60, 120, 240, 480, 960]
 )
 def copy_datablock_from_LTS_to_scratch(dataset_id: str, datablock: DataBlock):
     datablocks_operations.copy_from_LTS_to_scratch_retrieval(dataset_id, datablock)

backend/archiver/flows/tests/helpers.py

@@ -1,18 +1,18 @@
 from typing import List, Dict, Any
 
 from pydantic import SecretStr
-from archiver.utils.model import DataFile, OrigDataBlock, DataBlock
-from archiver.utils.model import (
+from utils.model import DataFile, OrigDataBlock, DataBlock
+from utils.model import (
     Job,
     Dataset,
     DatasetLifecycle,
     JobResultObject,
     JobResultEntry,
 )
-from archiver.scicat.scicat_interface import SciCatClient
+from scicat.scicat_interface import SciCatClient
 from pathlib import Path
 
-from archiver.utils.s3_storage_interface import S3Storage
+from utils.s3_storage_interface import S3Storage
 
 
 def mock_s3client() -> S3Storage:

backend/archiver/flows/tests/scicat_unittest_mock.py

@@ -3,9 +3,9 @@
 from uuid import UUID
 import urllib.parse
 
-from archiver.scicat.scicat_interface import SciCatClient
-from archiver.utils.model import OrigDataBlock, DataBlock
-from archiver.utils.model import DatasetListEntry, Job
+from scicat.scicat_interface import SciCatClient
+from utils.model import OrigDataBlock, DataBlock
+from utils.model import DatasetListEntry, Job
 
 
 def mock_scicat_get_token() -> str:
@@ -64,23 +64,30 @@ def __init__(
         )
 
         json_list = []
-        for o in origDataBlocks:
-            json_list.append(o.model_dump_json())
+        for d in origDataBlocks:
+            json_list.append(d.model_dump_json())
 
         self.matchers["origdatablocks"] = self.get(
             f"{self.ENDPOINT}{self.API_PREFIX}/datasets/{safe_dataset_url}/origdatablocks",
             json=json_list,
         )
 
         json_list = []
-        for o in datablocks:
-            json_list.append(o.model_dump_json())
+        for d in datablocks:
+            json_list.append(d.model_dump_json())
 
         self.matchers["get_datablocks"] = self.get(
             f"{self.ENDPOINT}{self.API_PREFIX}/datasets/{safe_dataset_url}/datablocks",
             json=json_list,
         )
 
+        self.matchers["delete_datablocks"] = []
+
+        for d in datablocks:
+            self.matchers["delete_datablocks"].append(self.delete(
+                f"{self.ENDPOINT}{self.API_PREFIX}/datasets/{safe_dataset_url}/datablocks/{d.id}")
+            )
+
     @property
     def jobs_matcher(self):
         return self.matchers["jobs"]
@@ -97,6 +104,10 @@ def datablocks_post_matcher(self):
     def datablocks_get_matcher(self):
         return self.matchers["get_datablocks"]
 
+    @property
+    def datablocks_delete_matcher(self):
+        return self.matchers["delete_datablocks"]
+
     @property
     def origdatablocks_matcher(self):
         return self.matchers["origdatablocks"]