BulkDeleter: normalize return keys and support > 10000 objects

pyrax/object_storage.py

@@ -55,6 +55,8 @@
 DEFAULT_CDN_TTL = 86400
 # When comparing files dates, represents a date older than anything.
 EARLY_DATE_STR = "1900-01-01T00:00:00"
+# Maximum number of objects that can be passed to bulk-delete
+MAX_BULK_DELETE = 10000
 
 # Used to indicate values that are lazy-loaded
 class Fault_cls(object):
@@ -850,7 +852,7 @@ def delete(self, container, del_objects=False):
         each object will be deleted first, and then the container.
         """
         if del_objects:
-            nms = self.list_object_names(container)
+            nms = self.list_object_names(container, full_listing=True)
             self.api.bulk_delete(container, nms, async=False)
         uri = "/%s" % utils.get_name(container)
         resp, resp_body = self.api.method_delete(uri)
@@ -2044,7 +2046,7 @@ def delete_all_objects(self, nms, async=False):
             errors - a list of any errors returned by the bulk delete call
         """
         if nms is None:
-            nms = self.api.list_object_names(self.name)
+            nms = self.api.list_object_names(self.name, full_listing=True)
         return self.api.bulk_delete(self.name, nms, async=async)
 
 
@@ -3312,7 +3314,12 @@ def __init__(self, client, container, object_names):
         self.container = container
         self.object_names = object_names
         self.completed = False
-        self.results = None
+        self.results = {
+            "deleted": 0,
+            "not_found": 0,
+            "status": "",
+            "errors": []
+        }
         threading.Thread.__init__(self)
 
 
@@ -3322,14 +3329,40 @@ def run(self):
         object_names = self.object_names
         cname = utils.get_name(container)
         ident = self.client.identity
-        headers = {"X-Auth-Token": ident.token,
-                "Content-Type": "text/plain",
-                }
-        obj_paths = ("%s/%s" % (cname, nm) for nm in object_names)
-        body = "\n".join(obj_paths)
+        headers = {
+            "X-Auth-Token": ident.token,
+            "Content-Type": "text/plain",
+        }
         uri = "/?bulk-delete=1"
-        resp, resp_body = self.client.method_delete(uri, data=body,
-                headers=headers)
-        status = resp_body.get("Response Status", "").split(" ")[0]
-        self.results = resp_body
+        key_map = {
+            "Number Not Found": "not_found",
+            "Response Status": "status",
+            "Errors": "errors",
+            "Number Deleted": "deleted",
+            "Response Body": None,
+        }
+        batch = []
+        while object_names:
+            batch[:] = object_names[:MAX_BULK_DELETE]
+            del object_names[:MAX_BULK_DELETE]
+
+            obj_paths = ("%s/%s" % (cname, nm) for nm in batch)
+            body = "\n".join(obj_paths)
+            resp, resp_body = self.client.method_delete(uri, data=body,
+                    headers=headers)
+            for k, v in six.iteritems(resp_body):
+                if key_map[k] == "errors":
+                    status_code = int(resp_body.get("Response Status", "200")[:3])
+                    if status_code != 200 and not v:
+                        self.results["errors"].extend([[
+                            resp_body.get("Response Body"),
+                            resp_body.get("Response Status")
+                        ]])
+                    else:
+                        self.results["errors"].extend(v)
+                elif key_map[k] in ("deleted", "not_found"):
+                    self.results[key_map[k]] += int(v)
+                elif key_map[k]:
+                    self.results[key_map[k]] = v
+
         self.completed = True

tests/unit/test_object_storage.py

@@ -776,7 +776,7 @@ def test_cmgr_delete(self):
         exp_uri = "/%s" % cont.name
         mgr.api.method_delete = Mock(return_value=(None, None))
         mgr.delete(cont, del_objects=True)
-        mgr.list_object_names.assert_called_once_with(cont)
+        mgr.list_object_names.assert_called_once_with(cont, full_listing=True)
         mgr.api.bulk_delete.assert_called_once_with(cont, names, async=False)
         mgr.api.method_delete.assert_called_once_with(exp_uri)
 
@@ -2175,7 +2175,8 @@ def test_sobj_mgr_delete_all_objects_no_names(self):
         mgr.api.list_object_names = Mock(return_value=nms)
         mgr.api.bulk_delete = Mock()
         mgr.delete_all_objects(None, async=async)
-        mgr.api.list_object_names.assert_called_once_with(mgr.name)
+        mgr.api.list_object_names.assert_called_once_with(mgr.name,
+                                                          full_listing=True)
         mgr.api.bulk_delete.assert_called_once_with(mgr.name, nms, async=async)
 
     def test_sobj_mgr_download_no_directory(self):
@@ -3313,7 +3314,19 @@ def test_clt_bulk_delete_sync(self):
         obj_names = ["test1", "test2"]
         resp = fakes.FakeResponse()
         fake_res = utils.random_unicode()
-        body = {"Response Status": "foo " + fake_res}
+        body = {
+            "Number Not Found": 1,
+            "Response Status": "200 OK",
+            "Errors": [],
+            "Number Deleted": 10,
+            "Response Body": ""
+        }
+        expected = {
+            'deleted': 10,
+            'errors': [],
+            'not_found': 1,
+            'status': '200 OK'
+        }
         clt.bulk_delete_interval = 0.01
 
         def fake_bulk_resp(uri, data=None, headers=None):
@@ -3322,7 +3335,41 @@ def fake_bulk_resp(uri, data=None, headers=None):
 
         clt.method_delete = Mock(side_effect=fake_bulk_resp)
         ret = clt.bulk_delete(cont, obj_names, async=False)
-        self.assertEqual(ret, body)
+        self.assertEqual(ret, expected)
+
+    def test_clt_bulk_delete_sync_413(self):
+        clt = self.client
+        cont = self.container
+        obj_names = ["test1", "test2"]
+        resp = fakes.FakeResponse()
+        fake_res = utils.random_unicode()
+        body = {
+            "Number Not Found": 0,
+            "Response Status": "413 Request Entity Too Large",
+            "Errors": [],
+            "Number Deleted": 0,
+            "Response Body": "Maximum Bulk Deletes: 10000 per request"
+        }
+        expected = {
+            'deleted': 0,
+            'errors': [
+                [
+                    'Maximum Bulk Deletes: 10000 per request',
+                    '413 Request Entity Too Large'
+                ]
+            ],
+            'not_found': 0,
+            'status': '413 Request Entity Too Large'
+        }
+        clt.bulk_delete_interval = 0.01
+
+        def fake_bulk_resp(uri, data=None, headers=None):
+            time.sleep(0.05)
+            return (resp, body)
+
+        clt.method_delete = Mock(side_effect=fake_bulk_resp)
+        ret = clt.bulk_delete(cont, obj_names, async=False)
+        self.assertEqual(ret, expected)
 
     def test_clt_cdn_request_not_enabled(self):
         clt = self.client