Merge pull request #2105 from coreos-inc/frack-swift

Fix swift exception reporting on deletion and add async chunk cleanup

test/test_cloud_storage.py

@@ -3,7 +3,7 @@ import moto
 import boto
 import os
 
-from storage import S3Storage
+from storage import S3Storage, StorageContext
 from storage.cloud import _CloudStorage, _PartUploadMetadata
 from storage.cloud import _CHUNKS_KEY
 from StringIO import StringIO
@@ -13,6 +13,7 @@ _TEST_BUCKET = 'some_bucket'
 _TEST_USER = 'someuser'
 _TEST_PASSWORD = 'somepassword'
 _TEST_PATH = 'some/cool/path'
+_TEST_CONTEXT = StorageContext('nyc', None, None)
 
 class TestCloudStorage(unittest.TestCase):
   def setUp(self):
@@ -21,7 +22,7 @@ class TestCloudStorage(unittest.TestCase):
 
     # Create a test bucket and put some test content.
     boto.connect_s3().create_bucket(_TEST_BUCKET)
-    self.engine = S3Storage(None, 'some/path', _TEST_BUCKET, _TEST_USER, _TEST_PASSWORD)
+    self.engine = S3Storage(_TEST_CONTEXT, 'some/path', _TEST_BUCKET, _TEST_USER, _TEST_PASSWORD)
     self.engine.put_content(_TEST_PATH, _TEST_CONTENT)
 
   def tearDown(self):
@@ -51,7 +52,8 @@ class TestCloudStorage(unittest.TestCase):
 
   def test_copy_samecreds(self):
     # Copy the content to another engine.
-    another_engine = S3Storage(None, 'another/path', _TEST_BUCKET, _TEST_USER, _TEST_PASSWORD)
+    another_engine = S3Storage(_TEST_CONTEXT, 'another/path', _TEST_BUCKET, _TEST_USER,
+                               _TEST_PASSWORD)
     self.engine.copy_to(another_engine, _TEST_PATH)
 
     # Verify it can be retrieved.
@@ -59,7 +61,8 @@ class TestCloudStorage(unittest.TestCase):
 
   def test_copy_differentcreds(self):
     # Copy the content to another engine.
-    another_engine = S3Storage(None, 'another/path', 'another_bucket', 'blech', 'password')
+    another_engine = S3Storage(_TEST_CONTEXT, 'another/path', 'another_bucket', 'blech',
+                               'password')
     boto.connect_s3().create_bucket('another_bucket')
 
     self.engine.copy_to(another_engine, _TEST_PATH)

test/test_swift.py

@@ -2,10 +2,11 @@ import io
 import unittest
 
 from collections import defaultdict
-
-from storage.swift import SwiftStorage
 from mock import MagicMock
 
+from storage import StorageContext
+from storage.swift import SwiftStorage
+
 
 class MockSwiftStorage(SwiftStorage):
   def __init__(self, *args, **kwargs):
@@ -18,7 +19,7 @@ class MockSwiftStorage(SwiftStorage):
 
 class MockSwiftTests(unittest.TestCase):
   base_args = {
-    'metric_queue': None,
+    'context': StorageContext('nyc', None, None),
     'swift_container': 'container-name',
     'storage_path': '/basepath',
     'auth_url': 'https://auth.com',
@@ -39,6 +40,11 @@ class MockSwiftTests(unittest.TestCase):
     swift.exists('object/path')
     swift._get_connection().head_object.assert_called_with('container-name', 'basepathobject/path')
 
+  def test_delete_unknown_path(self):
+    swift = SwiftStorage(**self.base_args)
+    with self.assertRaises(IOError):
+      swift.remove('someunknownpath')
+
 
 class FakeSwiftStorage(SwiftStorage):
   def __init__(self, *args, **kwargs):
@@ -88,9 +94,26 @@ class FakeSwift(object):
     self.containers[container].pop(path, None)
 
 
+class FakeQueue(object):
+  def __init__(self):
+    self.items = []
+
+  def get(self):
+    if not self.items:
+      return None
+
+    return self.items.pop()
+
+  def put(self, names, item, available_after=0):
+    self.items.append({
+      'names': names,
+      'item': item,
+      'available_after': available_after,
+    })
+
 class FakeSwiftTests(unittest.TestCase):
   base_args = {
-    'metric_queue': None,
+    'context': StorageContext('nyc', None, None),
     'swift_container': 'container-name',
     'storage_path': '/basepath',
     'auth_url': 'https://auth.com',
@@ -165,6 +188,36 @@ class FakeSwiftTests(unittest.TestCase):
     for segment in SwiftStorage._segment_list_from_metadata(metadata):
       self.assertFalse(swift.exists(segment.path))
 
+  def test_empty_chunks_queued_for_deletion(self):
+    chunk_cleanup_queue = FakeQueue()
+    args = dict(self.base_args)
+    args['context'] = StorageContext('nyc', None, chunk_cleanup_queue)
+
+    swift = FakeSwiftStorage(**args)
+    uuid, metadata = swift.initiate_chunked_upload()
+
+    chunks = ['this', '', 'is', 'some', '', 'chunked', 'data', '']
+    offset = 0
+    for chunk in chunks:
+      length = len(chunk)
+      if length == 0:
+        length = 1
+
+      bytes_written, metadata, error = swift.stream_upload_chunk(uuid, offset, length,
+                                                                 io.BytesIO(chunk), metadata)
+      self.assertIsNone(error)
+      self.assertEquals(bytes_written, len(chunk))
+      offset += len(chunk)
+
+    swift.complete_chunked_upload(uuid, 'somepath', metadata)
+    self.assertEquals(''.join(chunks), swift.get_content('somepath'))
+
+    # Check the chunk deletion queue and ensure we have the last chunk queued.
+    found = chunk_cleanup_queue.get()
+    self.assertIsNotNone(found)
+
+    found2 = chunk_cleanup_queue.get()
+    self.assertIsNone(found2)
 
 if __name__ == '__main__':
   unittest.main()