Fix KeyError when auditor finds an expired object

Since the related change [1] the auditor checks deleted objects in
case their tombstone is ready to be reclaimed. However, the auditor
mistakes an expired object for a deleted object because
DiskFileExpired is a subclass of DiskFileDeleted. This causes a
KeyError to be raised and logged because the expired object has no
tombstone info in the ondisk_info data structure.

This patch makes the auditor catch and ignore DiskFileExpired
exceptions before handling DiskFileDeleted exceptions.

[1] Related-Change: I3e99dc702d55a7424c6482969e03cb4afac854a4

Change-Id: I9872b6997d09dcfd8a868c4b91b9379407283b8e
Closes-Bug: #1638016

swift/obj/auditor.py

@@ -30,7 +30,7 @@ from swift.common.utils import (
     get_logger, ratelimit_sleep, dump_recon_cache, list_from_csv, listdir,
     unlink_paths_older_than, readconf, config_auto_int_value)
 from swift.common.exceptions import DiskFileQuarantined, DiskFileNotExist,\
-    DiskFileDeleted
+    DiskFileDeleted, DiskFileExpired
 from swift.common.daemon import Daemon
 from swift.common.storage_policy import POLICIES
 
@@ -265,6 +265,8 @@ class AuditorWorker(object):
             self.logger.error(_('ERROR Object %(obj)s failed audit and was'
                                 ' quarantined: %(err)s'),
                               {'obj': location, 'err': err})
+        except DiskFileExpired:
+            pass  # ignore expired objects
         except DiskFileDeleted:
             # If there is a reclaimable tombstone, we'll invalidate the hash
             # to trigger the replicator to rehash/cleanup this suffix

test/unit/obj/test_auditor.py

@@ -35,7 +35,7 @@ from swift.common.utils import (
     mkdirs, normalize_timestamp, Timestamp, readconf)
 from swift.common.storage_policy import (
     ECStoragePolicy, StoragePolicy, POLICIES, EC_POLICY)
-
+from test.unit.obj.common import write_diskfile
 
 _mocked_policies = [
     StoragePolicy(0, 'zero', False),
@@ -900,6 +900,32 @@ class TestAuditor(unittest.TestCase):
         self.assertFalse(os.path.exists(
             os.path.join(part_dir, HASH_INVALIDATIONS_FILE)))
 
+    def _test_expired_object_is_ignored(self, zero_byte_fps):
+        # verify that an expired object does not get mistaken for a tombstone
+        audit = auditor.ObjectAuditor(self.conf)
+        audit.logger = FakeLogger()
+        audit.log_time = 0
+        now = time.time()
+        write_diskfile(self.disk_file, Timestamp(now - 20),
+                       extra_metadata={'X-Delete-At': now - 10})
+        files = os.listdir(self.disk_file._datadir)
+        self.assertTrue([f for f in files if f.endswith('.data')])  # sanity
+        with mock.patch.object(auditor, 'dump_recon_cache'):
+            audit.run_audit(mode='once', zero_byte_fps=zero_byte_fps)
+        self.assertTrue(os.path.exists(self.disk_file._datadir))
+        part_dir = dirname(dirname(self.disk_file._datadir))
+        self.assertFalse(os.path.exists(
+            os.path.join(part_dir, HASH_INVALIDATIONS_FILE)))
+        self.assertEqual(files, os.listdir(self.disk_file._datadir))
+        self.assertFalse(audit.logger.get_lines_for_level('error'))
+        self.assertFalse(audit.logger.get_lines_for_level('warning'))
+
+    def test_expired_object_is_ignored(self):
+        self._test_expired_object_is_ignored(0)
+
+    def test_expired_object_is_ignored_with_zero_byte_fps(self):
+        self._test_expired_object_is_ignored(50)
+
     def test_auditor_reclaim_age(self):
         # if we don't have access to the replicator config section we'll use
         # diskfile's default