Backup to swift using wrong large object type

Trove backups are stored in Swift as Dynamic Large Objects. This is not
ideal, and could cause restore to fail as DLOs may not return all of
their parts. Static Large Objects should be used instead. SLOs perform
checksum and object size validation and can reduce object download times
significantly.

Changes are needed to the Swift storage strategy and taskmanager's
backup delete model to perform actions on SLOs. The corresponding
unittests and fakes are also updated.

A special case occurs when the backup object fits in a single Swift
object segment, in which case it will not be made a SLO.

Change-Id: Ide30a1e7f909a3b7fecb16268367b56468306d42
Closes-bug: 1489997

releasenotes/notes/slo-backups-3c35135316f837e1.yaml

@@ -0,0 +1,9 @@
+---
+fixes:
+  - Backups to Swift will now use Static Large Objects
+    for larger backups. A new configuration option
+    'backup_segment_max_size' can be set to adjust the
+    segment size of the SLO. Backups that are smaller
+    than the segment size will be uploaded as regular
+    objects. This is an improvement over old Dynamic
+    Large Object implementation. Bug 1489997.

trove/common/strategies/storage/base.py

@@ -28,7 +28,7 @@ class Storage(Strategy):
         super(Storage, self).__init__()
 
     @abc.abstractmethod
-    def save(self, filename, stream):
+    def save(self, filename, stream, metadata=None):
         """Persist information from the stream."""
 
     @abc.abstractmethod

trove/common/strategies/storage/swift.py

@@ -15,6 +15,7 @@
 #
 
 import hashlib
+import json
 
 from oslo_log import log as logging
 
@@ -64,8 +65,12 @@ class StreamReader(object):
         return '%s_%08d' % (self.base_filename, self.file_number)
 
     @property
-    def prefix(self):
+    def first_segment(self):
-        return '%s/%s_' % (self.container, self.base_filename)
+        return '%s_%08d' % (self.base_filename, 0)
+
+    @property
+    def segment_path(self):
+        return '%s/%s' % (self.container, self.segment)
 
     def read(self, chunk_size=CHUNK_SIZE):
         if self.end_of_segment:
@@ -97,29 +102,40 @@ class SwiftStorage(base.Storage):
         super(SwiftStorage, self).__init__(*args, **kwargs)
         self.connection = create_swift_client(self.context)
 
-    def save(self, filename, stream):
+    def save(self, filename, stream, metadata=None):
         """Persist information from the stream to swift.
 
         The file is saved to the location <BACKUP_CONTAINER>/<filename>.
+        It will be a Swift Static Large Object (SLO).
         The filename is defined on the backup runner manifest property
         which is typically in the format '<backup_id>.<ext>.gz'
         """
 
+        LOG.info(_('Saving %(filename)s to %(container)s in swift.')
+                 % {'filename': filename, 'container': BACKUP_CONTAINER})
+
         # Create the container if it doesn't already exist
+        LOG.debug('Creating container %s.' % BACKUP_CONTAINER)
         self.connection.put_container(BACKUP_CONTAINER)
 
         # Swift Checksum is the checksum of the concatenated segment checksums
         swift_checksum = hashlib.md5()
 
         # Wrap the output of the backup process to segment it for swift
-        stream_reader = StreamReader(stream, filename)
+        stream_reader = StreamReader(stream, filename, MAX_FILE_SIZE)
+        LOG.debug('Using segment size %s' % stream_reader.max_file_size)
 
         url = self.connection.url
         # Full location where the backup manifest is stored
         location = "%s/%s/%s" % (url, BACKUP_CONTAINER, filename)
 
+        # Information about each segment upload job
+        segment_results = []
+
         # Read from the stream and write to the container in swift
         while not stream_reader.end_of_file:
+            LOG.debug('Saving segment %s.' % stream_reader.segment)
+            path = stream_reader.segment_path
             etag = self.connection.put_object(BACKUP_CONTAINER,
                                               stream_reader.segment,
                                               stream_reader)
@@ -134,30 +150,71 @@ class SwiftStorage(base.Storage):
                           {'tag': etag, 'checksum': segment_checksum})
                 return False, "Error saving data to Swift!", None, location
 
