use_multi_tenancy_endpoints_II: tests running (#208)

* use_multi_tenancy_endpoints_II: tests running

* use_multi_tenancy_endpoints_II: use bump manifesto model in rollback

* use_multi_tenancy_endpoints_II: format

Author: DanielElisenberg

job_executor/adapter/datastore_api/__init__.py

@@ -105,6 +105,12 @@ def is_system_paused() -> bool:
     return maintenance_status.paused
 
 
+def get_datastores() -> list[str]:
+    """Get a list of all datastore rdns on this tenant"""
+    request_url = f"{DATASTORE_API_URL}/datastores"
+    return execute_request("GET", request_url, True).json()
+
+
 def get_datastore_directory(rdn: str) -> Path:
     request_url = f"{DATASTORE_API_URL}/datastores/{rdn}"
     response = execute_request("GET", request_url, True)

job_executor/adapter/fs/datastore_files.py

@@ -240,7 +240,7 @@ def save_temporary_backup(self) -> None:
             draft_version = json.load(f)
         with open(self.draft_metadata_all_path, encoding="utf-8") as f:
             metadata_all_draft = json.load(f)
-        tmp_dir = self.root_dir / "tmp"
+        tmp_dir = self.metadata_dir / "tmp"
         if os.path.isdir(tmp_dir):
             raise LocalStorageError("tmp directory already exists")
         os.mkdir(tmp_dir)
@@ -263,7 +263,7 @@ def restore_from_temporary_backup(self) -> str | None:
         Returns None if no released version in backup, else returns the
         latest release version number as dotted four part version.
         """
-        tmp_dir = Path(self.root_dir) / "tmp"
+        tmp_dir = self.metadata_dir / "tmp"
         draft_version_backup = tmp_dir / "draft_version.json"
         metadata_all_draft_backup = tmp_dir / "metadata_all__DRAFT.json"
         datastore_versions_backup = tmp_dir / "datastore_versions.json"
@@ -294,10 +294,10 @@ def archive_temporary_backup(self) -> None:
         exists. Raises `LocalStorageError` if there are any unrecognized files
         in the directory.
         """
-        tmp_dir = Path(self.root_dir) / "tmp"
+        tmp_dir = self.metadata_dir / "tmp"
         os.makedirs(self.archive_dir, exist_ok=True)
 
-        if not os.path.isdir(Path(self.root_dir) / "tmp"):
+        if not os.path.isdir(tmp_dir):
             raise LocalStorageError(
                 "Could not find a tmp directory to archive."
             )
@@ -312,20 +312,18 @@ def archive_temporary_backup(self) -> None:
                     "directory. Aborting tmp archiving."
                 )
         timestamp = datetime.now(UTC).replace(tzinfo=None)
-        shutil.move(
-            self.root_dir / "tmp", self.archive_dir / f"tmp_{timestamp}"
-        )
+        shutil.move(tmp_dir, self.archive_dir / f"tmp_{timestamp}")
 
     def delete_temporary_backup(self) -> None:
         """
         Deletes the tmp directory within the datastore if the directory
         exists. Raises `LocalStorageError` if there are any unrecognized files
         in the directory.
         """
-        tmp_dir = Path(self.root_dir) / "tmp"
+        tmp_dir = self.metadata_dir / "tmp"
         os.makedirs(self.archive_dir, exist_ok=True)
 
-        if not os.path.isdir(Path(self.root_dir) / "tmp"):
+        if not os.path.isdir(tmp_dir):
             raise LocalStorageError("Could not find a tmp directory to delete.")
         for content in os.listdir(tmp_dir):
             if content not in [
@@ -337,13 +335,13 @@ def delete_temporary_backup(self) -> None:
                     "Found unrecognized files and/or directories in the tmp "
                     "directory. Aborting tmp deleting."
                 )
-        shutil.rmtree(self.root_dir / "tmp")
+        shutil.rmtree(tmp_dir)
 
     def temporary_backup_exists(self) -> bool:
         """
         Returns a boolean representing if the tmp directory exists.
         """
-        tmp_dir = self.root_dir / "tmp"
+        tmp_dir = self.metadata_dir / "tmp"
         return os.path.isdir(tmp_dir)
 
     def archive_draft_version(self, version: str) -> None:

job_executor/app.py

@@ -18,10 +18,12 @@
 
 def initialize_app() -> None:
     try:
-        local_storage = LocalStorageAdapter(Path(environment.datastore_dir))
         rollback.fix_interrupted_jobs()
