Skip to content

Commit b959253

Browse files
author
Scott Collins
committed
wip failure injection
1 parent 4ca6dff commit b959253

4 files changed

Lines changed: 179 additions & 15 deletions

File tree

setup.cfg

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ install_requires =
3434
more-itertools>=9.0,<10.8
3535
joblib>=1.3.1,<1.6.0
3636
requests~=2.23
37+
requests-mock~=1.12.1
3738
types-requests~=2.23
3839
PyYAML~=6.0
3940
setuptools~=75.8.1

src/pds/ingress/client/pds_ingress_client.py

Lines changed: 24 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
from more_itertools import chunked as batched
2727
from pds.ingress import __version__
2828
from pds.ingress.util.auth_util import AuthUtil
29+
from pds.ingress.util.backoff_util import simulate_batch_request_failure
30+
from pds.ingress.util.backoff_util import simulate_ingress_failure
2931
from pds.ingress.util.config_util import ConfigUtil
3032
from pds.ingress.util.hash_util import md5_for_path
3133
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
353355
@backoff.on_exception(
354356
backoff.expo,
355357
Exception,
356-
max_time=120,
358+
max_time=10, # TODO: for testing only, revert to 120
357359
logger="request_batch_for_ingress",
358360
)
359361
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
410412
"x-amz-docs-region": api_gateway_region,
411413
}
412414

413-
response = requests.post(
414-
api_gateway_url, params=params, data=json.dumps(request_batch), headers=headers, timeout=600
415-
)
415+
# Simulate a random failure for the batch request if configured to do so
416+
with simulate_batch_request_failure(api_gateway_url.split("?")[0]):
417+
response = requests.post(
418+
api_gateway_url, params=params, data=json.dumps(request_batch), headers=headers, timeout=600
419+
)
420+
416421
elapsed_time = time.time() - start_time
417422

418423
# Ingress request successful
@@ -429,7 +434,7 @@ def request_batch_for_ingress(request_batch, batch_index, node_id, force_overwri
429434
@backoff.on_exception(
430435
backoff.expo,
431436
Exception,
432-
max_time=120,
437+
max_time=10, # TODO: for testing only, revert to 120
433438
logger="ingress_file_to_s3",
434439
)
435440
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):
488493
batch_pbar, total=os.stat(ingress_path).st_size, filename=os.path.basename(ingress_path)
489494
)
490495

491-
with open(ingress_path, "rb") as infile:
492-
# Wrap file I/O with our upload bar to automatically track file upload progress
493-
wrapped_file = CallbackIOWrapper(upload_pbar.update, infile, "read")
496+
# Simulate a random failure for the S3 ingress request if configured to do so
497+
with simulate_ingress_failure(s3_ingress_url.split("?")[0]):
498+
with open(ingress_path, "rb") as infile:
499+
# Wrap file I/O with our upload bar to automatically track file upload progress
500+
wrapped_file = CallbackIOWrapper(upload_pbar.update, infile, "read")
494501

495-
# Only send the file data if the file is non-empty
496-
response = requests.put(s3_ingress_url, data=wrapped_file if file_length > 0 else b"", headers=headers)
497-
response.raise_for_status()
502+
# Only send the file data if the file is non-empty
503+
response = requests.put(s3_ingress_url, data=wrapped_file if file_length > 0 else b"", headers=headers)
504+
response.raise_for_status()
498505

499506
logger.info("Batch %d : %s Ingest complete", batch_index, trimmed_path)
500507
update_summary_table(SUMMARY_TABLE, "uploaded", ingress_path)
@@ -518,7 +525,7 @@ def ingress_file_to_s3(ingress_response, batch_index, batch_pbar):
518525
@backoff.on_exception(
519526
backoff.expo,
520527
Exception,
521-
max_time=120,
528+
max_time=10, # TODO: for testing only, revert to 120
522529
logger="ingress_multipart_file_to_s3",
523530
)
524531
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):
583590
upload_pbar, f"{os.path.basename(ingress_path)} (Part {part_number}/{len(s3_ingress_urls)})"
584591
)
585592

