Retry calls to S3 (#105)

Author: paulineribeyre

gen3workflow/app.py

@@ -67,14 +67,12 @@ async def get_status():
         generate_unique_id_function=generate_unique_route_id,
     )
 
-    # `async_client` is used to hit the TES API, the Arborist service and AWS S3.
-    # Calls to S3 tend to timeout when uploading large files (and we might also be rate-limited).
-    # AsyncHTTPTransport supports retrying on httpx.ConnectError or httpx.ConnectTimeout.
+    # `async_client` is used to hit the TES API and AWS S3.
+    # Calls to AWS S3 tend to timeout when uploading large files, so we increase the timeout.
+    # We may also be rate-limited, so calls are retried (see routes/s3.py).
     # The `httpx_client` parameter is not meant to be used in production. It allows mocking
     # external calls when testing.
-    app.async_client = httpx_client or httpx.AsyncClient(
-        transport=httpx.AsyncHTTPTransport(retries=3), timeout=120
-    )
+    app.async_client = httpx_client or httpx.AsyncClient(timeout=120)
 
     app.include_router(ga4gh_tes_router, tags=["GA4GH TES"])
     app.include_router(s3_router, tags=["S3"])

gen3workflow/routes/ga4gh_tes.py

@@ -208,6 +208,8 @@ def apply_view_to_task(view: str, task: dict) -> dict:
         task["inputs"][i].pop("content", None)
     for i in range(len(task.get("logs", []))):
         task["logs"][i].pop("system_logs", None)
+        # TODO if the TES server returns a SYSTEM_ERROR, we may want to keep the system_logs, or at
+        # least log them before removing them
 
     return task
 

gen3workflow/routes/s3.py

@@ -1,6 +1,7 @@
+import asyncio
 from datetime import datetime, timezone
 import hashlib
-import traceback
+import random
 import urllib.parse
 
 import boto3
@@ -14,18 +15,23 @@
     HTTP_400_BAD_REQUEST,
     HTTP_401_UNAUTHORIZED,
     HTTP_403_FORBIDDEN,
+    HTTP_404_NOT_FOUND,
 )
 
 from gen3workflow import aws_utils, logger
 from gen3workflow.auth import Auth
 from gen3workflow.config import config
-from gen3workflow.routes.system import get_status
 
 
 s3_root_router = APIRouter(include_in_schema=False)
 s3_router = APIRouter(prefix="/s3")
 
 
+S3_MAX_RETRIES = 3
+S3_RETRY_BASE_DELAY = 0.5
+S3_RETRY_BACKOFF_FACTOR = 2
+
+
 async def set_access_token_and_get_user_id(auth: Auth, headers: Headers) -> str:
     """
     Extract the user's access token and (in some cases) the user's ID, which should have been
@@ -336,28 +342,65 @@ async def s3_endpoint(path: str, request: Request):
         signing_key, string_to_sign.encode("utf-8"), hashlib.sha256
     ).hexdigest()
 
-    # construct the Authorization header from the credentials and the signature, and forward the
-    # call to AWS S3 with the new Authorization header
+    # construct the Authorization header from the credentials and the signature
     headers["authorization"] = (
         f"AWS4-HMAC-SHA256 Credential={credentials.access_key}/{date}/{region}/{service}/aws4_request, SignedHeaders={signed_headers}, Signature={signature}"
     )
     s3_api_url = f"https://{user_bucket}.s3.{region}.amazonaws.com/{api_endpoint}"
     logger.debug(f"Outgoing S3 request: '{request.method} {s3_api_url}'")
-    response = await request.app.async_client.request(
-        method=request.method,
-        url=s3_api_url,
-        headers=headers,
-        params=query_params,
-        data=body,
-    )
 
-    if response.status_code >= 300:
-        logger.debug(f"Received a failure status code from AWS: {response.status_code}")
-        # no need to log 404 errors except in debug mode: they are are expected when running
-        # workflows (e.g. for Nextflow workflows, error output files may not be present when there
-        # were no errors)
-        if response.status_code != 404:
-            logger.error(f"Error from AWS: {response.status_code} {response.text}")
+    # forward the call to AWS S3 with the new Authorization header.
+    # this call is retried with exponential backoff in case of unexpected error from S3.
+    for attempt in range(1, S3_MAX_RETRIES + 1):
+        proceed = True
+        exception = None
+        try:
+            response = await request.app.async_client.request(
+                method=request.method,
+                url=s3_api_url,
+                headers=headers,
+                params=query_params,
+                data=body,
+            )
+
+            if response.status_code >= 300:
+                # no need to log details (unless in debug mode) or retry in the case of a 404
+                # error: 404s are are expected when running workflows (e.g. for Nextflow workflows,
+                # stderr output files may not be present when there were no errors)
+                if response.status_code != HTTP_404_NOT_FOUND:
+                    logger.error(
+                        f"Error from S3: {response.status_code} {response.text}"
+                    )
+                    # do not retry in the case of a 403 error: authentication is done internally by
+                    # this function, so 403 errors are internal service errors
+                    if response.status_code != HTTP_403_FORBIDDEN:
+                        proceed = False
+                else:
+                    logger.debug(f"Error from S3: {response.status_code}")
+        except Exception as e:
+            logger.error(f"Exception while attempting to make a call to S3: {e}")
+            proceed = False
+            exception = e
+
+        # exit if the call succeeded or should not be retried, or we reached the max number of
+        # retries
+        if proceed:
+            break
+        if attempt == S3_MAX_RETRIES:
+            logger.error(
+                f"Outgoing S3 request failed (attempt {attempt}/{S3_MAX_RETRIES}). Giving up"
+            )
+            if exception:
+                raise exception
+            break
+
+        # retry with exponential backoff
+        delay = S3_RETRY_BASE_DELAY * (S3_RETRY_BACKOFF_FACTOR**attempt)
+        delay += delay * 0.1 * random.uniform(-1, 1)  # add jitter
+        logger.warning(
+            f"Outgoing S3 request failed (attempt {attempt}/{S3_MAX_RETRIES}). Retrying in {delay:.2f} seconds"
+        )
+        await asyncio.sleep(delay)
 
     # return the response from AWS S3.
     # - mask the details of 403 errors from the end user: authentication is done internally by this