Merge "Add unit test for diskfile.relink_paths"

4 files changed, 140 insertions, 9 deletions

diff --git a/swift/common/utils.py b/swift/common/utils.py
index 56a297839..e91f67e16 100644
--- a/swift/common/utils.py
+++ b/swift/common/utils.py
@@ -5780,6 +5780,18 @@ def md5_hash_for_file(fname):
     return md5sum.hexdigest()
 
 
+def get_partition_for_hash(hex_hash, part_power):
+    """
+    Return partition number for given hex hash and partition power.
+    :param hex_hash: A hash string
+    :param part_power: partition power
+    :returns: partition number
+    """
+    raw_hash = binascii.unhexlify(hex_hash)
+    part_shift = 32 - int(part_power)
+    return struct.unpack_from('>I', raw_hash)[0] >> part_shift
+
+
 def replace_partition_in_path(path, part_power):
     """
     Takes a full path to a file and a partition power and returns
@@ -5790,15 +5802,9 @@ def replace_partition_in_path(path, part_power):
     :param part_power: partition power to compute correct partition number
     :returns: Path with re-computed partition power
     """
-
     path_components = path.split(os.sep)
-    digest = binascii.unhexlify(path_components[-2])
-
-    part_shift = 32 - int(part_power)
-    part = struct.unpack_from('>I', digest)[0] >> part_shift
-
+    part = get_partition_for_hash(path_components[-2], part_power)
     path_components[-4] = "%d" % part
-
     return os.sep.join(path_components)
 
 
diff --git a/test/unit/common/test_utils.py b/test/unit/common/test_utils.py
index cc2785d2f..bbaa21bf2 100644
--- a/test/unit/common/test_utils.py
+++ b/test/unit/common/test_utils.py
@@ -4328,6 +4328,14 @@ cluster_dfw1 = http://dfw1.host/v1/
             self.fail('Invalid results from pure function:\n%s' %
                       '\n'.join(failures))
 
+    def test_get_partition_for_hash(self):
+        hex_hash = 'af088baea4806dcaba30bf07d9e64c77'
+        self.assertEqual(43, utils.get_partition_for_hash(hex_hash, 6))
+        self.assertEqual(87, utils.get_partition_for_hash(hex_hash, 7))
+        self.assertEqual(350, utils.get_partition_for_hash(hex_hash, 9))
+        self.assertEqual(700, utils.get_partition_for_hash(hex_hash, 10))
+        self.assertEqual(1400, utils.get_partition_for_hash(hex_hash, 11))
+
     def test_replace_partition_in_path(self):
         # Check for new part = part * 2
         old = '/s/n/d/o/700/c77/af088baea4806dcaba30bf07d9e64c77/f'
diff --git a/test/unit/obj/test_diskfile.py b/test/unit/obj/test_diskfile.py
index 70b5e876b..b80ce3f89 100644
--- a/test/unit/obj/test_diskfile.py
+++ b/test/unit/obj/test_diskfile.py
@@ -262,6 +262,29 @@ class TestDiskFileModuleMethods(unittest.TestCase):
         with open(new_target_path_2, 'r') as fd:
             self.assertEqual(target_path_2, fd.read())
 
+    def test_relink_paths_object_dir_exists_but_not_dir(self):
+        target_dir = os.path.join(self.testdir, 'd1')
+        os.mkdir(target_dir)
+        target_path = os.path.join(target_dir, 't1.data')
+        with open(target_path, 'w') as fd:
+            fd.write(target_path)
+        # make a file where the new object dir should be
+        new_target_dir = os.path.join(self.testdir, 'd2')
+        with open(new_target_dir, 'w') as fd:
+            fd.write(new_target_dir)
+        new_target_path = os.path.join(new_target_dir, 't1.data')
+
+        with self.assertRaises(OSError) as cm:
+            diskfile.relink_paths(target_path, new_target_path)
+        self.assertEqual(errno.ENOTDIR, cm.exception.errno)
+
+        # make a symlink to target where the new object dir should be
+        os.unlink(new_target_dir)
+        os.symlink(target_path, new_target_dir)
+        with self.assertRaises(OSError) as cm:
+            diskfile.relink_paths(target_path, new_target_path)
+        self.assertEqual(errno.ENOTDIR, cm.exception.errno)
+
     def test_extract_policy(self):
         # good path names
         pn = 'objects/0/606/1984527ed7ef6247c78606/1401379842.14643.data'
diff --git a/test/unit/obj/test_server.py b/test/unit/obj/test_server.py
index a9f87dacb..a041d4f5f 100644
--- a/test/unit/obj/test_server.py
+++ b/test/unit/obj/test_server.py
@@ -147,7 +147,7 @@ class TestObjectController(unittest.TestCase):
         mkdirs(os.path.join(self.testdir, 'sda1'))
         self.conf = {'devices': self.testdir, 'mount_check': 'false',
                      'container_update_timeout': 0.0}
-        self.logger = debug_logger()
+        self.logger = debug_logger('test-object-controller')
         self.object_controller = object_server.ObjectController(
             self.conf, logger=self.logger)
         self.object_controller.bytes_per_sync = 1
