Merge lp:~david-goetz/swift/slow_audit into lp:~hudson-openstack/swift/trunk

[David Goetz]

Rate limit the object auditor, add unit tests, and fix a bug in the location generator.

[gholt (gholt)]

Looks good. You'll have to manually merge trunk into this though before final merge approval, as there's a small conflict at the moment.

review: Approve

=== modified file 'doc/source/deployment_guide.rst'
--- doc/source/deployment_guide.rst	2010-12-02 01:08:49 +0000
+++ doc/source/deployment_guide.rst	2011-01-11 17:02:10 +0000
@@ -229,7 +229,12 @@
 log_name            object-auditor  Label used when logging
 log_facility        LOG_LOCAL0      Syslog log facility
 log_level           INFO            Logging level
-interval            1800            Minimum time for a pass to take
+files_per_second    20              Maximum files audited per second. Should
+                                    be tuned according to individual system
+                                    specs. 0 is unlimited.
+bytes_per_second    10000000        Maximum bytes audited per second. Should
+                                    be tuned according to individual system
+                                    specs. 0 is unlimited.
 ==================  ==============  ==========================================
 
 ------------------------------

=== modified file 'etc/object-server.conf-sample'
--- etc/object-server.conf-sample	2010-10-19 15:02:36 +0000
+++ etc/object-server.conf-sample	2011-01-11 17:02:10 +0000
@@ -55,5 +55,5 @@
 
 [object-auditor]
 # log_name = object-auditor
-# Will audit, at most, 1 object per device per interval
-# interval = 1800
+# files_per_second = 20
+# bytes_per_second = 10000000

