wip report table updates

Author: Scott Collins

src/pds/ingress/client/pds_ingress_client.py

@@ -164,6 +164,8 @@ def _process_batch(batch_index, request_batch, node_id, force_overwrite, api_gat
         fully processed.
 
     """
+    global SUMMARY_TABLE
+
     logger = get_logger("_process_batch", console=False)
 
     # Get an avaialble Batch progress bar to update while iterating through this
@@ -451,6 +453,8 @@ 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
+
     logger = get_logger("ingress_file_to_s3", console=False)
 
     response_result = int(ingress_response.get("result", -1))
@@ -477,12 +481,14 @@ def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
         with open(ingress_path, "rb") as infile:
             # Wrap file I/O with our upload bar to automatically track file upload progress
             wrapped_file = CallbackIOWrapper(upload_pbar.update, infile, "read")
-            files = {"file": (os.path.basename(ingress_path), wrapped_file, "application/octet-stream")}
-            response = requests.put(s3_ingress_url, files=files, headers=headers)
+            #files = {"file": (os.path.basename(ingress_path), wrapped_file, "application/octet-stream")}
+            #response = requests.put(s3_ingress_url, files=files, headers=headers)
+            response = requests.put(s3_ingress_url, data=wrapped_file, headers=headers)
             response.raise_for_status()
 
         logger.info("Batch %d : %s Ingest complete", batch_index, trimmed_path)
         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")
@@ -529,6 +535,8 @@ 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
+
     logger = get_logger("ingress_multipart_file_to_s3", console=False)
 
     response_result = int(ingress_response.get("result", -1))

src/pds/ingress/util/report_util.py

@@ -8,13 +8,17 @@
 
 """
 import json
+import multiprocessing
 import os
 import time
 from datetime import datetime
 from datetime import timezone
 
 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"""
 
@@ -57,18 +61,19 @@ def update_summary_table(summary_table, key, paths):
     if not isinstance(paths, list):
         paths = [paths]
 
-    summary_table[key].update(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 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)
+            # 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)
+        # Prune any now-visted paths from the unprocessed set
+        if key != "unprocessed":
+            summary_table["unprocessed"] -= set(paths)
 
 
 def print_ingress_summary(summary_table):