[S3] More comprehensive error handling

metaflow/plugins/datatools/s3/s3.py

@@ -853,6 +853,8 @@ def _head():
                                 raise MetaflowS3NotFound()
                         elif info["error"] == s3op.ERROR_URL_ACCESS_DENIED:
                             raise MetaflowS3AccessDenied()
+                        elif info["error"] == s3op.ERROR_INVALID_REQUEST:
+                            raise MetaflowS3Exception("Invalid request for %s" % s3url)
                         else:
                             raise MetaflowS3Exception("Got error: %d" % info["error"])
                     else:
@@ -1379,6 +1381,8 @@ def _one_boto_op(self, op, url, create_tmp_file=True):
                     raise MetaflowS3AccessDenied(url)
                 elif error_code == 416:
                     raise MetaflowS3InvalidRange(err)
+                elif error_code == 400:
+                    raise MetaflowS3Exception(f"Invalid request for {url}: {str(err)}")
                 elif error_code == "NoSuchBucket":
                     raise MetaflowS3URLException("Specified S3 bucket doesn't exist.")
                 error = str(err)

metaflow/plugins/datatools/s3/s3op.py

@@ -107,6 +107,7 @@ def __str__(self):
 ERROR_INVALID_RANGE = 11
 ERROR_TRANSIENT = 12
 ERROR_OUT_OF_DISK_SPACE = 13
+ERROR_INVALID_REQUEST = 14
 
 
 def format_result_line(idx, prefix, url="", local=""):
@@ -129,37 +130,95 @@ def normalize_client_error(err):
     try:
         return int(error_code)
     except ValueError:
-        if error_code in ("AccessDenied", "AllAccessDisabled", "InvalidAccessKeyId"):
-            return 403
-        if error_code in ("NoSuchKey", "NoSuchBucket"):
-            return 404
-        if error_code == "InvalidRange":
-            return 416
-        # We "normalize" retriable server errors to 503. These are also considered
-        # transient by boto3 (see:
-        # https://boto3.amazonaws.com/v1/documentation/api/latest/guide/retries.html)
-        if error_code in (
-            "SlowDown",
-            "RequestTimeout",
-            "RequestTimeoutException",
-            "PriorRequestNotComplete",
-            "ConnectionError",
-            "HTTPClientError",
-            "Throttling",
-            "ThrottlingException",
-            "ThrottledException",
-            "RequestThrottledException",
-            "TooManyRequestsException",
-            "ProvisionedThroughputExceededException",
-            "TransactionInProgressException",
-            "RequestLimitExceeded",
-            "BandwidthLimitExceeded",
-            "LimitExceededException",
-            "RequestThrottled",
-            "EC2ThrottledException",
-            "InternalError",
-        ):
-            return 503
+        pass
+
+    # Permission or access-related errors → 403 Forbidden
+    permission_errors = {
+        "AccessDenied",
+        "AccessDeniedException",
+        "AccountProblem",
+        "AllAccessDisabled",
+        "AuthFailure",
+        "ExpiredToken",
+        "InvalidAccessKeyId",
+        "InvalidSecurity",
+        "SignatureDoesNotMatch",
+        "UnauthorizedOperation",
+        "UnrecognizedClientException",
+    }
+
+    # Not found errors → 404 Not Found
+    not_found_errors = {
+        "NoSuchBucket",
+        "NoSuchKey",
+        "NotFound",
+    }
+
+    # Range/invalid byte-range errors → 416
+    range_errors = {
+        "InvalidRange",
+    }
+
+    # Server-side throttling, timeout, or transient errors → 503
+    # https://boto3.amazonaws.com/v1/documentation/api/latest/guide/retries.html
+    transient_errors = {
+        "BandwidthLimitExceeded",
+        "ConnectionError",
+        "EC2ThrottledException",
+        "HTTPClientError",
+        "InternalError",
+        "InternalFailure",
+        "LimitExceededException",
+        "PriorRequestNotComplete",
+        "ProvisionedThroughputExceededException",
+        "RequestLimitExceeded",
+        "RequestThrottled",
+        "RequestThrottledException",
+        "RequestTimeout",
+        "RequestTimeoutException",
+        "ServerError",
+        "ServiceUnavailable",
+        "SlowDown",
+        "ThrottledException",
+        "Throttling",
+        "ThrottlingException",
+        "TooManyRequestsException",
+        "TransactionInProgressException",
+        "Unavailable",
+    }
+
+    # Fatal/unrecoverable → 400
+    fatal_errors = {
+        "BucketAlreadyExists",
+        "BucketAlreadyOwnedByYou",
+        "DryRunOperation",
+        "InvalidClientTokenId",
+        "InvalidParameterCombination",
+        "InvalidParameterValue",
+        "InvalidQueryParameter",
+        "MalformedPolicyDocument",
+        "MalformedQueryString",
+        "MethodNotAllowed",
+        "MissingParameter",
+        "OperationAborted",
+        "OptInRequired",
+        "UnsupportedOperation",
+        "UnsupportedProtocol",
+        "ValidationException",
+    }
+
+    if error_code in permission_errors:
+        return 403
+    elif error_code in not_found_errors:
+        return 404
+    elif error_code in range_errors:
+        return 416
+    elif error_code in fatal_errors:
+        return 400
+    elif error_code in transient_errors:
+        return 503
+
+    # Default: return original string code if unmapped
     return error_code
 
 