+            segment_results.append({
+                'path': path,
+                'etag': etag,
+                'size_bytes': stream_reader.segment_length
+            })
+
             swift_checksum.update(segment_checksum)
 
-        # Create the manifest file
+        # All segments uploaded.
-        # We create the manifest file after all the segments have been uploaded
+        num_segments = len(segment_results)
-        # so a partial swift object file can't be downloaded; if the manifest
+        LOG.debug('File uploaded in %s segments.' % num_segments)
-        # file exists then all segments have been uploaded so the whole backup
-        # file can be downloaded.
-        headers = {'X-Object-Manifest': stream_reader.prefix}
-        # The etag returned from the manifest PUT is the checksum of the
-        # manifest object (which is empty); this is not the checksum we want
-        self.connection.put_object(BACKUP_CONTAINER,
-                                   filename,
-                                   contents='',
-                                   headers=headers)
 
+        # An SLO will be generated if the backup was more than one segment in
+        # length.
+        large_object = num_segments > 1
+
+        # Meta data is stored as headers
+        if metadata is None:
+            metadata = {}
+        metadata.update(stream.metadata())
+        headers = {}
+        for key, value in metadata.iteritems():
+            headers[self._set_attr(key)] = value
+
+        LOG.debug('Metadata headers: %s' % str(headers))
+        if large_object:
+            LOG.info(_('Creating the manifest file.'))
+            manifest_data = json.dumps(segment_results)
+            LOG.debug('Manifest contents: %s' % manifest_data)
+            # The etag returned from the manifest PUT is the checksum of the
+            # manifest object (which is empty); this is not the checksum we
+            # want.
+            self.connection.put_object(BACKUP_CONTAINER,
+                                       filename,
+                                       manifest_data,
+                                       query_string='multipart-manifest=put')
+
+            # Validation checksum is the Swift Checksum
+            final_swift_checksum = swift_checksum.hexdigest()
+        else:
+            LOG.info(_('Backup fits in a single segment. Moving segment '
+                       '%(segment)s to %(filename)s.')
+                     % {'segment': stream_reader.first_segment,
+                        'filename': filename})
+            segment_result = segment_results[0]
+            # Just rename it via a special put copy.
+            headers['X-Copy-From'] = segment_result['path']
+            self.connection.put_object(BACKUP_CONTAINER,
+                                       filename, '',
+                                       headers=headers)
+            # Delete the old segment file that was copied
+            LOG.debug('Deleting the old segment file %s.'
+                      % stream_reader.first_segment)
+            self.connection.delete_object(BACKUP_CONTAINER,
+                                          stream_reader.first_segment)
+            final_swift_checksum = segment_result['etag']
+
+        # Validate the object by comparing checksums
+        # Get the checksum according to Swift
         resp = self.connection.head_object(BACKUP_CONTAINER, filename)
         # swift returns etag in double quotes
         # e.g. '"dc3b0827f276d8d78312992cc60c2c3f"'
         etag = resp['etag'].strip('"')
 
-        # Check the checksum of the concatenated segment checksums against
-        # swift manifest etag.
         # Raise an error and mark backup as failed
