Merge branch 'main' into development

nlds/routers/files.py

@@ -412,7 +412,7 @@ async def put(
             status_code=status.HTTP_400_BAD_REQUEST, detail=response_error.json()
         )
 
-    if user not in ("jleland", "nrmassey", "nlds"):
+    if user not in ("nlds"):
         response_error = ResponseError(
             loc=["files", "archive"],
             msg=("archive action is admin-only"),

nlds_processors/archive/archive_base.py

@@ -103,8 +103,17 @@ def callback(self, ch, method, properties, body, connection):
         scraping, then runs the appropriate transfer function.
         """
         if not self._callback_common(ch, method, properties, body, connection):
+            # fail all files if callback common fails
+            rk_transfer_failed = ".".join(
+                [self.rk_parts[0], self.rk_parts[1], RK.FAILED]
+            )
+            for file in self.filelist:
+                file.failure_reason = 'Failed in archive transfer init'
+
+            self.send_pathlist(
+                self.filelist, rk_transfer_failed, self.body_json, state=State.FAILED
+            )
             return
-
         # get tape_url for those routes that need it
         if self.rk_parts[2] in [RK.START, RK.PREPARE, RK.PREPARE_CHECK]:
             try:

nlds_processors/archive/s3_to_tarfile_tape.py

@@ -377,26 +377,26 @@ def _verify_tape_server(self):
         status, _ = self.tape_client.ping()
         if status.status != 0:
             msg = (
-                f"Failed status message: {status.message}. ",
-                f"Could not ping cta server at {self.tape_server_url}.",
+                f"Failed status message: {status.message}. "
+                f"Could not ping cta server at {self.tape_server_url}."
             )
             raise S3StreamError(msg)
 
         # Stat the base directory and check it's a directory.
         status, resp = self.tape_client.stat(self.tape_base_dir)
         if status.status != 0:
             msg = (
-                f"Failed status message: {status.message}. ",
-                f"Base dir {self.tape_base_dir} could not be statted",
+                f"Failed status message: {status.message}. "
+                f"Base dir {self.tape_base_dir} could not be statted"
             )
             raise S3StreamError(msg)
         # Check whether the flag indicates it's a directory
         elif not bool(resp.flags & StatInfoFlags.IS_DIR):
             msg = (
-                f"Failed status message: {status.message}. ",
-                f"Full status object: {status}. ",
-                f"Stat result for base dir {self.tape_base_dir} ",
-                f"indicates it is not a directory.",
+                f"Failed status message: {status.message}. "
+                f"Full status object: {status}. "
+                f"Stat result for base dir {self.tape_base_dir} "
+                f"indicates it is not a directory."
             )
             raise S3StreamError(msg)
 

nlds_processors/archive/send_archive_next.py

@@ -30,6 +30,8 @@ def send_archive_next():
         MSG.SECRET_KEY: None,
         MSG.TAPE_URL: None,
         MSG.TENANCY: None,
+        MSG.USER: "admin-placeholder",
+        MSG.GROUP: "admin-placeholder"
     }
     # Load any cronjob config, if present
     cronjob_config = DEFAULT_CONFIG
@@ -40,8 +42,6 @@ def send_archive_next():
         MSG.DETAILS: {
             MSG.TRANSACT_ID: str(uuid4()),
             MSG.SUB_ID: str(uuid4()),
-            MSG.USER: "admin-placeholder",
-            MSG.GROUP: "admin-placeholder",
             MSG.TARGET: None,
             MSG.API_ACTION: "archive-put",
             MSG.JOB_LABEL: "archive-next",

nlds_processors/catalog/catalog_worker.py

@@ -901,6 +901,7 @@ def _catalog_archive_put(self, body: Dict, rk_origin: str) -> None:
         # Forward successful file details to archive for tape write
         rk_complete = ".".join([rk_origin, RK.CATALOG_ARCHIVE_NEXT, RK.COMPLETE])
 
+        # the user and group need to be set here
         body[MSG.DETAILS][MSG.USER] = next_holding.user
         body[MSG.DETAILS][MSG.GROUP] = next_holding.group
         body[MSG.META][MSG.HOLDING_ID] = next_holding.id

nlds_processors/index.py

@@ -95,7 +95,11 @@ def _scan(
         body_json: Dict[str, Any],
     ) -> None:
         # First change user and group so file permissions can be checked
-        self.set_ids(body_json)
+        try:
+            self.set_ids(body_json)
+        except KeyError:
+            self.log("Problem running set_ids, exiting _scan.", RK.LOG_ERROR)
+            return
 
         # Append routing info and then run the index
         body_json = self.append_route_info(body_json)

nlds_processors/transfer/base_transfer.py

@@ -83,25 +83,29 @@ def _callback_common(self, cm, method, properties, body, connection):
             self.transaction_id = self.body_json[MSG.DETAILS][MSG.TRANSACT_ID]
         except KeyError:
             self.log("Transaction id unobtainable, exiting callback.", RK.LOG_ERROR)
-            return
+            return False
 
         try:
             self.filelist = self.parse_filelist(self.body_json)
         except TypeError as e:
             self.log("Filelist not parseable, exiting callback", RK.LOG_ERROR)
-            return
+            return False
 
         try:
             (self.access_key, self.secret_key, self.tenancy) = (
                 self.get_objectstore_config(self.body_json)
             )
         except TransferError:
             self.log("Objectstore config unobtainable, exiting callback.", RK.LOG_ERROR)
-            return
+            return False
 
         # Set uid and gid from message contents
         self.log("Setting uid and gids now.", RK.LOG_INFO)
-        self.set_ids(self.body_json)
+        try:
+            self.set_ids(self.body_json)
+        except KeyError as e:
+            self.log("Problem running set_ids, exiting callback", RK.LOG_ERROR)
+            return False
 
         # Append route info to message to track the route of the message
         self.body_json = self.append_route_info(self.body_json)
@@ -110,6 +114,16 @@ def _callback_common(self, cm, method, properties, body, connection):
     def callback(self, ch, method, properties, body, connection):
 
         if not self._callback_common(ch, method, properties, body, connection):
+            # fail all files if callback common fails
+            rk_transfer_failed = ".".join(
+                [self.rk_parts[0], self.rk_parts[1], RK.FAILED]
+            )
+            for file in self.filelist:
+                file.failure_reason = 'Failed in transfer init'
+
+            self.send_pathlist(
+                self.filelist, rk_transfer_failed, self.body_json, state=State.FAILED
+            )
             return
 
         # API-methods that have an INITIATE phase will split the files across

nlds_processors/transfer/get_transfer.py

@@ -16,6 +16,7 @@
 import minio
 from minio.error import S3Error
 from retry import retry
+from urllib3.exceptions import HTTPError
 
 from nlds_processors.transfer.base_transfer import BaseTransferConsumer
 from nlds.rabbit.consumer import State
@@ -203,10 +204,12 @@ def _transfer_files(
         # build the routing keys
         rk_complete = ".".join([rk_origin, RK.TRANSFER_GET, RK.COMPLETE])
         rk_failed = ".".join([rk_origin, RK.TRANSFER_GET, RK.FAILED])
-        # set the ids for the
+        # set the ids for the files
         if self.chown_fl:
-            self.set_ids(body_json)
-
+            try:
+                self.set_ids(body_json)
+            except KeyError as e:
+                self.log("Problem running set_ids in _transfer_files", RK.LOG_ERROR)
         # Create client
         self.client = minio.Minio(
             tenancy,