@@ -156,7 +156,6 @@ class TestObjectController(unittest.TestCase):
         self.df_mgr = diskfile.DiskFileManager(self.conf,
                                                self.object_controller.logger)
 
-        self.logger = debug_logger('test-object-controller')
         self.ts = make_timestamp_iter()
         self.ec_policies = [p for p in POLICIES if p.policy_type == EC_POLICY]
 
@@ -2834,6 +2833,101 @@ class TestObjectController(unittest.TestCase):
                             % (data_file, os.listdir(obj_dir), int(policy)))
             rmtree(obj_dir)
 
+    def test_PUT_next_part_power(self):
+        hash_path_ = hash_path('a', 'c', 'o')
+        part_power = 10
+        old_part = utils.get_partition_for_hash(hash_path_, part_power)
+        new_part = utils.get_partition_for_hash(hash_path_, part_power + 1)
+        policy = POLICIES.default
+        timestamp = utils.Timestamp(int(time())).internal
+        headers = {'X-Timestamp': timestamp,
+                   'Content-Length': '6',
+                   'Content-Type': 'application/octet-stream',
+                   'X-Backend-Storage-Policy-Index': int(policy),
+                   'X-Backend-Next-Part-Power': part_power + 1}
+        req = Request.blank(
+            '/sda1/%s/a/c/o' % old_part, method='PUT',
+            headers=headers, body=b'VERIFY')
+        resp = req.get_response(self.object_controller)
+
+        self.assertEqual(resp.status_int, 201)
+
+        def check_file(part):
+            data_file = os.path.join(
+                self.testdir, 'sda1',
+                storage_directory(diskfile.get_data_dir(int(policy)),
+                                  part, hash_path_), timestamp + '.data')
+            self.assertTrue(os.path.isfile(data_file))
+
+        check_file(old_part)
+        check_file(new_part)
+
+    def test_PUT_next_part_power_races_around_makedirs_eexist(self):
+        # simulate two 'concurrent' racing to create the new object dir in the
+        # new partition and check that relinking tolerates the dir already
+        # existing when they attempt to create it
+        hash_path_ = hash_path('a', 'c', 'o')
+        part_power = 10
+        old_part = utils.get_partition_for_hash(hash_path_, part_power)
+        new_part = utils.get_partition_for_hash(hash_path_, part_power + 1)
+        policy = POLICIES.default
+
+        def make_request(timestamp):
+            headers = {'X-Timestamp': timestamp.internal,
+                       'Content-Length': '6',
+                       'Content-Type': 'application/octet-stream',
+                       'X-Backend-Storage-Policy-Index': int(policy),
+                       'X-Backend-Next-Part-Power': part_power + 1}
+            req = Request.blank(
+                '/sda1/%s/a/c/o' % old_part, method='PUT',
+                headers=headers, body=b'VERIFY')
+            resp = req.get_response(self.object_controller)
+            self.assertEqual(resp.status_int, 201)
+
+        def data_file(part, timestamp):
+            return os.path.join(
+                self.testdir, 'sda1',
+                storage_directory(diskfile.get_data_dir(int(policy)),
+                                  part, hash_path_),
+                timestamp.internal + '.data')
+
+        ts_1 = next(self.ts)
+        ts_2 = next(self.ts)
+        calls = []
+        orig_makedirs = os.makedirs
+
+        def mock_makedirs(path, *args, **kwargs):
+            # let another request catch up just as the first is about to create
+            # the next part power object dir, then pretend the first request
+            # process actually makes the dir
+            if path == os.path.dirname(data_file(new_part, ts_1)):
+                calls.append(path)
+                if len(calls) == 1:
+                    # pretend 'yield' to other request process
+                    make_request(ts_2)
+                if len(calls) == 2:
+                    # pretend 'yield' back to first request process for
+                    # its call to makedirs
+                    orig_makedirs(calls[0])
+            return orig_makedirs(path, *args, **kwargs)
+        with mock.patch('swift.obj.diskfile.os.makedirs', mock_makedirs):
+            make_request(ts_1)
+
+        self.assertEqual(
+            [os.path.dirname(data_file(new_part, ts_1)),
+             os.path.dirname(data_file(new_part, ts_1))], calls)
+        self.assertTrue(os.path.isfile(data_file(old_part, ts_2)))
+        self.assertTrue(os.path.isfile(data_file(new_part, ts_2)))
+        self.assertFalse(os.path.isfile(data_file(new_part, ts_1)))
+        self.assertFalse(os.path.isfile(data_file(old_part, ts_1)))
+        error_lines = self.logger.get_lines_for_level('error')
+        # the older request's data file in the old partition will have been
+        # cleaned up by the newer request, so it's attempt to relink will fail
+        self.assertEqual(1, len(error_lines))
+        self.assertIn(ts_1.internal + '.data failed: '
+                                      '[Errno 2] No such file or directory',
+                      error_lines[0])
+
     def test_HEAD(self):
         # Test swift.obj.server.ObjectController.HEAD
         req = Request.blank('/sda1/p/a/c', environ={'REQUEST_METHOD': 'HEAD'})