Tar files not yet deleted

Author: forsyth2

zstash/create.py

@@ -10,10 +10,11 @@
 
 from six.moves.urllib.parse import urlparse
 
-from .globus import GlobusTransferCollection, globus_activate, globus_finalize
+from .globus import globus_activate, globus_finalize
 from .hpss import hpss_put
 from .hpss_utils import add_files
 from .settings import DEFAULT_CACHE, config, get_db_filename, logger
+from .transfer_tracking import GlobusTransferCollection, HPSSTransferCollection
 from .utils import (
     create_tars_table,
     get_files_to_archive,
@@ -89,16 +90,23 @@ def create():
 
     # Create and set up the database
     logger.debug(f"{ts_utc()}: Calling create_database()")
-    failures: List[str] = create_database(cache, args, gtc=gtc)
+    htc: HPSSTransferCollection = HPSSTransferCollection()
+    failures: List[str] = create_database(cache, args, gtc=gtc, htc=htc)
 
     # Transfer to HPSS. Always keep a local copy.
     logger.debug(f"{ts_utc()}: calling hpss_put() for {get_db_filename(cache)}")
     hpss_put(
-        hpss, get_db_filename(cache), cache, keep=args.keep, is_index=True, gtc=gtc
+        hpss,
+        get_db_filename(cache),
+        cache,
+        keep=args.keep,
+        is_index=True,
+        gtc=gtc,
+        # htc=htc,
     )
 
     logger.debug(f"{ts_utc()}: calling globus_finalize()")
-    globus_finalize(gtc, non_blocking=args.non_blocking)
+    globus_finalize(gtc, htc, non_blocking=args.non_blocking)
 
     if len(failures) > 0:
         # List the failures
@@ -208,7 +216,10 @@ def setup_create() -> Tuple[str, argparse.Namespace]:
 
 
 def create_database(
-    cache: str, args: argparse.Namespace, gtc: Optional[GlobusTransferCollection]
+    cache: str,
+    args: argparse.Namespace,
+    gtc: Optional[GlobusTransferCollection],
+    htc: Optional[HPSSTransferCollection],
 ) -> List[str]:
     # Create new database
     logger.debug(f"{ts_utc()}:Creating index database")
@@ -284,6 +295,7 @@ def create_database(
             overwrite_duplicate_tars=args.overwrite_duplicate_tars,
             force_database_corruption=args.for_developers_force_database_corruption,
             gtc=gtc,
+            htc=htc,
         )
     except FileNotFoundError as e:
         if args.follow_symlinks:

zstash/globus.py

@@ -3,49 +3,31 @@
 import sys
 from typing import List, Optional
 
-from globus_sdk import TransferAPIError, TransferClient, TransferData
+from globus_sdk import TransferAPIError, TransferData
 from globus_sdk.response import GlobusHTTPResponse
-from globus_sdk.services.transfer.response.iterable import IterableTransferResponse
 from six.moves.urllib.parse import urlparse
 
 from .globus_utils import (
     HPSS_ENDPOINT_MAP,
     check_state_files,
+    file_exists,
     get_local_endpoint_id,
     get_transfer_client_with_auth,
+    globus_block_wait,
+    globus_wait,
     set_up_TransferData,
     submit_transfer_with_checks,
 )
 from .settings import logger
+from .transfer_tracking import (
+    GlobusTransfer,
+    GlobusTransferCollection,
+    HPSSTransferCollection,
+    delete_transferred_files,
+)
 from .utils import ts_utc
 
 
-class GlobusTransfer(object):
-    def __init__(self):
-        self.transfer_data: Optional[TransferData] = None
-        self.task_id: Optional[str] = None
-        # https://docs.globus.org/api/transfer/task/#task_fields
-        # ACTIVE, SUCCEEDED, FAILED, INACTIVE
-        self.task_status: Optional[str] = None
-
-
-class GlobusTransferCollection(object):
-    def __init__(self):
-        # Attributes common to all the transfers
-        self.remote_endpoint: Optional[str] = None
-        self.local_endpoint: Optional[str] = None
-        self.transfer_client: Optional[TransferClient] = None
-        self.archive_directory_listing: Optional[IterableTransferResponse] = None
-
-        self.transfers: List[GlobusTransfer] = (
-            []
-        )  # TODO: Replace with collections.deque?
-        self.cumulative_tarfiles_pushed: int = 0
-
-    def get_most_recent_transfer(self) -> Optional[GlobusTransfer]:
-        return self.transfers[-1] if self.transfers else None
-
-
 def globus_activate(
     hpss: str, gtc: Optional[GlobusTransferCollection] = None
 ) -> Optional[GlobusTransferCollection]:
@@ -72,13 +54,6 @@ def globus_activate(
     return gtc
 
 
-def file_exists(archive_directory_listing, name: str) -> bool:
-    for entry in archive_directory_listing:
-        if entry.get("name") == name:
-            return True
-    return False
-
-
 # C901 'globus_transfer' is too complex (20)
 def globus_transfer(  # noqa: C901
     gtc: Optional[GlobusTransferCollection],
@@ -168,6 +143,9 @@ def globus_transfer(  # noqa: C901
         logger.info(
             f"{ts_utc()}: SURFACE Submit Transfer returned new task_id = {task_id} for label {transfer_data['label']}"
         )
+        # Nullify the submitted transfer data structure so that a new one will be created on next call.
+        transfer_data = None
+
         new_transfer = GlobusTransfer()
         new_transfer.transfer_data = transfer_data
         new_transfer.task_id = task_id
@@ -196,7 +174,6 @@ def globus_transfer(  # noqa: C901
                 transfer_client=gtc.transfer_client,
                 task_id=new_mrt.task_id,
                 wait_timeout=7200,
-                polling_interval=10,
                 max_retries=5,
             )
         else:
@@ -208,146 +185,67 @@ def globus_transfer(  # noqa: C901
             globus_wait(gtc.transfer_client, new_mrt.task_id)
 
 
-def globus_block_wait(
-    transfer_client: TransferClient,
-    task_id: str,
-    wait_timeout: int,
-    polling_interval: int,
-    max_retries: int,
-):
-
-    # poll every "polling_interval" seconds to speed up small transfers.  Report every 2 hours, stop waiting aftert 5*2 = 10 hours
-    logger.info(
-        f"{ts_utc()}: BLOCKING START: invoking task_wait for task_id = {task_id}"
-    )
-    task_status = "UNKNOWN"
-    retry_count = 0
-    while retry_count < max_retries:
-        try:
-            # Wait for the task to complete
-            logger.info(
-                f"{ts_utc()}: on task_wait try {retry_count+1} out of {max_retries}"
-            )
-            transfer_client.task_wait(
-                task_id, timeout=wait_timeout, polling_interval=10
-            )
-            logger.info(f"{ts_utc()}: done with wait")
-        except Exception as e:
-            logger.error(f"Unexpected Exception: {e}")
-        else:
-            curr_task = transfer_client.get_task(task_id)
-            task_status = curr_task["status"]
-            if task_status == "SUCCEEDED":
-                break
-        finally:
-            retry_count += 1
-            logger.info(
-                f"{ts_utc()}: BLOCKING retry_count = {retry_count} of {max_retries} of timeout {wait_timeout} seconds"
-            )
-
-    if retry_count == max_retries:
-        logger.info(
-            f"{ts_utc()}: BLOCKING EXHAUSTED {max_retries} of timeout {wait_timeout} seconds"
-        )
-        task_status = "EXHAUSTED_TIMEOUT_RETRIES"
-
-    logger.info(
-        f"{ts_utc()}: BLOCKING ENDS: task_id {task_id} returned from task_wait with status {task_status}"
-    )
-
-    return task_status
-
-
-def globus_wait(transfer_client: TransferClient, task_id: str):
-    try:
-        """
-        A Globus transfer job (task) can be in one of the three states:
-        ACTIVE, SUCCEEDED, FAILED. The script every 20 seconds polls a
-        status of the transfer job (task) from the Globus Transfer service,
-        with 20 second timeout limit. If the task is ACTIVE after time runs
-        out 'task_wait' returns False, and True otherwise.
-        """
-        while not transfer_client.task_wait(task_id, timeout=300, polling_interval=20):
-            pass
-        """
-        The Globus transfer job (task) has been finished (SUCCEEDED or FAILED).
-        Check if the transfer SUCCEEDED or FAILED.
-        """
-        task = transfer_client.get_task(task_id)
-        if task["status"] == "SUCCEEDED":
-            src_ep = task["source_endpoint_id"]
-            dst_ep = task["destination_endpoint_id"]
-            label = task["label"]
-            logger.info(
-                "Globus transfer {}, from {} to {}: {} succeeded".format(
-                    task_id, src_ep, dst_ep, label
-                )
-            )
-        else:
-            logger.error("Transfer FAILED")
-    except TransferAPIError as e:
-        if e.code == "NoCredException":
-            logger.error(
-                "{}. Please go to https://app.globus.org/endpoints and activate the endpoint.".format(
-                    e.message
-                )
-            )
-        else:
-            logger.error(e)
-        sys.exit(1)
-    except Exception as e:
-        logger.error("Exception: {}".format(e))
-        sys.exit(1)
-
-
 def globus_finalize(
-    gtc: Optional[GlobusTransferCollection], non_blocking: bool = False
+    gtc: Optional[GlobusTransferCollection],
+    htc: HPSSTransferCollection,
+    non_blocking: bool = False,
 ):
     last_task_id = None
 
     if gtc is None:
         logger.warning("No GlobusTransferCollection object provided for finalization")
         return
 
-    transfer = gtc.get_most_recent_transfer()
-
-    if transfer and transfer.transfer_data:
-        # DEBUG: review accumulated items in TransferData
-        logger.info(f"{ts_utc()}: FINAL TransferData: accumulated items:")
-        attribs = transfer.transfer_data.__dict__
-        for item in attribs["data"]["DATA"]:
-            if item["DATA_TYPE"] == "transfer_item":
-                gtc.cumulative_tarfiles_pushed += 1
-                print(
-                    f"    (finalize) PUSHING ({gtc.cumulative_tarfiles_pushed}) source item: {item['source_path']}",
-                    flush=True,
-                )
+    transfer: Optional[GlobusTransfer] = gtc.get_most_recent_transfer()
+    if transfer:
+        if transfer.transfer_data:
+            # DEBUG: review accumulated items in TransferData
+            logger.info(f"{ts_utc()}: FINAL TransferData: accumulated items:")
+            attribs = transfer.transfer_data.__dict__
+            for item in attribs["data"]["DATA"]:
+                if item["DATA_TYPE"] == "transfer_item":
+                    gtc.cumulative_tarfiles_pushed += 1
+                    print(
+                        f"    (finalize) PUSHING ({gtc.cumulative_tarfiles_pushed}) source item: {item['source_path']}",
+                        flush=True,
+                    )
 
-        # SUBMIT new transfer here
-        logger.info(
-            f"{ts_utc()}: DIVING: Submit Transfer for {transfer.transfer_data['label']}"
-        )
-        try:
-            last_task = submit_transfer_with_checks(
-                gtc.transfer_client, transfer.transfer_data
+            # SUBMIT new transfer here
+            logger.info(
+                f"{ts_utc()}: DIVING: Submit Transfer for {transfer.transfer_data['label']}"
             )
-            last_task_id = last_task.get("task_id")
-        except TransferAPIError as e:
-            if e.code == "NoCredException":
-                logger.error(
-                    "{}. Please go to https://app.globus.org/endpoints and activate the endpoint.".format(
-                        e.message
-                    )
+            try:
+                last_task = submit_transfer_with_checks(
+                    gtc.transfer_client, transfer.transfer_data
                 )
-            else:
-                logger.error(e)
-            sys.exit(1)
-        except Exception as e:
-            logger.error("Exception: {}".format(e))
-            sys.exit(1)
+                last_task_id = last_task.get("task_id")
+            except TransferAPIError as e:
+                if e.code == "NoCredException":
+                    logger.error(
+                        "{}. Please go to https://app.globus.org/endpoints and activate the endpoint.".format(
+                            e.message
+                        )
+                    )
+                else:
+                    logger.error(e)
+                sys.exit(1)
+            except Exception as e:
+                logger.error("Exception: {}".format(e))
+                sys.exit(1)
 
-    if not non_blocking:
-        if transfer and transfer.task_id:
-            globus_wait(gtc.transfer_client, transfer.task_id)
+        if not non_blocking:
+            if transfer and transfer.task_id:
+                globus_wait(gtc.transfer_client, transfer.task_id)
         if last_task_id:
             globus_wait(gtc.transfer_client, last_task_id)
+
+        # TODO: figure out how to end!
+        # new_mrt: Optional[GlobusTransfer] = gtc.get_most_recent_transfer()
+        # if new_mrt and new_mrt.task_id:
+        #     new_mrt.task_status = globus_block_wait(
+        #         transfer_client=gtc.transfer_client,
+        #         task_id=new_mrt.task_id,
+        #         wait_timeout=7200,
+        #         max_retries=5,
+        #     )
+        delete_transferred_files(htc)