-        final_swift_checksum = swift_checksum.hexdigest()
         if etag != final_swift_checksum:
             LOG.error(
                 _("Error saving data to swift. Manifest "
@@ -229,8 +286,7 @@ class SwiftStorage(base.Storage):
 
         storage_url, container, filename = self._explodeLocation(location)
 
-        _headers = self.connection.head_object(container, filename)
+        headers = {}
-        headers = {'X-Object-Manifest': _headers.get('x-object-manifest')}
         for key, value in metadata.items():
             headers[self._set_attr(key)] = value
 

trove/guestagent/backup/backupagent.py

@@ -80,9 +80,13 @@ class BackupAgent(object):
                     **parent_metadata) as bkup:
             try:
                 LOG.debug("Starting backup %s.", backup_id)
+                meta = {}
+                meta['datastore'] = backup_info['datastore']
+                meta['datastore_version'] = backup_info['datastore_version']
                 success, note, checksum, location = storage.save(
                     bkup.manifest,
-                    bkup)
+                    bkup,
+                    metadata=meta)
 
                 backup_state.update({
                     'checksum': checksum,
@@ -102,12 +106,6 @@ class BackupAgent(object):
                 if not success:
                     raise BackupError(note)
 
-                meta = bkup.metadata()
-                meta['datastore'] = backup_info['datastore']
-                meta['datastore_version'] = backup_info[
-                    'datastore_version']
-                storage.save_metadata(location, meta)
-
                 backup_state.update({'state': BackupState.COMPLETED})
 
                 return meta

trove/taskmanager/models.py

@@ -1372,25 +1372,17 @@ class BackupTasks(object):
         container = CONF.backup_swift_container
         client = remote.create_swift_client(context)
         obj = client.head_object(container, filename)
-        manifest = obj.get('x-object-manifest', '')
+        if 'x-static-large-object' in obj:
-        cont, prefix = cls._parse_manifest(manifest)
+            # Static large object
-        if all([cont, prefix]):
+            LOG.debug("Deleting large object file: %(cont)s/%(filename)s" %
-            # This is a manifest file, first delete all segments.
+                      {'cont': container, 'filename': filename})
-            LOG.debug("Deleting files with prefix: %(cont)s/%(prefix)s" %
+            client.delete_object(container, filename,
-                      {'cont': cont, 'prefix': prefix})
+                                 query_string='multipart-manifest=delete')
-            # list files from container/prefix specified by manifest
+        else:
-            headers, segments = client.get_container(cont, prefix=prefix)
+            # Single object
-            LOG.debug(headers)
+            LOG.debug("Deleting object file: %(cont)s/%(filename)s" %
-            for segment in segments:
+                      {'cont': container, 'filename': filename})
-                name = segment.get('name')
+            client.delete_object(container, filename)
-                if name:
-                    LOG.debug("Deleting file: %(cont)s/%(name)s" %
-                              {'cont': cont, 'name': name})
-                    client.delete_object(cont, name)
-        # Delete the manifest file
-        LOG.debug("Deleting file: %(cont)s/%(filename)s" %
-                  {'cont': cont, 'filename': filename})
-        client.delete_object(container, filename)
 
     @classmethod
     def delete_backup(cls, context, backup_id):

trove/tests/fakes/swift.py

@@ -44,7 +44,9 @@ class FakeSwiftClient(object):
 
 class FakeSwiftConnection(object):
     """Logging calls instead of executing."""
-    MANIFEST_HEADER_KEY = 'X-Object-Manifest'
+    MANIFEST_QUERY_STRING_PUT = 'multipart-manifest=put'
+    MANIFEST_QUERY_STRING_DELETE = 'multipart-manifest=delete'
+    COPY_OBJECT_HEADER_KEY = 'X-Copy-From'
     url = 'http://mockswift/v1'
 
     def __init__(self, *args, **kwargs):
@@ -101,7 +103,7 @@ class FakeSwiftConnection(object):
         LOG.debug("fake put_container(%(container)s, %(name)s)" %
                   {'container': container, 'name': name})
         checksum = md5()
-        if self.manifest_prefix and self.manifest_name == name:
+        if self.manifest_name == name:
             for object_name in sorted(self.container_objects):
                 object_checksum = md5(self.container_objects[object_name])
                 # The manifest file etag for a HEAD or GET is the checksum of
@@ -163,14 +165,19 @@ class FakeSwiftConnection(object):
         if container == 'socket_error_on_put':
             raise socket.error(111, 'ECONNREFUSED')
         headers = kwargs.get('headers', {})
+        query_string = kwargs.get('query_string', '')
         object_checksum = md5()
-        if self.MANIFEST_HEADER_KEY in headers:
+        if query_string == self.MANIFEST_QUERY_STRING_PUT:
             # the manifest prefix format is <container>/<prefix> where
             # container is where the object segments are in and prefix is the
             # common prefix for all segments.
-            self.manifest_prefix = headers.get(self.MANIFEST_HEADER_KEY)
             self.manifest_name = name
             object_checksum.update(contents)
+        elif self.COPY_OBJECT_HEADER_KEY in headers:
+            # this is a copy object operation
+            source_path = headers.get(self.COPY_OBJECT_HEADER_KEY)
+            source_name = source_path.split('/')[1]
+            self.container_objects[name] = self.container_objects[source_name]
         else:
             if hasattr(contents, 'read'):
                 chunk_size = 128

trove/tests/unittests/backup/test_backupagent.py

@@ -106,7 +106,7 @@ class MockSwift(object):
         self.etag.update(self.store)
         return self.etag.hexdigest()
 
-    def save(self, filename, stream):
+    def save(self, filename, stream, metadata=None):
         location = '%s/%s/%s' % (self.url, self.container, filename)
         return True, 'w00t', 'fake-checksum', location
 
@@ -128,7 +128,7 @@ class MockStorage(Storage):
     def load(self, location, backup_checksum):
         pass
 
-    def save(self, filename, stream):
+    def save(self, filename, stream, metadata=None):
         pass
 
     def load_metadata(self, location, checksum):
@@ -407,6 +407,8 @@ class BackupAgentTest(trove_testtools.TestCase):
                            'location': 'fake-location',
                            'type': 'InnoBackupEx',
                            'checksum': 'fake-checksum',
+                           'datastore': 'mysql',
+                           'datastore_version': '5.5'
                            }
 
             self.assertRaises(backupagent.BackupError, agent.execute_backup,

trove/tests/unittests/backup/test_storage.py

@@ -32,12 +32,47 @@ class SwiftStorageSaveChecksumTests(trove_testtools.TestCase):
 
     def setUp(self):
         super(SwiftStorageSaveChecksumTests, self).setUp()
+        self.max_file_size = swift.MAX_FILE_SIZE
+        swift.MAX_FILE_SIZE = 128
 
     def tearDown(self):
+        swift.MAX_FILE_SIZE = self.max_file_size
         super(SwiftStorageSaveChecksumTests, self).tearDown()
 
+    def test_swift_small_file_checksum_save(self):
+        """This tests that SwiftStorage.save returns the swift checksum
+        for small files.
+        """
+        context = trove_testtools.TroveTestContext(self)
+        backup_id = '123'
+        user = 'user'
+        password = 'password'
+
+        swift.MAX_FILE_SIZE = 2 * (1024 ** 3)
+
+        swift_client = FakeSwiftConnection()
+        with patch.object(swift, 'create_swift_client',
+                          return_value=swift_client):
+            storage_strategy = SwiftStorage(context)
+
+            with MockBackupRunner(filename=backup_id,
+                                  user=user,
+                                  password=password) as runner:
+                (success,
+                 note,
+                 checksum,
+                 location) = storage_strategy.save(runner.manifest, runner)
+
+        self.assertTrue(success, "The backup should have been successful.")
+        self.assertIsNotNone(note, "A note should have been returned.")
+        self.assertEqual('http://mockswift/v1/database_backups/123.gz.enc',
+                         location,
+                         "Incorrect swift location was returned.")
+
     def test_swift_checksum_save(self):
-        """This tests that SwiftStorage.save returns the swift checksum."""
+        """This tests that SwiftStorage.save returns the swift checksum for
+        large files.
+        """
         context = trove_testtools.TroveTestContext(self)
         backup_id = '123'
         user = 'user'
@@ -240,9 +275,6 @@ class StreamReaderTests(trove_testtools.TestCase):
         stream_reader = StreamReader(self.runner, 'foo')
         self.assertEqual('foo', stream_reader.base_filename)
 
-    def test_prefix(self):
-        self.assertEqual('database_backups/123_', self.stream.prefix)
-
     def test_segment(self):
         self.assertEqual('123_00000000', self.stream.segment)
 
@@ -326,8 +358,7 @@ class SwiftMetadataTests(trove_testtools.TestCase):
         self.swift.save_metadata(location, metadata=metadata)
 
         headers = {
-            'X-Object-Meta-lsn': '1234567',
+            'X-Object-Meta-lsn': '1234567'
-            'X-Object-Manifest': None
         }
         self.swift_client.post_object.assert_called_with(
             'backups', 'mybackup.tar', headers=headers)