Initial addition of a randomized failure injection capability for use with the requests package

This new feature adds the ability to simulate randomized failures when invoking the requests package to submit data from the DUM client to the service in AWS.
Configuration of the new the feature is controlled within the INI config.

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

@@ -26,6 +26,8 @@
 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
@@ -353,7 +355,7 @@ 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):
@@ -410,9 +412,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
@@ -429,7 +434,7 @@ 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):
@@ -488,13 +493,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)
@@ -518,7 +525,7 @@ 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):
@@ -583,9 +590,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:

src/pds/ingress/conf.default.ini

@@ -19,3 +19,11 @@ 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
+batch_request_failure_class = requests.exceptions.HTTPError
+simulate_ingress_failures = false
+ingress_failure_rate = 0
+ingress_failure_class = requests.exceptions.HTTPError

src/pds/ingress/util/backoff_util.py

@@ -7,8 +7,16 @@
 automatic backoff/retry of HTTP requests.
 
 """
+import importlib
+import multiprocessing
+import random
+from contextlib import contextmanager
 from http import HTTPStatus
 
+import requests_mock
+from distutils.util import strtobool
+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
@@ -24,6 +32,9 @@
     SSLError = MagicMock()
 
 
+MOCK_REQUESTS_SEMAPHORE = multiprocessing.Semaphore(1)
+
+
 def fatal_code(err: requests.exceptions.RequestException) -> bool:
     """
     Determines if the HTTP return code associated with a requests exception
@@ -60,3 +71,138 @@ 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
+
+
+def _simulate_requests_failure(mock_requests, url, http_method, enable_key, failure_rate_key, failure_class_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
+        The requests mocker instance to register the simulated failure with.
+    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.
+    failure_class_key : str
+        Name of the INI key that specifies the exception class to raise on failure.
+
+    """
+    config = ConfigUtil.get_config()
+
+    # 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"))):
+            # Dynamically import the exception class to raise
+            failure_class_str = config.get("DEBUG", failure_class_key, fallback="builtins.RuntimeError")
+            failure_class_module, failure_exception_name = failure_class_str.rsplit(".", 1)
+
+            module = importlib.import_module(failure_class_module)
+            exception_klass = getattr(module, failure_exception_name)
+
+            # Register the URL with the mock_requests to raise the specified exception
+            mock_requests.register_uri(http_method, url, exc=exception_klass)
+
+    return mock_requests
+
+
+@contextmanager
+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
+    ----------
+    api_gateway_url : str
+        The API Gateway URL to which the simulated failure will be applied.
+
+    """
+    with MOCK_REQUESTS_SEMAPHORE:
+        with requests_mock.Mocker(real_http=True) as mock_requests:
+            try:
+                yield _simulate_requests_failure(
+                    mock_requests,
+                    api_gateway_url,
+                    "POST",
+                    "simulate_batch_request_failures",
+                    "batch_request_failure_rate",
+                    "batch_request_failure_class",
+                )
+            finally:
+                # Remove any previously registered URL(s)
+                mock_requests.reset()
+
+
+@contextmanager
+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
+    ----------
+    s3_ingress_url : str
+        The S3 ingress URL to which the simulated failure will be applied.
+
+    """
+    with MOCK_REQUESTS_SEMAPHORE:
+        with requests_mock.Mocker(real_http=True) as mock_requests:
+            try:
+                yield _simulate_requests_failure(
+                    mock_requests,
+                    s3_ingress_url,
+                    "PUT",
+                    "simulate_ingress_failures",
+                    "ingress_failure_rate",
+                    "ingress_failure_class",
+                )
+            finally:
+                # Remove any previously registered URL(s)
+                mock_requests.reset()

tests/pds/ingress/util/data/mock.backoff.config.ini

@@ -0,0 +1,29 @@
+# Configuration file intended for testing the DEBUG section and associated failure injection feature
+[API_GATEWAY]
+url_template = https://{id}.execute-api.{region}.amazonaws.com/{stage}/{resource}
+id           = abcdefghi
+region       = us-west-2
+stage        = test
+
+[COGNITO]
+client_id    = 123456789
+username     = cognito_user
+password     = cognito_pass  #pragma: allowlist secret
+region       = us-west-2
+
+[OTHER]
+log_level = INFO
+file_format = "[%(asctime)s] %(levelname)s %(threadName)s %(funcName)s : %(message)s"
+cloudwatch_format = '%(levelname)s %(threadName)s %(funcName)s : %(message)s'
+console_format = "%(message)s"
+log_group_name = "/pds/nucleus/dum/client-log-group"
+log_file_path =
+batch_size = 250
+
+[DEBUG]
+simulate_batch_request_failures = true
+batch_request_failure_rate = 100
+batch_request_failure_class = builtins.TypeError
+simulate_ingress_failures = true
+ingress_failure_rate = 100
+ingress_failure_class = requests.exceptions.HTTPError

tests/pds/ingress/util/data/mock.nodebug.config.ini

@@ -0,0 +1,21 @@
+# Default configuration file with no DEBUG section included
+[API_GATEWAY]
+url_template = https://{id}.execute-api.{region}.amazonaws.com/{stage}/{resource}
+id           = abcdefghi
+region       = us-west-2
+stage        = test
+
+[COGNITO]
+client_id    = 123456789
+username     = cognito_user
+password     = cognito_pass  #pragma: allowlist secret
+region       = us-west-2
+
+[OTHER]
+log_level = INFO
+file_format = "[%(asctime)s] %(levelname)s %(threadName)s %(funcName)s : %(message)s"
+cloudwatch_format = '%(levelname)s %(threadName)s %(funcName)s : %(message)s'
+console_format = "%(message)s"
+log_group_name = "/pds/nucleus/dum/client-log-group"
+log_file_path =
+batch_size = 250