586-
# Submit a single chunk to AWS
587-
response = requests.put(s3_ingress_url, data=next(chunk_iterator))
588-
response.raise_for_status()
593+
# Simulate a random failure for the S3 ingress request if configured to do so
594+
with simulate_ingress_failure(s3_ingress_url.split("?")[0]):
595+
# Submit a single chunk to AWS
596+
response = requests.put(s3_ingress_url, data=next(chunk_iterator))
597+
response.raise_for_status()
589598

590599
completed_parts.append({"ETag": response.headers["ETag"], "PartNumber": part_number})
591600
except Exception as err:

src/pds/ingress/conf.default.ini

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,3 +19,9 @@ console_format = "%(message)s"
1919
log_group_name = "/pds/nucleus/dum/client-log-group"
2020
log_file_path =
2121
batch_size = 250
22+
23+
[DEBUG]
24+
simulate_batch_request_failures = false
25+
batch_request_failure_rate = 0
26+
simulate_ingress_failures = false
27+
ingress_failure_rate = 0

src/pds/ingress/util/backoff_util.py

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,16 @@
77
automatic backoff/retry of HTTP requests.
88
99
"""
10+
import multiprocessing
11+
import random
12+
import requests_mock
13+
14+
from contextlib import contextmanager
15+
from distutils.util import strtobool
1016
from http import HTTPStatus
1117

18+
from pds.ingress.util.config_util import ConfigUtil
19+
1220
# When leveraging this module with the Lambda service functions, the requests
1321
# module will not be available within the Python runtime.
1422
# It should not be needed by those Lambda's, so use MagicMock to bypass any
@@ -24,6 +32,9 @@
2432
SSLError = MagicMock()
2533

2634

35+
MOCK_REQUESTS_SEMAPHORE = multiprocessing.Semaphore(1)
36+
37+
2738
def fatal_code(err: requests.exceptions.RequestException) -> bool:
2839
"""
2940
Determines if the HTTP return code associated with a requests exception
@@ -60,3 +71,140 @@ def fatal_code(err: requests.exceptions.RequestException) -> bool:
6071
else:
6172
# No response to interrogate, so default to no retry
6273
return True
74+
75+
76+
def check_failure_chance(percentage: int) -> bool:
77+
"""
78+
Checks if a simulated failure event should occur based on a given percentage
79+
chance.
80+
81+
Parameters
82+
----------
83+
percentage : int
84+
The desired percentage chance (e.g., 70 for 70%).
85+
86+
Returns
87+
-------
88+
bool: True if the failure should occur, False otherwise.
89+
90+
Raises
91+
------
92+
ValueError: If the percentage is not between 0 and 100.
93+
94+
"""
95+
if not (0 <= percentage <= 100):
96+
raise ValueError("Percentage must be between 0 and 100.")
97+
98+
# Generate a random float between 0.0 and 1.0
99+
random_number = random.random()
100+
101+
# Convert the percentage to a decimal for comparison
102+
chance_threshold = percentage / 100.0
103+
104+
return random_number < chance_threshold
105+
106+
107+
@contextmanager
108+
def _simulate_requests_failure(url, http_method, enable_key, failure_rate_key):
109+
"""
110+
Simulates a random failure for S3 ingress by registering the provided
111+
ingress URL with the requests mocker to raise an HTTPError exception.
112+
113+
Whether the failure is simulated is determined by the `simulate_ingress_failures`
114+
configuration option and the `ingress_failure_rate` percentage chance within
115+
the optional DEBUG section of the INI config. If this section is not present,
116+
this function should always default to not simulating a failure.
117+
118+
If the failure is not simulated, the mock_requests instance is reset to
119+
ensure no previous failures are registered.
120+
121+
Parameters
122+
----------
123+
url : str
124+
The URL to which the simulated failure will be applied.
125+
http_method : str
126+
HTTP method to register the failure for (e.g., 'POST', 'PUT').
127+
enable_key : str
128+
Name of the INI key to check if failure simulation is enabled.
129+
failure_rate_key : str
130+
Name of the INI key that specifies the percentage chance of failure.
131+
132+
"""
133+
config = ConfigUtil.get_config()
134+
135+
with requests_mock.Mocker(real_http=True) as mock_requests:
136+
try:
137+
# Check if simulated failures are enabled, and if so, if we should simulate
138+
# a failure via mock_requests based on the configured failure chance
139+
if bool(strtobool(config.get("DEBUG", enable_key, fallback="false"))):
140+
if check_failure_chance(int(config.get("DEBUG", failure_rate_key, fallback="0"))):
141+
mock_requests.register_uri(http_method, url, exc=requests.exceptions.HTTPError)
142+
143+
yield mock_requests
144+
finally:
145+
# Remove any previously registered URL(s)
146+
mock_requests.reset()
147+
148+
149+
@contextmanager
150+
def simulate_batch_request_failure(api_gateway_url):
151+
"""
152+
Simulates a random failure for an ingress batch request by registering the
153+
provided ingress URL with the requests mocker to raise an HTTPError exception.
154+
155+
Parameters
156+
----------
157+
api_gateway_url : str
158+
The API Gateway URL to which the simulated failure will be applied.
159+
160+
"""
161+
with MOCK_REQUESTS_SEMAPHORE:
162+
_simulate_requests_failure(
163+
api_gateway_url, "POST", "simulate_batch_request_failures", "batch_request_failure_rate"
164+
)
165+
# config = ConfigUtil.get_config()
166+
#
167+
# with requests_mock.Mocker(real_http=True) as mock_requests:
168+
# try:
169+
# # Check if simulated failures are enabled, and if so, if we should simulate
170+
# # a failure via mock_requests based on the configured failure chance
171+
# if bool(strtobool(config.get("DEBUG", "simulate_batch_request_failures", fallback="false"))):
172+
# if check_failure_chance(int(config.get("DEBUG", "batch_request_failure_rate", fallback="0"))):
173+
# mock_requests.register_uri("POST", api_gateway_url, exc=requests.exceptions.HTTPError)
174+
#
175+
# yield mock_requests
176+
# finally:
177+
# # Remove any previously registered URL(s)
178+
# mock_requests.reset()
179+
180+
@contextmanager
181+
def simulate_ingress_failure(s3_ingress_url):
182+
"""
183+
Simulates a random failure for S3 ingress by registering the provided
184+
ingress URL with the requests mocker to raise an HTTPError exception.
185+
186+
Parameters
187+
----------
188+
s3_ingress_url : str
189+
The S3 ingress URL to which the simulated failure will be applied.
190+
191+
"""
192+
#_simulate_requests_failure(mock_requests, s3_ingress_url, "PUT", "simulate_ingress_failures", "ingress_failure_rate")
193+
with MOCK_REQUESTS_SEMAPHORE:
194+
_simulate_requests_failure(
195+
s3_ingress_url, "PUT", "simulate_ingress_failures", "ingress_failure_rate"
196+
)
197+
# config = ConfigUtil.get_config()
198+
#
199+
# with requests_mock.Mocker(real_http=True) as mock_requests:
200+
# try:
201+
# # Check if simulated failures are enabled, and if so, if we should simulate
202+
# # a failure via mock_requests based on the configured failure chance
203+
# if bool(strtobool(config.get("DEBUG", "simulate_ingress_failures", fallback="false"))):
204+
# if check_failure_chance(int(config.get("DEBUG", "ingress_failure_rate", fallback="0"))):
205+
# mock_requests.register_uri("PUT", s3_ingress_url, exc=requests.exceptions.HTTPError)
206+
#
207+
# yield mock_requests
208+
# finally:
209+
# # Remove any previously registered URL(s)
210+
# mock_requests.reset()

0 commit comments

Comments
 (0)