-        if local_storage.datastore_dir.temporary_backup_exists():
-            raise StartupException("tmp directory exists")
+        rdns = datastore_api.get_datastores()
+        for rdn in rdns:
+            local_storage = LocalStorageAdapter(Path(environment.datastore_dir))
+            if local_storage.datastore_dir.temporary_backup_exists():
+                raise StartupException(f"tmp directory exists for {rdn}")
     except Exception as e:
         raise StartupException("Exception when initializing") from e
 

job_executor/domain/datastores.py

@@ -214,16 +214,12 @@ def patch_metadata(
     except PatchingError as e:
         logger.error(f"{job.job_id}: Patching error occured")
         logger.exception(f"{job.job_id}: {str(e)}", exc_info=e)
-        rollback_manager_phase_import_job(
-            job.job_id, "PATCH_METADATA", dataset_name
-        )
+        rollback_manager_phase_import_job(job, "PATCH_METADATA", dataset_name)
         datastore_api.update_job_status(job.job_id, JobStatus.FAILED, str(e))
     except Exception as e:
         logger.error(f"{job.job_id}: An unexpected error occured")
         logger.exception(f"{job.job_id}: {str(e)}", exc_info=e)
-        rollback_manager_phase_import_job(
-            job.job_id, "PATCH_METADATA", dataset_name
-        )
+        rollback_manager_phase_import_job(job, "PATCH_METADATA", dataset_name)
         datastore_api.update_job_status(job.job_id, JobStatus.FAILED)
 
 
@@ -273,7 +269,7 @@ def add(
     except Exception as e:
         logger.error(f"{job.job_id}: An unexpected error occured")
         logger.exception(f"{job.job_id}: {str(e)}", exc_info=e)
-        rollback_manager_phase_import_job(job.job_id, "ADD", dataset_name)
+        rollback_manager_phase_import_job(job, "ADD", dataset_name)
         datastore_api.update_job_status(job.job_id, JobStatus.FAILED)
 
 
@@ -325,7 +321,7 @@ def change(
     except Exception as e:
         logger.error(f"{job.job_id}: An unexpected error occured")
         logger.exception(f"{job.job_id}: {str(e)}", exc_info=e)
-        rollback_manager_phase_import_job(job.job_id, "CHANGE", dataset_name)
+        rollback_manager_phase_import_job(job, "CHANGE", dataset_name)
         datastore_api.update_job_status(job.job_id, JobStatus.FAILED)
 
 
@@ -564,8 +560,8 @@ def bump_version(
         logger.error(f"{job.job_id}: An unexpected error occured")
         logger.exception(f"{job.job_id}: {str(e)}", exc_info=e)
         rollback_bump(
-            job.job_id,
-            bump_manifesto.model_dump(by_alias=True, exclude_none=True),
+            job,
+            bump_manifesto,
         )
         datastore_api.update_job_status(job.job_id, JobStatus.FAILED)
 

job_executor/domain/rollback.py

@@ -4,7 +4,11 @@
 from pathlib import Path
 
 from job_executor.adapter import datastore_api
-from job_executor.adapter.datastore_api.models import Job, JobStatus
+from job_executor.adapter.datastore_api.models import (
+    DatastoreVersion,
+    Job,
+    JobStatus,
+)
 from job_executor.adapter.fs import LocalStorageAdapter
 from job_executor.adapter.fs.models.datastore_versions import (
     bump_dotted_version_number,
@@ -16,28 +20,30 @@
     RollbackException,
     StartupException,
 )
-from job_executor.config import environment
 
-WORKING_DIR_PATH = Path(environment.datastore_dir + "_working")
 logger = logging.getLogger()
 
 
-def rollback_bump(job_id: str, bump_manifesto: dict) -> None:
-    local_storage = LocalStorageAdapter(Path(environment.datastore_dir))
+def rollback_bump(job: Job, bump_manifesto: DatastoreVersion) -> None:
+    job_id = job.job_id
+    local_storage = LocalStorageAdapter(
+        datastore_api.get_datastore_directory(job.datastore_rdn)
+    )
     try:
         logger.info(f"{job_id}: Restoring files from temporary backup")
         restored_version_number = (
             local_storage.datastore_dir.restore_from_temporary_backup()
         )
-        update_type = bump_manifesto["updateType"]
-        bumped_version_number = (
-            "1.0.0.0"
-            if restored_version_number is None
-            else bump_dotted_version_number(
+        bumped_version_number = "1.0.0.0"
+        update_type = bump_manifesto.update_type
+        if restored_version_number is not None:
+            assert update_type is not None
+            bumped_version_number = bump_dotted_version_number(
                 underscored_to_dotted_version(restored_version_number),
                 update_type,
             )
-        )
+        else:
+            update_type = "MAJOR"
         logger.warning(
             f"{job_id}: Rolling back to {restored_version_number} "
             f"from bump to {bumped_version_number}"
@@ -47,9 +53,9 @@ def rollback_bump(job_id: str, bump_manifesto: dict) -> None:
         )
         bumped_version_data = "_".join(bumped_version_metadata.split("_")[:-1])
         manifesto_datasets = [
-            dataset["name"]
-            for dataset in bump_manifesto["dataStructureUpdates"]
-            if dataset["releaseStatus"] != "DRAFT"
+            dataset.name
+            for dataset in bump_manifesto.data_structure_updates
+            if dataset.release_status != "DRAFT"
         ]
         logger.info(
             f"{job_id}: Found {len(manifesto_datasets)} "
@@ -127,9 +133,12 @@ def rollback_bump(job_id: str, bump_manifesto: dict) -> None:
 
 
 def rollback_worker_phase_import_job(
-    job_id: str, operation: str, dataset_name: str
+    job: Job, operation: str, dataset_name: str
 ) -> None:
-    local_storage = LocalStorageAdapter(Path(environment.datastore_dir))
+    job_id = job.job_id
+    local_storage = LocalStorageAdapter(
+        datastore_api.get_datastore_directory(job.datastore_rdn)
+    )
     logger.warning(
         f"{job_id}: Rolling back worker job "
         f'with target: "{dataset_name}" and operation "{operation}"'
@@ -154,17 +163,19 @@ def rollback_worker_phase_import_job(
 
     if operation in ["ADD", "CHANGE"]:
         for file in generated_data_files:
-            filepath = WORKING_DIR_PATH / file
+            filepath = local_storage.working_dir.path / file
             if filepath.exists():
                 logger.info(f'{job_id}: Deleting data file "{filepath}"')
                 os.remove(filepath)
-        parquet_directory = WORKING_DIR_PATH / generated_data_directory
+        parquet_directory = (
+            local_storage.working_dir.path / generated_data_directory
+        )
         if parquet_directory.exists() and os.path.isdir(parquet_directory):
             logger.info(
                 f'{job_id}: Deleting data directory "{parquet_directory}"'
             )
             shutil.rmtree(parquet_directory)
-        dataset_directory = WORKING_DIR_PATH / dataset_name
+        dataset_directory = local_storage.working_dir.path / dataset_name
         if dataset_directory.exists() and os.path.isdir(dataset_directory):
             logger.info(
                 f'{job_id}: Deleting dataset directory "{dataset_directory}"'
@@ -173,14 +184,17 @@ def rollback_worker_phase_import_job(
 
 
 def rollback_manager_phase_import_job(
-    job_id: str, operation: str, dataset_name: str
+    job: Job, operation: str, dataset_name: str
 ) -> None:
     """
     Rolls back manager phase import job.
     Exceptions are not handled here on purpose. It is a catastrophic thing
     if a rollback fails.
     """
-    local_storage = LocalStorageAdapter(Path(environment.datastore_dir))
+    job_id = job.job_id
+    local_storage = LocalStorageAdapter(
+        datastore_api.get_datastore_directory(job.datastore_rdn)
+    )
     logger.warning(
         f"{job_id}: Rolling back import job "
         f'with target: "{dataset_name}" and operation "{operation}"'
@@ -218,7 +232,7 @@ def fix_interrupted_job(job: Job) -> None:
     if job_operation in ["ADD", "CHANGE", "PATCH_METADATA"]:
         if job.status == "importing":
             rollback_manager_phase_import_job(
-                job.job_id, job_operation, job.parameters.target
+                job, job_operation, job.parameters.target
             )
             logger.info(
                 f"{job.job_id}: Rolled back importing of job with "
@@ -232,7 +246,7 @@ def fix_interrupted_job(job: Job) -> None:
             )
         else:
             rollback_worker_phase_import_job(
-                job.job_id, job_operation, job.parameters.target
+                job, job_operation, job.parameters.target
             )
             logger.info(
                 f'{job.job_id}: Setting status to "failed" for interrupted job'
@@ -262,7 +276,7 @@ def fix_interrupted_job(job: Job) -> None:
             bump_manifesto = job.parameters.bump_manifesto
             if bump_manifesto is None:
                 raise RollbackException("No bump manifesto available")
-            rollback_bump(job.job_id, bump_manifesto.model_dump())
+            rollback_bump(job, bump_manifesto)
         except Exception as exc:
             error_message = f"Failed rollback for {job.job_id}"
             logger.exception(error_message, exc_info=exc)