Terminate early on ENOSPC (#2392)

Running out of disk space should be treated as a fatal error. It is
currently treated as a transient error, leading to retries which never
succeed. The task eventually dies after all the retries are exhausted.

This change makes `ENOSPC` a fatal error and we exit immediately.

---------

Co-authored-by: Nissan Pow <npow@netflix.com>

Author: npow

metaflow/plugins/datatools/s3/s3.py

@@ -1,3 +1,4 @@
+import errno
 import json
 import os
 import re
@@ -137,6 +138,10 @@ class MetaflowS3InvalidRange(MetaflowException):
     headline = "S3 invalid range"
 
 
+class MetaflowS3InsufficientDiskSpace(MetaflowException):
+    headline = "Insufficient disk space"
+
+
 class S3Object(object):
     """
     This object represents a path or an object in S3,
@@ -1377,8 +1382,10 @@ def _one_boto_op(self, op, url, create_tmp_file=True):
                 elif error_code == "NoSuchBucket":
                     raise MetaflowS3URLException("Specified S3 bucket doesn't exist.")
                 error = str(err)
+            except OSError as e:
+                if e.errno == errno.ENOSPC:
+                    raise MetaflowS3InsufficientDiskSpace(str(e))
             except Exception as ex:
-                # TODO specific error message for out of disk space
                 error = str(ex)
             if tmp:
                 os.unlink(tmp.name)

metaflow/plugins/datatools/s3/s3op.py

@@ -1,5 +1,6 @@
 from __future__ import print_function
 
+import errno
 import json
 import time
 import math
@@ -108,6 +109,7 @@ def __str__(self):
 ERROR_LOCAL_FILE_NOT_FOUND = 10
 ERROR_INVALID_RANGE = 11
 ERROR_TRANSIENT = 12
+ERROR_OUT_OF_DISK_SPACE = 13
 
 
 def format_result_line(idx, prefix, url="", local=""):
@@ -277,6 +279,17 @@ def op_info(url):
                             err = convert_to_client_error(e)
                             handle_client_error(err, idx, result_file)
                             continue
+                        except OSError as e:
+                            tmp.close()
+                            os.unlink(tmp.name)
+                            if e.errno == errno.ENOSPC:
+                                result_file.write(
+                                    "%d %d\n" % (idx, -ERROR_OUT_OF_DISK_SPACE)
+                                )
+                            else:
+                                result_file.write("%d %d\n" % (idx, -ERROR_TRANSIENT))
+                            result_file.flush()
+                            continue
                     except (SSLError, Exception) as e:
                         tmp.close()
                         os.unlink(tmp.name)
@@ -643,6 +656,8 @@ def exit(exit_code, url):
         msg = "Local file not found: %s" % url
     elif exit_code == ERROR_TRANSIENT:
         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
     else:
         msg = "Unknown error"
     print("s3op failed:\n%s" % msg, file=sys.stderr)
@@ -1173,6 +1188,8 @@ def get(
             )
             if verify:
                 verify_info.append((url, sz))
+        elif sz == -ERROR_OUT_OF_DISK_SPACE:
+            exit(ERROR_OUT_OF_DISK_SPACE, url)
         elif sz == -ERROR_URL_ACCESS_DENIED:
             denied_url = url
             break