wip failure simulation

Author: Scott Collins

setup.cfg

@@ -34,6 +34,7 @@ install_requires =
     more-itertools>=9.0,<10.8
     joblib>=1.3.1,<1.6.0
     requests~=2.23
+    requests-mock~=1.12.1
     types-requests~=2.23
     PyYAML~=6.0
     setuptools~=75.8.1

src/pds/ingress/client/pds_ingress_client.py

@@ -21,11 +21,14 @@
 import backoff
 import pds.ingress.util.log_util as log_util
 import requests
+import requests_mock
 from joblib import delayed
 from joblib import Parallel
 from more_itertools import chunked as batched
 from pds.ingress import __version__
 from pds.ingress.util.auth_util import AuthUtil
+from pds.ingress.util.backoff_util import simulate_batch_request_failure
+from pds.ingress.util.backoff_util import simulate_ingress_failure
 from pds.ingress.util.config_util import ConfigUtil
 from pds.ingress.util.hash_util import md5_for_path
 from pds.ingress.util.log_util import get_log_level
@@ -193,7 +196,8 @@ def _process_batch(batch_index, request_batch, node_id, force_overwrite, api_gat
             except Exception as err:
                 # If here, the HTTP request error was unrecoverable by a backoff/retry
                 trimmed_path = ingress_response.get("trimmed_path")
-                update_summary_table(SUMMARY_TABLE, "failed", 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, Reason:\n%s", batch_index, trimmed_path, str(err))
 
@@ -352,10 +356,11 @@ def _prepare_batch_for_ingress(ingress_path_batch, prefix, batch_index, batch_pb
 @backoff.on_exception(
     backoff.expo,
     Exception,
-    max_time=120,
+    max_time=10,  # TODO: for testing only, revert to 120
     logger="request_batch_for_ingress",
 )
-def request_batch_for_ingress(request_batch, batch_index, node_id, force_overwrite, api_gateway_config):
+#@requests_mock.Mocker(kw="mock_requests", real_http=True)
+def request_batch_for_ingress(request_batch, batch_index, node_id, force_overwrite, api_gateway_config, **kwargs):
     """
     Submits a batch of ingress requests to the PDS Ingress App API.
 
@@ -409,9 +414,12 @@ def request_batch_for_ingress(request_batch, batch_index, node_id, force_overwri
         "x-amz-docs-region": api_gateway_region,
     }
 
-    response = requests.post(
-        api_gateway_url, params=params, data=json.dumps(request_batch), headers=headers, timeout=600
-    )
+    # Simulate a random failure for the batch request if configured to do so
+    with simulate_batch_request_failure(api_gateway_url.split("?")[0]):
+        response = requests.post(
+            api_gateway_url, params=params, data=json.dumps(request_batch), headers=headers, timeout=600
+        )
+
     elapsed_time = time.time() - start_time
 
     # Ingress request successful
@@ -428,10 +436,11 @@ def request_batch_for_ingress(request_batch, batch_index, node_id, force_overwri
 @backoff.on_exception(
     backoff.expo,
     Exception,
-    max_time=120,
+    max_time=10,  # TODO: for testing only, revert to 120
     logger="ingress_file_to_s3",
 )
-def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
+#@requests_mock.Mocker(kw="mock_requests", real_http=True)
+def ingress_file_to_s3(ingress_response, batch_index, batch_pbar, **kwargs):
     """
     Copies the local file path using the pre-signed S3 URL returned from the
     Ingress Lambda App.
@@ -487,13 +496,15 @@ def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
             batch_pbar, total=os.stat(ingress_path).st_size, filename=os.path.basename(ingress_path)
         )
 
-        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")
+        # Simulate a random failure for the S3 ingress request if configured to do so
+        with simulate_ingress_failure(s3_ingress_url.split("?")[0]):
+            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")
 
-            # Only send the file data if the file is non-empty
-            response = requests.put(s3_ingress_url, data=wrapped_file if file_length > 0 else b"", headers=headers)
-            response.raise_for_status()
+                # Only send the file data if the file is non-empty
+                response = requests.put(s3_ingress_url, data=wrapped_file if file_length > 0 else b"", 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)
@@ -517,10 +528,11 @@ def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
 @backoff.on_exception(
     backoff.expo,
     Exception,
-    max_time=120,
+    max_time=10,  # TODO: for testing only, revert to 120
     logger="ingress_multipart_file_to_s3",
 )
-def ingress_multipart_file_to_s3(ingress_response, batch_index, batch_pbar):
+#@requests_mock.Mocker(kw="mock_requests", real_http=True)
+def ingress_multipart_file_to_s3(ingress_response, batch_index, batch_pbar, **kwargs):
     """
     Performs an ingress request for a file that is too large to be uploaded
     in a single request. The file is instead uploaded in multiple parts using
@@ -582,9 +594,11 @@ def ingress_multipart_file_to_s3(ingress_response, batch_index, batch_pbar):
                     upload_pbar, f"{os.path.basename(ingress_path)} (Part {part_number}/{len(s3_ingress_urls)})"
                 )
 
-                # Submit a single chunk to AWS
-                response = requests.put(s3_ingress_url, data=next(chunk_iterator))
-                response.raise_for_status()
+                # Simulate a random failure for the S3 ingress request if configured to do so
+                with simulate_ingress_failure(s3_ingress_url.split("?")[0]):
+                    # Submit a single chunk to AWS
+                    response = requests.put(s3_ingress_url, data=next(chunk_iterator))
+                    response.raise_for_status()
 
                 completed_parts.append({"ETag": response.headers["ETag"], "PartNumber": part_number})
         except Exception as err:
@@ -848,16 +862,22 @@ def main(args):
         try:
             init_batch_progress_bars(args.num_threads)
             perform_ingress(request_batchs, node_id, args.force_overwrite, config["API_GATEWAY"])
+        finally:
+            close_batch_progress_bars()
 
-            logger.info("All batches processed")
+        logger.info("All batches processed")
 
+        try:
             if len(SUMMARY_TABLE["failed"]) > 0:
+                logger.info("----------------------------------------")
                 logger.info("Reattempting ingress for failed files...")
+
                 failed_ingresses = SUMMARY_TABLE["failed"]
                 batched_failed_ingresses = list(batched(failed_ingresses, batch_size))
                 failed_request_batchs = prepare_batches(batched_failed_ingresses, args.prefix)
+
+                init_batch_progress_bars(args.num_threads)
                 perform_ingress(failed_request_batchs, node_id, args.force_overwrite, config["API_GATEWAY"])
-                logger.info("Reattempted ingress complete")
         finally:
             close_batch_progress_bars()
 

src/pds/ingress/conf.default.ini

@@ -19,3 +19,9 @@ console_format = "%(message)s"
 log_group_name = "/pds/nucleus/dum/client-log-group"
 log_file_path =
 batch_size = 250
+
+[DEBUG]
+simulate_batch_request_failures = false
+batch_request_failure_rate = 0
+simulate_ingress_failures = false
+ingress_failure_rate = 0

src/pds/ingress/util/backoff_util.py

@@ -7,8 +7,15 @@
 automatic backoff/retry of HTTP requests.
 
 """
+import random
+import requests_mock
+
+from contextlib import contextmanager
+from distutils.util import strtobool
 from http import HTTPStatus
 
+from pds.ingress.util.config_util import ConfigUtil
+
 # When leveraging this module with the Lambda service functions, the requests
 # module will not be available within the Python runtime.
 # It should not be needed by those Lambda's, so use MagicMock to bypass any
@@ -60,3 +67,142 @@ def fatal_code(err: requests.exceptions.RequestException) -> bool:
     else:
         # No response to interrogate, so default to no retry
         return True
+
+
+def check_failure_chance(percentage: int) -> bool:
+    """
+    Checks if a simulated failure event should occur based on a given percentage
+    chance.
+
+    Parameters
+    ----------
+    percentage : int
+        The desired percentage chance (e.g., 70 for 70%).
+
+    Returns
+    -------
+        bool: True if the failure should occur, False otherwise.
+
+    Raises
+    ------
+    ValueError: If the percentage is not between 0 and 100.
+
+    """
+    if not (0 <= percentage <= 100):
+        raise ValueError("Percentage must be between 0 and 100.")
+
+    # Generate a random float between 0.0 and 1.0
+    random_number = random.random()
+
+    # Convert the percentage to a decimal for comparison
+    chance_threshold = percentage / 100.0
+
+    return random_number < chance_threshold
+
+
+@contextmanager
+def _simulate_requests_failure(mock_requests, url, http_method, enable_key, failure_rate_key):
+    """
+    Simulates a random failure for S3 ingress by registering the provided
+    ingress URL with the requests mocker to raise an HTTPError exception.
+
+    Whether the failure is simulated is determined by the `simulate_ingress_failures`
+    configuration option and the `ingress_failure_rate` percentage chance within
+    the optional DEBUG section of the INI config. If this section is not present,
+    this function should always default to not simulating a failure.
+
+    If the failure is not simulated, the mock_requests instance is reset to
+    ensure no previous failures are registered.
+
+    Parameters
+    ----------
+    mock_requests : requests_mock.Mocker
+        Mocked requests instance to register the simulated failure.
+    url : str
+        The URL to which the simulated failure will be applied.
+    http_method : str
+        HTTP method to register the failure for (e.g., 'POST', 'PUT').
+    enable_key : str
+        Name of the INI key to check if failure simulation is enabled.
+    failure_rate_key : str
+        Name of the INI key that specifies the percentage chance of failure.
+
+    """
+    config = ConfigUtil.get_config()
+
+    try:
+        # Check if simulated failures are enabled, and if so, if we should simulate
+        # a failure via mock_requests based on the configured failure chance
+        if bool(strtobool(config.get("DEBUG", enable_key, fallback="false"))):
+            if check_failure_chance(int(config.get("DEBUG", failure_rate_key, fallback="0"))):
+                mock_requests.register_uri(http_method, url, exc=requests.exceptions.HTTPError)
+
+        yield mock_requests
+    finally:
+        # Remove any previously registered URL(s)
+        mock_requests.reset()
+
+
+@contextmanager
+#def simulate_batch_request_failure(mock_requests, api_gateway_url):
+def simulate_batch_request_failure(api_gateway_url):
+    """
+    Simulates a random failure for an ingress batch request by registering the
+    provided ingress URL with the requests mocker to raise an HTTPError exception.
+
+    Parameters
+    ----------
+    mock_requests : requests_mock.Mocker
+        Mocked requests instance to register the simulated failure.
+    api_gateway_url : str
+        The API Gateway URL to which the simulated failure will be applied.
+
+    """
+    #_simulate_requests_failure(
+    #    mock_requests, api_gateway_url, "POST", "simulate_batch_request_failures", "batch_request_failure_rate"
+    #)
+    config = ConfigUtil.get_config()
+
+    with requests_mock.Mocker(real_http=True) as mock_requests:
+        try:
+            # Check if simulated failures are enabled, and if so, if we should simulate
+            # a failure via mock_requests based on the configured failure chance
+            if bool(strtobool(config.get("DEBUG", "simulate_batch_request_failures", fallback="false"))):
+                if check_failure_chance(int(config.get("DEBUG", "batch_request_failure_rate", fallback="0"))):
+                    mock_requests.register_uri("POST", api_gateway_url, exc=requests.exceptions.HTTPError)
+
+            yield mock_requests
+        finally:
+            # Remove any previously registered URL(s)
+            mock_requests.reset()
+
+@contextmanager
+#def simulate_ingress_failure(mock_requests, s3_ingress_url):
+def simulate_ingress_failure(s3_ingress_url):
+    """
+    Simulates a random failure for S3 ingress by registering the provided
+    ingress URL with the requests mocker to raise an HTTPError exception.
+
+    Parameters
+    ----------
+    mock_requests : requests_mock.Mocker
+        Mocked requests instance to register the simulated failure.
+    s3_ingress_url : str
+        The S3 ingress URL to which the simulated failure will be applied.
+
+    """
+    #_simulate_requests_failure(mock_requests, s3_ingress_url, "PUT", "simulate_ingress_failures", "ingress_failure_rate")
+    config = ConfigUtil.get_config()
+
+    with requests_mock.Mocker(real_http=True) as mock_requests:
+        try:
+            # Check if simulated failures are enabled, and if so, if we should simulate
+            # a failure via mock_requests based on the configured failure chance
+            if bool(strtobool(config.get("DEBUG", "simulate_ingress_failures", fallback="false"))):
+                if check_failure_chance(int(config.get("DEBUG", "ingress_failure_rate", fallback="0"))):
+                    mock_requests.register_uri("PUT", s3_ingress_url, exc=requests.exceptions.HTTPError)
+
+            yield mock_requests
+        finally:
+            # Remove any previously registered URL(s)
+            mock_requests.reset()

src/pds/ingress/util/progress_util.py

@@ -279,7 +279,12 @@ def update_upload_pbar_filename(upload_pbar, filename):
 
 def close_batch_progress_bars():
     """Closes all Batch progress bars and associated Upload sub-bars."""
+    global BATCH_BARS, TOTAL_INGRESS_BAR  # noqa F824
+
     with BATCH_SEMAPHORE:
         for batch_pbar in BATCH_BARS:
             batch_pbar.upload_pbar.close()
             batch_pbar.close()
+
+        BATCH_BARS.clear()
+        TOTAL_INGRESS_BAR = None

src/pds/ingress/util/report_util.py

@@ -50,6 +50,8 @@ def update_summary_table(summary_table, key, paths):
         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"):