@@ -199,6 +258,8 @@ def op_info(url):
                 to_return = {"error": ERROR_URL_ACCESS_DENIED, "raise_error": err}
             elif error_code == 416:
                 to_return = {"error": ERROR_INVALID_RANGE, "raise_error": err}
+            elif error_code == 400:
+                to_return = {"error": ERROR_INVALID_REQUEST, "raise_error": err}
             elif error_code in (500, 502, 503, 504):
                 to_return = {"error": ERROR_TRANSIENT, "raise_error": err}
             else:
@@ -392,6 +453,9 @@ def handle_client_error(err, idx, result_file):
     elif error_code == 403:
         result_file.write("%d %d\n" % (idx, -ERROR_URL_ACCESS_DENIED))
         result_file.flush()
+    elif error_code == 400:
+        result_file.write("%d %d\n" % (idx, -ERROR_INVALID_REQUEST))
+        result_file.flush()
     elif error_code == 503:
         result_file.write("%d %d\n" % (idx, -ERROR_TRANSIENT))
         result_file.flush()
@@ -564,6 +628,8 @@ def get_info(self, url):
                 return False, url, ERROR_URL_NOT_FOUND
             elif error_code == 403:
                 return False, url, ERROR_URL_ACCESS_DENIED
+            elif error_code == 400:
+                return False, url, ERROR_INVALID_REQUEST
             # Transient errors are going to be retried by the aws_retry decorator
             else:
                 raise
@@ -612,6 +678,8 @@ def list_prefix(self, prefix_url, delimiter=""):
                 return False, prefix_url, ERROR_URL_NOT_FOUND
             elif error_code == 403:
                 return False, prefix_url, ERROR_URL_ACCESS_DENIED
+            elif error_code == 400:
+                return False, prefix_url, ERROR_INVALID_REQUEST
             # Transient errors are going to be retried by the aws_retry decorator
             else:
                 raise
@@ -655,6 +723,8 @@ def exit(exit_code, url):
         msg = "Transient error for url: %s" % url
     elif exit_code == ERROR_OUT_OF_DISK_SPACE:
         msg = "Out of disk space when downloading URL: %s" % url
+    elif exit_code == ERROR_INVALID_REQUEST:
+        msg = "Invalid request for URL: %s" % url
     else:
         msg = "Unknown error"
     print("s3op failed:\n%s" % msg, file=sys.stderr)