Added unprocessed category to summary table, and flattened table to remove grouping by batch index

Author: Scott Collins

src/pds/ingress/client/pds_ingress_client.py

@@ -46,6 +46,7 @@
 from pds.ingress.util.report_util import parts_to_xml
 from pds.ingress.util.report_util import print_ingress_summary
 from pds.ingress.util.report_util import read_manifest_file
+from pds.ingress.util.report_util import update_summary_table
 from pds.ingress.util.report_util import write_manifest_file
 from requests.exceptions import RequestException
 from tqdm.utils import CallbackIOWrapper
@@ -165,6 +166,8 @@ def _process_batch(batch_index, request_batch, node_id, force_overwrite, api_gat
         fully processed.
 
     """
+    global SUMMARY_TABLE  # noqa: F824
+
     logger = get_logger("_process_batch", console=False)
 
     # Get an avaialble Batch progress bar to update while iterating through this
@@ -191,7 +194,8 @@ def _process_batch(batch_index, request_batch, node_id, force_overwrite, api_gat
             except RequestException as err:
                 # If here, the HTTP request error was unrecoverable by a backoff/retry
                 trimmed_path = ingress_response.get("trimmed_path")
-                SUMMARY_TABLE["failed"][batch_index].add(trimmed_path)
+                ingress_path = ingress_response.get("ingress_path")
+                update_summary_table(SUMMARY_TABLE, "failed", ingress_path)
 
                 logger.error(
                     "Batch %d : Ingress failed for %s, HTTP code: %s\n HTTP response text:\n%s",
@@ -455,18 +459,19 @@ def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
         If an unexpected response is received from the Ingress Lambda app.
 
     """
+    global SUMMARY_TABLE  # noqa: F824
+
     logger = get_logger("ingress_file_to_s3", console=False)
 
     response_result = int(ingress_response.get("result", -1))
     trimmed_path = ingress_response.get("trimmed_path")
+    ingress_path = ingress_response.get("ingress_path")
 
     if response_result == HTTPStatus.OK:
         s3_ingress_url = ingress_response.get("s3_url")
 
         logger.info("Batch %d : Ingesting %s to %s", batch_index, trimmed_path, s3_ingress_url.split("?")[0])
 
-        ingress_path = ingress_response.get("ingress_path")
-
         if not ingress_path:
             raise ValueError("No ingress path provided with response for %s", trimmed_path)
 
@@ -496,25 +501,24 @@ def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
             response.raise_for_status()
 
         logger.info("Batch %d : %s Ingest complete", batch_index, trimmed_path)
-        SUMMARY_TABLE["uploaded"][batch_index].add(trimmed_path)
-
-        # Update total number of bytes transferrred
-        SUMMARY_TABLE["transferred"] += os.stat(ingress_path).st_size
+        update_summary_table(SUMMARY_TABLE, "uploaded", ingress_path)
+        upload_pbar.reset()
     elif response_result == HTTPStatus.NO_CONTENT:
         logger.info(
             "Batch %d : Skipping ingress for %s, reason %s", batch_index, trimmed_path, ingress_response.get("message")
         )
-        SUMMARY_TABLE["skipped"][batch_index].add(trimmed_path)
+        update_summary_table(SUMMARY_TABLE, "skipped", ingress_path)
     elif response_result == HTTPStatus.NOT_FOUND:
         logger.warning(
             "Batch %d : Ingress failed for %s, reason: %s", batch_index, trimmed_path, ingress_response.get("message")
         )
-        SUMMARY_TABLE["failed"][batch_index].add(trimmed_path)
+        update_summary_table(SUMMARY_TABLE, "failed", ingress_path)
     else:
         logger.error("Batch %d : Unexepected response code (%d) from Ingress service", batch_index, response_result)
         raise RuntimeError
 
 
+# noinspection PyUnreachableCode
 @backoff.on_exception(
     backoff.expo,
     Exception,
@@ -545,15 +549,17 @@ def ingress_multipart_file_to_s3(ingress_response, batch_index, batch_pbar):
         If an unexpected response is received from the Ingress Lambda app.
 
     """
+    global SUMMARY_TABLE  # noqa: F824
+
     logger = get_logger("ingress_multipart_file_to_s3", console=False)
 
     response_result = int(ingress_response.get("result", -1))
     trimmed_path = ingress_response.get("trimmed_path")
+    ingress_path = ingress_response.get("ingress_path")
 
     if response_result == HTTPStatus.OK:
         logger.info("Batch %d : Performing Multipart Upload for %s", batch_index, trimmed_path)
 
-        ingress_path = ingress_response.get("ingress_path")
         s3_ingress_urls = ingress_response.get("s3_urls", [])
         upload_complete_url = ingress_response.get("upload_complete_url")
         upload_abort_url = ingress_response.get("upload_abort_url")
@@ -601,20 +607,17 @@ def ingress_multipart_file_to_s3(ingress_response, batch_index, batch_pbar):
         response.raise_for_status()
 
         logger.info("Batch %d : %s Multipart Upload complete", batch_index, trimmed_path)
-        SUMMARY_TABLE["uploaded"][batch_index].add(trimmed_path)
-
-        # Update total number of bytes transferrred
-        SUMMARY_TABLE["transferred"] += os.stat(ingress_path).st_size
+        update_summary_table(SUMMARY_TABLE, "uploaded", ingress_path)
     elif response_result == HTTPStatus.NO_CONTENT:
         logger.info(
             "Batch %d : Skipping ingress for %s, reason %s", batch_index, trimmed_path, ingress_response.get("message")
         )
-        SUMMARY_TABLE["skipped"][batch_index].add(trimmed_path)
+        update_summary_table(SUMMARY_TABLE, "skipped", ingress_path)
     elif response_result == HTTPStatus.NOT_FOUND:
         logger.warning(
             "Batch %d : Ingress failed for %s, reason: %s", batch_index, trimmed_path, ingress_response.get("message")
         )
-        SUMMARY_TABLE["failed"][batch_index].add(trimmed_path)
+        update_summary_table(SUMMARY_TABLE, "failed", ingress_path)
     else:
         logger.error("Batch %d : Unexepected response code (%d) from Ingress service", batch_index, response_result)
         raise RuntimeError
@@ -789,17 +792,24 @@ def main(args):
     with get_path_progress_bar(args.ingress_paths) as pbar:
         resolved_ingress_paths = PathUtil.resolve_ingress_paths(args.ingress_paths, pbar)
 
+    # Initialize the summary table, and populate the "unprocessed" table the set
+    # of resolved ingress paths
+    SUMMARY_TABLE = initialize_summary_table()
+    update_summary_table(SUMMARY_TABLE, "unprocessed", resolved_ingress_paths)
+
     node_id = args.node
 
     # Set the joblib pool size based on the number of "threads" requested
     PARALLEL.n_jobs = args.num_threads
 
     # Break the set of ingress paths into batches based on configured size
     batch_size = int(config["OTHER"].get("batch_size", fallback=1))
+    SUMMARY_TABLE["batch_size"] = batch_size
 
     batched_ingress_paths = list(batched(resolved_ingress_paths, batch_size))
     logger.info("Using batch size of %d", batch_size)
     logger.info("Request (%d files) split into %d batches", len(resolved_ingress_paths), len(batched_ingress_paths))
+    SUMMARY_TABLE["num_batches"] = len(batched_ingress_paths)
 
     if args.manifest_path and os.path.exists(args.manifest_path):
         logger.info("Reading existing manifest file %s", args.manifest_path)
@@ -813,8 +823,6 @@ def main(args):
         write_manifest_file(MANIFEST, os.path.abspath(args.manifest_path))
 
     if not args.dry_run:
-        SUMMARY_TABLE = initialize_summary_table()
-
         cognito_config = config["COGNITO"]
 
         # TODO: add support for command-line username/password?
@@ -848,23 +856,21 @@ def main(args):
         finally:
             close_batch_progress_bars()
 
-            # Capture completion time of transfer and batch configuration
-            SUMMARY_TABLE["end_time"] = time.time()
-            SUMMARY_TABLE["batch_size"] = batch_size
-            SUMMARY_TABLE["num_batches"] = len(batched_ingress_paths)
-
-            # Create the JSON report file, if requested
-            if args.report_path:
-                create_report_file(args, SUMMARY_TABLE)
-
-            # Print the summary table
-            print_ingress_summary(SUMMARY_TABLE)
-
             # Flush all logged statements to CloudWatch Logs
             log_util.CLOUDWATCH_HANDLER.flush()
     else:
         logger.info("Dry run requested, skipping ingress request submission.")
 
+    # Capture completion time
+    SUMMARY_TABLE["end_time"] = time.time()
+
+    # Create the JSON report file, if requested
+    if args.report_path:
+        create_report_file(args, SUMMARY_TABLE)
+
+    # Print the summary table
+    print_ingress_summary(SUMMARY_TABLE)
+
 
 def console_main():
     """No argument entrypoint for use with setuptools"""

src/pds/ingress/util/report_util.py

@@ -8,24 +8,28 @@
 
 """
 import json
+import multiprocessing
+import os
 import time
-from collections import defaultdict
 from datetime import datetime
 from datetime import timezone
-from itertools import chain
 
 from pds.ingress.util.log_util import get_logger
 
+REPORT_SEMAPHORE = multiprocessing.Semaphore(1)
+"""Semaphore used to control write access to Batch progress bars"""
+
 EXPECTED_MANIFEST_KEYS = ("ingress_path", "md5", "size", "last_modified")
 """The keys we expect to find assigned to each mapping within a read manifest"""
 
 
 def initialize_summary_table():
     """Returns a summary table initialized to its default state."""
     return {
-        "uploaded": defaultdict(set),
-        "skipped": defaultdict(set),
-        "failed": defaultdict(set),
+        "uploaded": set(),
+        "skipped": set(),
+        "failed": set(),
+        "unprocessed": set(),
         "transferred": 0,
         "start_time": time.time(),
         "end_time": None,
@@ -34,6 +38,46 @@ def initialize_summary_table():
     }
 
 
+def update_summary_table(summary_table, key, paths):
+    """
+    Updates the summary table with the provided key, index, and value.
+
+    Parameters
+    ----------
+    summary_table : dict
+        The summary table to update.
+    key : str
+        The key in the summary table to update (e.g., "uploaded", "skipped", "failed").
+    paths : str or list of str
+        The path value (or values) to add to the summary table for the specified key.
+        Note, these paths should be the absolute paths to files that were processed,
+        not the "trimmed" relative paths.
+
+    """
+    if key not in ("uploaded", "skipped", "failed", "unprocessed"):
+        raise KeyError(f"Invalid key '{key}' provided for summary table update.")
+
+    if key not in summary_table:
+        raise KeyError(f"Key '{key}' not found in summary table.")
+
+    if not isinstance(paths, list):
+        paths = [paths]
+
+    with REPORT_SEMAPHORE:
+        summary_table[key].update(paths)
+
+        if key == "uploaded":
+            # Update total number of bytes transferrred for successful uploads
+            summary_table["transferred"] += sum(os.stat(path).st_size for path in paths)
+
+            # If this file or files previous failed, remove from the failed set
+            summary_table["failed"] -= set(paths)
+
+        # Prune any now-visted paths from the unprocessed set
+        if key != "unprocessed":
+            summary_table["unprocessed"] -= set(paths)
+
+
 def print_ingress_summary(summary_table):
     """
     Prints the summary report for last execution of the client script.
@@ -46,9 +90,10 @@ def print_ingress_summary(summary_table):
     """
     logger = get_logger("print_ingress_summary")
 
-    num_uploaded = sum(len(batch) for batch in summary_table["uploaded"].values())
-    num_skipped = sum(len(batch) for batch in summary_table["skipped"].values())
-    num_failed = sum(len(batch) for batch in summary_table["failed"].values())
+    num_uploaded = len(summary_table["uploaded"])
+    num_skipped = len(summary_table["skipped"])
+    num_failed = len(summary_table["failed"])
+    num_unprocessed = len(summary_table["unprocessed"])
     start_time = summary_table["start_time"]
     end_time = summary_table["end_time"]
     transferred = summary_table["transferred"]
@@ -61,7 +106,8 @@ def print_ingress_summary(summary_table):
     logger.info("Uploaded: %d file(s)", num_uploaded)
     logger.info("Skipped: %d file(s)", num_skipped)
     logger.info("Failed: %d file(s)", num_failed)
-    logger.info("Total: %d files(s)", num_uploaded + num_skipped + num_failed)
+    logger.info("Unprocessed: %d file(s)", num_unprocessed)
+    logger.info("Total: %d files(s)", num_uploaded + num_skipped + num_failed + num_unprocessed)
     logger.info("Time elapsed: %.2f seconds", end_time - start_time)
     logger.info("Bytes tranferred: %d", transferred)
 
@@ -147,9 +193,10 @@ def create_report_file(args, summary_table):
     """
     logger = get_logger("create_report_file")
 
-    uploaded = list(sorted(chain(*summary_table["uploaded"].values())))
-    skipped = list(sorted(chain(*summary_table["skipped"].values())))
-    failed = list(sorted(chain(*summary_table["failed"].values())))
+    uploaded = list(sorted(summary_table["uploaded"]))
+    skipped = list(sorted(summary_table["skipped"]))
+    failed = list(sorted(summary_table["failed"]))
+    unprocessed = list(sorted(summary_table["unprocessed"]))
 
     report = {
         "Arguments": str(args),
@@ -163,10 +210,14 @@ def create_report_file(args, summary_table):
         "Total Skipped": len(skipped),
         "Failed": failed,
         "Total Failed": len(failed),
+        "Unprocessed": unprocessed,
+        "Total Unprocessed": len(unprocessed),
         "Bytes Transferred": summary_table["transferred"],
     }
 
-    report["Total Files"] = report["Total Uploaded"] + report["Total Skipped"] + report["Total Failed"]
+    report["Total Files"] = (
+        report["Total Uploaded"] + report["Total Skipped"] + report["Total Failed"] + report["Total Unprocessed"]
+    )
 
     try:
         logger.info("Writing JSON summary report to %s", args.report_path)