=== modified file 'swift/common/utils.py'
--- swift/common/utils.py	2011-01-10 20:37:49 +0000
+++ swift/common/utils.py	2011-01-11 17:02:10 +0000
@@ -779,19 +779,22 @@
                     on devices
     :param logger: a logger object
     '''
-    for device in os.listdir(devices):
-        if mount_check and not\
+    device_dir = os.listdir(devices)
+    # randomize devices in case of process restart before sweep completed
+    shuffle(device_dir)
+    for device in device_dir:
+        if mount_check and not \
                 os.path.ismount(os.path.join(devices, device)):
             if logger:
                 logger.debug(
                     _('Skipping %s as it is not mounted'), device)
             continue
-        datadir = os.path.join(devices, device, datadir)
-        if not os.path.exists(datadir):
+        datadir_path = os.path.join(devices, device, datadir)
+        if not os.path.exists(datadir_path):
             continue
-        partitions = os.listdir(datadir)
+        partitions = os.listdir(datadir_path)
         for partition in partitions:
-            part_path = os.path.join(datadir, partition)
+            part_path = os.path.join(datadir_path, partition)
             if not os.path.isdir(part_path):
                 continue
             suffixes = os.listdir(part_path)
@@ -808,3 +811,30 @@
                                         reverse=True):
                         path = os.path.join(hash_path, fname)
                         yield path, device, partition
+
+
+def ratelimit_sleep(running_time, max_rate, incr_by=1):
+    '''
+    Will eventlet.sleep() for the appropriate time so that the max_rate
+    is never exceeded.  If max_rate is 0, will not ratelimit.  The
+    maximum recommended rate should not exceed (1000 * incr_by) a second
+    as eventlet.sleep() does involve some overhead.  Returns running_time
+    that should be used for subsequent calls.
+
+    :param running_time: the running time of the next allowable request. Best
+                         to start at zero.
+    :param max_rate: The maximum rate per second allowed for the process.
+    :param incr_by: How much to increment the counter.  Useful if you want
+                    to ratelimit 1024 bytes/sec and have differing sizes
+                    of requests. Must be >= 0.
+    '''
+    if not max_rate or incr_by <= 0:
+        return running_time
+    clock_accuracy = 1000.0
+    now = time.time() * clock_accuracy
+    time_per_request = clock_accuracy * (float(incr_by) / max_rate)
+    if running_time < now:
+        running_time = now
+    elif running_time - now > time_per_request:
+        eventlet.sleep((running_time - now) / clock_accuracy)
+    return running_time + time_per_request

=== modified file 'swift/obj/auditor.py'
--- swift/obj/auditor.py	2011-01-04 23:34:43 +0000
+++ swift/obj/auditor.py	2011-01-11 17:02:10 +0000
@@ -20,7 +20,8 @@
 
 from swift.obj import server as object_server
 from swift.obj.replicator import invalidate_hash
-from swift.common.utils import get_logger, renamer, audit_location_generator
+from swift.common.utils import get_logger, renamer, audit_location_generator, \
+    ratelimit_sleep
 from swift.common.exceptions import AuditException
 from swift.common.daemon import Daemon
 
@@ -34,39 +35,30 @@
         self.devices = conf.get('devices', '/srv/node')
         self.mount_check = conf.get('mount_check', 'true').lower() in \
                               ('true', 't', '1', 'on', 'yes', 'y')
-        self.interval = int(conf.get('interval', 1800))
+        self.max_files_per_second = float(conf.get('files_per_second', 20))
+        self.max_bytes_per_second = float(conf.get('bytes_per_second',
+                                                   10000000))
+        self.files_running_time = 0
+        self.bytes_running_time = 0
+        self.bytes_processed = 0
+        self.total_bytes_processed = 0
+        self.total_files_processed = 0
         self.passes = 0
         self.quarantines = 0
         self.errors = 0
+        self.log_time = 3600  # once an hour
 
-    def run_forever(self):    # pragma: no cover
+    def run_forever(self):
         """Run the object audit until stopped."""
-        reported = time.time()
-        time.sleep(random() * self.interval)
         while True:
-            begin = time.time()
-            all_locs = audit_location_generator(self.devices,
-                                                object_server.DATADIR,
-                                                mount_check=self.mount_check,
-                                                logger=self.logger)
-            for path, device, partition in all_locs:
-                self.object_audit(path, device, partition)
-                if time.time() - reported >= 3600:  # once an hour
-                    self.logger.info(_('Since %(time)s: Locally: %(pass)d '
-                        'passed audit, %(quar)d quarantined, %(error)d errors'),
-                        {'time': time.ctime(reported), 'pass': self.passes,
-                         'quar': self.quarantines, 'error': self.errors})
-                    reported = time.time()
-                    self.passes = 0
-                    self.quarantines = 0
-                    self.errors = 0
-            elapsed = time.time() - begin
-            if elapsed < self.interval:
-                time.sleep(self.interval - elapsed)
+            self.run_once('forever')
+            self.total_bytes_processed = 0
+            self.total_files_processed = 0
+            time.sleep(30)
 
-    def run_once(self):
+    def run_once(self, mode='once'):
         """Run the object audit once."""
-        self.logger.info(_('Begin object audit "once" mode'))
+        self.logger.info(_('Begin object audit "%s" mode' % mode))
         begin = reported = time.time()
         all_locs = audit_location_generator(self.devices,
                                             object_server.DATADIR,
@@ -74,18 +66,35 @@
                                             logger=self.logger)
         for path, device, partition in all_locs:
             self.object_audit(path, device, partition)
-            if time.time() - reported >= 3600:  # once an hour
-                self.logger.info(_('Since %(time)s: Locally: %(pass)d '
-                    'passed audit, %(quar)d quarantined, %(error)d errors'),
-                    {'time': time.ctime(reported), 'pass': self.passes,
-                     'quar': self.quarantines, 'error': self.errors})
+            self.files_running_time = ratelimit_sleep(
+                self.files_running_time, self.max_files_per_second)
+            self.total_files_processed += 1
+            if time.time() - reported >= self.log_time:
+                self.logger.info(_(
+                    'Since %(start_time)s: Locally: %(passes)d passed audit, '
+                    '%(quars)d quarantined, %(errors)d errors '
+                    'files/sec: %(frate).2f , bytes/sec: %(brate).2f') % {
+                            'start_time': time.ctime(reported),
+                            'passes': self.passes,
+                            'quars': self.quarantines,
+                            'errors': self.errors,
+                            'frate': self.passes / (time.time() - reported),
+                            'brate': self.bytes_processed /
+                                     (time.time() - reported)})
                 reported = time.time()
                 self.passes = 0
                 self.quarantines = 0
                 self.errors = 0
+                self.bytes_processed = 0
         elapsed = time.time() - begin
-        self.logger.info(
-            _('Object audit "once" mode completed: %.02fs'), elapsed)
+        self.logger.info(_(
+                'Object audit "%(mode)s" mode completed: %(elapsed).02fs. '
+                'Total files/sec: %(frate).2f , '
+                'Total bytes/sec: %(brate).2f ') % {
+                    'mode': mode,
+                    'elapsed': elapsed,
+                    'frate': self.total_files_processed / elapsed,
+                    'brate': self.total_bytes_processed / elapsed})
 
     def object_audit(self, path, device, partition):
         """
@@ -102,7 +111,7 @@
                 name = object_server.read_metadata(path)['name']
             except Exception, exc:
                 raise AuditException('Error when reading metadata: %s' % exc)
-            _, account, container, obj = name.split('/', 3)
+            _junk, account, container, obj = name.split('/', 3)
             df = object_server.DiskFile(self.devices, device,
                                         partition, account,
                                         container, obj,
@@ -117,15 +126,20 @@
                                          os.path.getsize(df.data_file)))
             etag = md5()
             for chunk in df:
+                self.bytes_running_time = ratelimit_sleep(
+                    self.bytes_running_time, self.max_bytes_per_second,
+                    incr_by=len(chunk))
                 etag.update(chunk)
+                self.bytes_processed += len(chunk)
+                self.total_bytes_processed += len(chunk)
             etag = etag.hexdigest()
             if etag != df.metadata['ETag']:
                 raise AuditException("ETag of %s does not match file's md5 of "
                     "%s" % (df.metadata['ETag'], etag))
         except AuditException, err:
             self.quarantines += 1
-            self.logger.error(_('ERROR Object %(obj)s failed audit and will be '
-                'quarantined: %(err)s'), {'obj': path, 'err': err})
+            self.logger.error(_('ERROR Object %(obj)s failed audit and will '
+                'be quarantined: %(err)s'), {'obj': path, 'err': err})
             invalidate_hash(os.path.dirname(path))
             renamer_path = os.path.dirname(path)
             renamer(renamer_path, os.path.join(self.devices, device,

=== modified file 'swift/obj/server.py'
--- swift/obj/server.py	2011-01-07 21:17:29 +0000
+++ swift/obj/server.py	2011-01-11 17:02:10 +0000
@@ -72,6 +72,21 @@
     return pickle.loads(metadata)
 
 
+def write_metadata(fd, metadata):
+    """
+    Helper function to write pickled metadata for an object file.
+
+    :param fd: file descriptor to write the metadata
+    :param metadata: metadata to write
+    """
+    metastr = pickle.dumps(metadata, PICKLE_PROTOCOL)
+    key = 0
+    while metastr:
+        setxattr(fd, '%s%s' % (METADATA_KEY, key or ''), metastr[:254])
+        metastr = metastr[254:]
+        key += 1
+
+
 class DiskFile(object):
     """
     Manage object files on disk.
@@ -97,6 +112,7 @@
         self.metadata = {}
         self.meta_file = None
         self.data_file = None
+        self.fp = None
         if not os.path.exists(self.datadir):
             return
         files = sorted(os.listdir(self.datadir), reverse=True)
@@ -203,17 +219,12 @@
 
         :params fd: file descriptor of the temp file
         :param tmppath: path to the temporary file being used
-        :param metadata: dictionary of metada to be written
+        :param metadata: dictionary of metadata to be written
         :param extention: extension to be used when making the file
         """
         metadata['name'] = self.name
         timestamp = normalize_timestamp(metadata['X-Timestamp'])
-        metastr = pickle.dumps(metadata, PICKLE_PROTOCOL)
-        key = 0
-        while metastr:
-            setxattr(fd, '%s%s' % (METADATA_KEY, key or ''), metastr[:254])
-            metastr = metastr[254:]
-            key += 1
+        write_metadata(fd, metadata)
         if 'Content-Length' in metadata:
             drop_buffer_cache(fd, 0, int(metadata['Content-Length']))
         os.fsync(fd)

=== modified file 'test/unit/common/test_utils.py'
--- test/unit/common/test_utils.py	2011-01-07 21:17:29 +0000
+++ test/unit/common/test_utils.py	2011-01-11 17:02:10 +0000
@@ -21,6 +21,7 @@
 import os
 import socket
 import sys
+import time
 import unittest
 from getpass import getuser
 from shutil import rmtree
@@ -34,6 +35,7 @@
 
 
 class MockOs():
+
     def __init__(self, pass_funcs=[], called_funcs=[], raise_funcs=[]):
         self.closed_fds = []
         for func in pass_funcs:
@@ -183,12 +185,12 @@
         print 'test2'
         self.assertEquals(sio.getvalue(), 'STDOUT: test2\n')
         sys.stderr = lfo
-        print >>sys.stderr, 'test4'
+        print >> sys.stderr, 'test4'
         self.assertEquals(sio.getvalue(), 'STDOUT: test2\nSTDOUT: test4\n')
         sys.stdout = orig_stdout
         print 'test5'
         self.assertEquals(sio.getvalue(), 'STDOUT: test2\nSTDOUT: test4\n')
-        print >>sys.stderr, 'test6'
+        print >> sys.stderr, 'test6'
         self.assertEquals(sio.getvalue(), 'STDOUT: test2\nSTDOUT: test4\n'
             'STDOUT: test6\n')
         sys.stderr = orig_stderr
@@ -316,7 +318,7 @@
 
     def test_hash_path(self):
         # Yes, these tests are deliberately very fragile. We want to make sure
-        # that if someones changes the results hash_path produces, they know it.
+        # that if someones changes the results hash_path produces, they know it
         self.assertEquals(utils.hash_path('a'),
                           '1c84525acb02107ea475dcd3d09c2c58')
         self.assertEquals(utils.hash_path('a', 'c'),
@@ -355,10 +357,12 @@
         result = utils.readconf('/tmp/test', 'section2').get('log_name')
         expected = 'yarr'
         self.assertEquals(result, expected)
-        result = utils.readconf('/tmp/test', 'section1', log_name='foo').get('log_name')
+        result = utils.readconf('/tmp/test', 'section1',
+                                log_name='foo').get('log_name')
         expected = 'foo'
         self.assertEquals(result, expected)
-        result = utils.readconf('/tmp/test', 'section1', defaults={'bar': 'baz'})
+        result = utils.readconf('/tmp/test', 'section1',
+                                defaults={'bar': 'baz'})
         expected = {'log_name': 'section1', 'foo': 'bar', 'bar': 'baz'}
         self.assertEquals(result, expected)
         os.unlink('/tmp/test')
@@ -438,5 +442,41 @@
         self.assertNotEquals(utils.get_logger.console, old_handler)
         logger.logger.removeHandler(utils.get_logger.console)
 
+    def test_ratelimit_sleep(self):
+        running_time = 0
+        start = time.time()
+        for i in range(100):
+            running_time = utils.ratelimit_sleep(running_time, 0)
+        self.assertTrue(abs((time.time() - start) * 100) < 1)
+
+        running_time = 0
+        start = time.time()
+        for i in range(50):
+            running_time = utils.ratelimit_sleep(running_time, 200)
+        # make sure its accurate to 10th of a second
+        self.assertTrue(abs(25 - (time.time() - start) * 100) < 10)
+
+    def test_ratelimit_sleep_with_sleep(self):
+        running_time = 0
+        start = time.time()
+        for i in range(25):
+            running_time = utils.ratelimit_sleep(running_time, 50)
+            time.sleep(1.0 / 75)
+        # make sure its accurate to 10th of a second
+        self.assertTrue(abs(50 - (time.time() - start) * 100) < 10)
+
+    def test_ratelimit_sleep_with_incr(self):
+        running_time = 0
+        start = time.time()
+        vals = [5, 17, 0, 3, 11, 30,
+                40, 4, 13, 2, -1] * 2  # adds up to 250 (with no -1)
+        total = 0
+        for i in vals:
+            running_time = utils.ratelimit_sleep(running_time,
+                                                 500, incr_by=i)
+            total += i
+        self.assertTrue(abs(50 - (time.time() - start) * 100) < 10)
+
+
 if __name__ == '__main__':
     unittest.main()

=== modified file 'test/unit/obj/test_auditor.py'
--- test/unit/obj/test_auditor.py	2011-01-04 23:34:43 +0000
+++ test/unit/obj/test_auditor.py	2011-01-11 17:02:10 +0000
@@ -14,14 +14,234 @@
 # limitations under the License.
 
 # TODO: Tests
-
 import unittest
+import tempfile
+import os
+import time
+from shutil import rmtree
+from hashlib import md5
 from swift.obj import auditor
+from swift.obj.server import DiskFile, write_metadata
+from swift.common.utils import hash_path, mkdirs, normalize_timestamp, renamer
+from swift.obj.replicator import invalidate_hash
+from swift.common.exceptions import AuditException
+
 
 class TestAuditor(unittest.TestCase):
 
-    def test_placeholder(self):
-        pass
+    def setUp(self):
+        self.path_to_test_xfs = os.environ.get('PATH_TO_TEST_XFS')
+        if not self.path_to_test_xfs or \
+                not os.path.exists(self.path_to_test_xfs):
+            print >> sys.stderr, 'WARNING: PATH_TO_TEST_XFS not set or not ' \
+                'pointing to a valid directory.\n' \
+                'Please set PATH_TO_TEST_XFS to a directory on an XFS file ' \
+                'system for testing.'
+            self.testdir = '/tmp/SWIFTUNITTEST'
+        else:
+            self.testdir = os.path.join(self.path_to_test_xfs,
+                            'tmp_test_object_auditor')
+
+        self.devices = os.path.join(self.testdir, 'node')
+        rmtree(self.testdir, ignore_errors=1)
+        os.mkdir(self.testdir)
+        os.mkdir(self.devices)
+        os.mkdir(os.path.join(self.devices, 'sda'))
+        self.objects = os.path.join(self.devices, 'sda', 'objects')
+
+        os.mkdir(os.path.join(self.devices, 'sdb'))
+        self.objects_2 = os.path.join(self.devices, 'sdb', 'objects')
+
+        os.mkdir(self.objects)
+        self.parts = {}
+        for part in ['0', '1', '2', '3']:
+            self.parts[part] = os.path.join(self.objects, part)
+            os.mkdir(os.path.join(self.objects, part))
+
+        self.conf = dict(
+            devices=self.devices,
+            mount_check='false')
+
+    def tearDown(self):
+        rmtree(self.testdir, ignore_errors=1)
+
+    def test_object_audit_extra_data(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        cur_part = '0'
+        disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
+        data = '0' * 1024
+        etag = md5()
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            timestamp = str(normalize_timestamp(time.time()))
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': timestamp,
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            pre_quarantines = self.auditor.quarantines
+
+            self.auditor.object_audit(
+                os.path.join(disk_file.datadir, timestamp + '.data'),
+                'sda', cur_part)
+            self.assertEquals(self.auditor.quarantines, pre_quarantines)
+
+            os.write(fd, 'extra_data')
+            self.auditor.object_audit(
+                os.path.join(disk_file.datadir, timestamp + '.data'),
+                'sda', cur_part)
+            self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
+
+    def test_object_audit_diff_data(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        cur_part = '0'
+        disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
+        data = '0' * 1024
+        etag = md5()
+        timestamp = str(normalize_timestamp(time.time()))
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': timestamp,
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            pre_quarantines = self.auditor.quarantines
+
+            self.auditor.object_audit(
+                os.path.join(disk_file.datadir, timestamp + '.data'),
+                'sda', cur_part)
+            self.assertEquals(self.auditor.quarantines, pre_quarantines)
+            etag = md5()
+            etag.update('1' + '0' * 1023)
+            etag = etag.hexdigest()
+            metadata['ETag'] = etag
+            write_metadata(fd, metadata)
+
+            self.auditor.object_audit(
+                os.path.join(disk_file.datadir, timestamp + '.data'),
+                'sda', cur_part)
+            self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
+
+    def test_object_audit_no_meta(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        cur_part = '0'
+        disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
+        data = '0' * 1024
+        etag = md5()
+        pre_quarantines = self.auditor.quarantines
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            timestamp = str(normalize_timestamp(time.time()))
+            os.fsync(fd)
+            invalidate_hash(os.path.dirname(disk_file.datadir))
+            renamer(tmppath, os.path.join(disk_file.datadir,
+                                          timestamp + '.data'))
+            self.auditor.object_audit(
+                os.path.join(disk_file.datadir, timestamp + '.data'),
+                'sda', cur_part)
+            self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
+
+    def test_object_audit_bad_args(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        pre_errors = self.auditor.errors
+        self.auditor.object_audit(5, 'sda', '0')
+        self.assertEquals(self.auditor.errors, pre_errors + 1)
+        pre_errors = self.auditor.errors
+        self.auditor.object_audit('badpath', 'sda', '0')
+        self.assertEquals(self.auditor.errors, pre_errors)  # just returns
+
+    def test_object_run_once_pass(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor.log_time = 0
+        cur_part = '0'
+        timestamp = str(normalize_timestamp(time.time()))
+        pre_quarantines = self.auditor.quarantines
+        disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
+        data = '0' * 1024
+        etag = md5()
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': timestamp,
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            disk_file.close()
+        self.auditor.run_once()
+        self.assertEquals(self.auditor.quarantines, pre_quarantines)
+
+    def test_object_run_once_no_sda(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        cur_part = '0'
+        timestamp = str(normalize_timestamp(time.time()))
+        pre_quarantines = self.auditor.quarantines
+        disk_file = DiskFile(self.devices, 'sdb', cur_part, 'a', 'c', 'o')
+        data = '0' * 1024
+        etag = md5()
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': timestamp,
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            disk_file.close()
+            os.write(fd, 'extra_data')
+        self.auditor.run_once()
+        self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
+
+    def test_object_run_once_multi_devices(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        cur_part = '0'
+        timestamp = str(normalize_timestamp(time.time()))
+        pre_quarantines = self.auditor.quarantines
+        disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
+        data = '0' * 10
+        etag = md5()
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': timestamp,
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            disk_file.close()
+        self.auditor.run_once()
+        disk_file = DiskFile(self.devices, 'sdb', cur_part, 'a', 'c', 'ob')
+        data = '1' * 10
+        etag = md5()
+        with disk_file.mkstemp() as (fd, tmppath):
+            os.write(fd, data)
+            etag.update(data)
+            etag = etag.hexdigest()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': timestamp,
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            disk_file.close()
+            os.write(fd, 'extra_data')
+        self.auditor.run_once()
+        self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
 
 
 if __name__ == '__main__':