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

[David Goetz]

Ability to fasttrack auditing of zero byte files.

[gholt (gholt)]

doc/source/admin_guide.rst
Typo/confusing phrase: "to a 1000 from the config file, ..."

Seems there are new conf variables but not documented and not in etc/-sample?

What's the difference between fasttrack_zero_byte_files and zero_byte_file_worker and zero_byte_fps and how do the interact? I think I can mostly make it out.

I guess my thought on this initially was that there would be an extra process forked that would simple scan for zero byte files exclusively and other processes would run as they always have. Then, the command line override would only do the zbf fork and not the normal.

[lp:~david-goetz/swift/zero_byte_audit_dos]

224. By David Goetz on 2011-02-25

peer review edits

225. By David Goetz on 2011-02-28

get rid of logger

226. By David Goetz on 2011-03-03

merge to trunk

[gholt (gholt)]

review: Approve

=== modified file 'bin/swift-log-uploader'
--- bin/swift-log-uploader	2011-02-10 17:57:51 +0000
+++ bin/swift-log-uploader	2011-03-03 21:53:01 +0000
@@ -15,16 +15,17 @@
 # limitations under the License.
 
 import sys
-
+from optparse import OptionParser
 from swift.stats.log_uploader import LogUploader
 from swift.common.utils import parse_options
 from swift.common import utils
 
 if __name__ == '__main__':
-    conf_file, options = parse_options(usage="Usage: %prog CONFIG_FILE PLUGIN")
+    parser = OptionParser("Usage: %prog CONFIG_FILE PLUGIN")
+    conf_file, options = parse_options(parser=parser)
     try:
         plugin = options['extra_args'][0]
-    except IndexError:
+    except (IndexError, KeyError):
         print "Error: missing plugin name"
         sys.exit(1)
 

=== modified file 'bin/swift-object-auditor'
--- bin/swift-object-auditor	2011-01-04 23:34:43 +0000
+++ bin/swift-object-auditor	2011-03-03 21:53:01 +0000
@@ -17,7 +17,11 @@
 from swift.obj.auditor import ObjectAuditor
 from swift.common.utils import parse_options
 from swift.common.daemon import run_daemon
+from optparse import OptionParser
 
 if __name__ == '__main__':
-    conf_file, options = parse_options(once=True)
+    parser = OptionParser("%prog CONFIG [options]")
+    parser.add_option('-z', '--zero_byte_fps',
+                      help='Audit only zero byte files at specified files/sec')
+    conf_file, options = parse_options(parser=parser, once=True)
     run_daemon(ObjectAuditor, conf_file, **options)

=== modified file 'doc/source/admin_guide.rst'
--- doc/source/admin_guide.rst	2010-12-10 01:57:26 +0000
+++ doc/source/admin_guide.rst	2011-03-03 21:53:01 +0000
@@ -288,3 +288,15 @@
 completely stopping the old service.  There is also a special case of 
 `swift-init all <command>`, which will run the command for all swift services.
 
+--------------
+Object Auditor
+--------------
+
+On system failures, the XFS file system can sometimes truncate files it's
+trying to write and produce zero byte files. The object-auditor will catch
+these problems but in the case of a system crash it would be advisable to run
+an extra, less rate limited sweep to check for these specific files. You can
+run this command as follows:
+`swift-object-auditor /path/to/object-server/config/file.conf once -z 1000`
+"-z" means to only check for zero-byte files at 1000 files per second.
+

=== modified file 'etc/object-server.conf-sample'
--- etc/object-server.conf-sample	2011-01-27 21:02:53 +0000
+++ etc/object-server.conf-sample	2011-03-03 21:53:01 +0000
@@ -72,3 +72,5 @@
 # files_per_second = 20
 # bytes_per_second = 10000000
 # log_time = 3600
+# zero_byte_files_per_second = 50
+

=== modified file 'swift/account/auditor.py'
--- swift/account/auditor.py	2011-02-02 21:39:08 +0000
+++ swift/account/auditor.py	2011-03-03 21:53:01 +0000
@@ -36,7 +36,7 @@
         self.account_passes = 0
         self.account_failures = 0
 
-    def run_forever(self):  # pragma: no cover
+    def run_forever(self, *args, **kwargs):
         """Run the account audit until stopped."""
         reported = time.time()
         time.sleep(random() * self.interval)
@@ -61,7 +61,7 @@
             if elapsed < self.interval:
                 time.sleep(self.interval - elapsed)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """Run the account audit once."""
         self.logger.info('Begin account audit "once" mode')
         begin = reported = time.time()

=== modified file 'swift/account/reaper.py'
--- swift/account/reaper.py	2011-02-02 21:39:08 +0000
+++ swift/account/reaper.py	2011-03-03 21:53:01 +0000
@@ -97,7 +97,7 @@
             self.object_ring = Ring(self.object_ring_path)
         return self.object_ring
 
-    def run_forever(self):
+    def run_forever(self, *args, **kwargs):
         """
         Main entry point when running the reaper in its normal daemon mode.
         This repeatedly calls :func:`reap_once` no quicker than the
@@ -112,7 +112,7 @@
             if elapsed < self.interval:
                 sleep(self.interval - elapsed)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """
         Main entry point when running the reaper in 'once' mode, where it will
         do a single pass over all accounts on the server. This is called

=== modified file 'swift/common/daemon.py'
--- swift/common/daemon.py	2011-02-14 23:02:08 +0000
+++ swift/common/daemon.py	2011-03-03 21:53:01 +0000
@@ -28,11 +28,11 @@
         self.conf = conf
         self.logger = utils.get_logger(conf, log_route='daemon')
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """Override this to run the script once"""
         raise NotImplementedError('run_once not implemented')
 
-    def run_forever(self):
+    def run_forever(self, *args, **kwargs):
         """Override this to run forever"""
         raise NotImplementedError('run_forever not implemented')
 
@@ -48,15 +48,13 @@
             sys.exit()
 
         signal.signal(signal.SIGTERM, kill_children)
-
         if once:
-            self.run_once()
+            self.run_once(**kwargs)
         else:
-            self.run_forever()
-
-
-def run_daemon(klass, conf_file, section_name='',
-               once=False, **kwargs):
+            self.run_forever(**kwargs)
+
+
+def run_daemon(klass, conf_file, section_name='', once=False, **kwargs):
     """
     Loads settings from conf, then instantiates daemon "klass" and runs the
     daemon with the specified once kwarg.  The section_name will be derived

=== modified file 'swift/common/db_replicator.py'
--- swift/common/db_replicator.py	2011-02-17 17:30:41 +0000
+++ swift/common/db_replicator.py	2011-03-03 21:53:01 +0000
@@ -396,7 +396,7 @@
                 except StopIteration:
                     its.remove(it)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """Run a replication pass once."""
         self._zero_stats()
         dirs = []
@@ -425,7 +425,7 @@
         self.logger.info(_('Replication run OVER'))
         self._report_stats()
 
-    def run_forever(self):
+    def run_forever(self, *args, **kwargs):
         """
         Replicate dbs under the given root in an infinite loop.
         """

=== modified file 'swift/common/utils.py'
--- swift/common/utils.py	2011-02-24 07:43:05 +0000
+++ swift/common/utils.py	2011-03-03 21:53:01 +0000
@@ -370,6 +370,7 @@
     Custom logging.Formatter will append txn_id to a log message if the record
     has one and the message does not.
     """
+
     def format(self, record):
         msg = logging.Formatter.format(self, record)
         if (record.txn_id and record.levelno != logging.INFO and
@@ -492,11 +493,11 @@
         sys.stderr = LoggerFileObject(logger)
 
 
-def parse_options(usage="%prog CONFIG [options]", once=False, test_args=None):
+def parse_options(parser=None, once=False, test_args=None):
     """
     Parse standard swift server/daemon options with optparse.OptionParser.
 
-    :param usage: String describing usage
+    :param parser: OptionParser to use. If not sent one will be created.
     :param once: Boolean indicating the "once" option is available
     :param test_args: Override sys.argv; used in testing
 
@@ -505,7 +506,8 @@
 
     :raises SystemExit: First arg (CONFIG) is required, file must exist
     """
-    parser = OptionParser(usage)
+    if not parser:
+        parser = OptionParser(usage="%prog CONFIG [options]")
     parser.add_option("-v", "--verbose", default=False, action="store_true",
                       help="log to console")
     if once:
@@ -534,7 +536,8 @@
             extra_args.append(arg)
 
     options = vars(options)
-    options['extra_args'] = extra_args
+    if extra_args:
+        options['extra_args'] = extra_args
     return config, options
 
 

=== modified file 'swift/container/auditor.py'
--- swift/container/auditor.py	2011-02-02 21:39:08 +0000
+++ swift/container/auditor.py	2011-03-03 21:53:01 +0000
@@ -37,7 +37,7 @@
         self.container_passes = 0
         self.container_failures = 0
 
-    def run_forever(self):  # pragma: no cover
+    def run_forever(self, *args, **kwargs):
         """Run the container audit until stopped."""
         reported = time.time()
         time.sleep(random() * self.interval)
@@ -63,7 +63,7 @@
             if elapsed < self.interval:
                 time.sleep(self.interval - elapsed)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """Run the container audit once."""
         self.logger.info(_('Begin container audit "once" mode'))
         begin = reported = time.time()

=== modified file 'swift/container/updater.py'
--- swift/container/updater.py	2011-02-02 21:39:08 +0000
+++ swift/container/updater.py	2011-03-03 21:53:01 +0000
@@ -98,7 +98,7 @@
         finally:
             os.unlink(filename)
 
-    def run_forever(self):   # pragma: no cover
+    def run_forever(self, *args, **kwargs):
         """
         Run the updator continuously.
         """
@@ -156,7 +156,7 @@
             if elapsed < self.interval:
                 time.sleep(self.interval - elapsed)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """
         Run the updater once.
         """

=== modified file 'swift/obj/auditor.py'
--- swift/obj/auditor.py	2011-02-02 21:39:08 +0000
+++ swift/obj/auditor.py	2011-03-03 21:53:01 +0000
@@ -21,23 +21,30 @@
 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, \
-    ratelimit_sleep
+    ratelimit_sleep, TRUE_VALUES
 from swift.common.exceptions import AuditException
 from swift.common.daemon import Daemon
 
-
-class ObjectAuditor(Daemon):
-    """Audit objects."""
-
-    def __init__(self, conf):
+SLEEP_BETWEEN_AUDITS = 30
+
+
+class AuditorWorker(object):
+    """Walk through file system to audit object"""
+
+    def __init__(self, conf, zero_byte_only_at_fps=0):
         self.conf = conf
         self.logger = get_logger(conf, log_route='object-auditor')
         self.devices = conf.get('devices', '/srv/node')
         self.mount_check = conf.get('mount_check', 'true').lower() in \
-                              ('true', 't', '1', 'on', 'yes', 'y')
+            TRUE_VALUES
         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.auditor_type = 'ALL'
+        self.zero_byte_only_at_fps = zero_byte_only_at_fps
+        if self.zero_byte_only_at_fps:
+            self.max_files_per_second = float(self.zero_byte_only_at_fps)
+            self.auditor_type = 'ZBF'
         self.log_time = int(conf.get('log_time', 3600))
         self.files_running_time = 0
         self.bytes_running_time = 0
@@ -48,18 +55,13 @@
         self.quarantines = 0
         self.errors = 0
 
-    def run_forever(self):
-        """Run the object audit until stopped."""
-        while True:
-            self.run_once('forever')
-            self.total_bytes_processed = 0
-            self.total_files_processed = 0
-            time.sleep(30)
-
-    def run_once(self, mode='once'):
-        """Run the object audit once."""
-        self.logger.info(_('Begin object audit "%s" mode' % mode))
+    def audit_all_objects(self, mode='once'):
+        self.logger.info(_('Begin object audit "%s" mode (%s)' %
+                           (mode, self.auditor_type)))
         begin = reported = time.time()
+        self.total_bytes_processed = 0
+        self.total_files_processed = 0
+        files_running_time = 0
         all_locs = audit_location_generator(self.devices,
                                             object_server.DATADIR,
                                             mount_check=self.mount_check,
@@ -71,9 +73,11 @@
             self.total_files_processed += 1
             if time.time() - reported >= self.log_time:
                 self.logger.info(_(
-                    'Since %(start_time)s: Locally: %(passes)d passed audit, '
+                    'Object audit (%(type)s). '
+                    'Since %(start_time)s: Locally: %(passes)d passed, '
                     '%(quars)d quarantined, %(errors)d errors '
                     'files/sec: %(frate).2f , bytes/sec: %(brate).2f') % {
+                            'type': self.auditor_type,
                             'start_time': time.ctime(reported),
                             'passes': self.passes,
                             'quars': self.quarantines,
@@ -88,9 +92,11 @@
                 self.bytes_processed = 0
         elapsed = time.time() - begin
         self.logger.info(_(
-                'Object audit "%(mode)s" mode completed: %(elapsed).02fs. '
+                'Object audit (%(type)s) "%(mode)s" mode '
+                'completed: %(elapsed).02fs. '
                 'Total files/sec: %(frate).2f , '
                 'Total bytes/sec: %(brate).2f ') % {
+                    'type': self.auditor_type,
                     'mode': mode,
                     'elapsed': elapsed,
                     'frate': self.total_files_processed / elapsed,
@@ -98,7 +104,7 @@
 
     def object_audit(self, path, device, partition):
         """
-        Audits the given object path
+        Audits the given object path.
 
         :param path: a path to an object
         :param device: the device the path is on
@@ -119,11 +125,13 @@
             if df.data_file is None:
                 # file is deleted, we found the tombstone
                 return
-            if os.path.getsize(df.data_file) != \
-                    int(df.metadata['Content-Length']):
+            obj_size = os.path.getsize(df.data_file)
+            if obj_size != int(df.metadata['Content-Length']):
                 raise AuditException('Content-Length of %s does not match '
                     'file size of %s' % (int(df.metadata['Content-Length']),
                                          os.path.getsize(df.data_file)))
+            if self.zero_byte_only_at_fps and obj_size:
+                return
             etag = md5()
             for chunk in df:
                 self.bytes_running_time = ratelimit_sleep(
@@ -150,3 +158,39 @@
             self.logger.exception(_('ERROR Trying to audit %s'), path)
             return
         self.passes += 1
+
+
+class ObjectAuditor(Daemon):
+    """Audit objects."""
+
+    def __init__(self, conf, **options):
+        self.conf = conf
+        self.conf_zero_byte_fps = int(conf.get(
+                'zero_byte_files_per_second', 50))
+
+    def _sleep(self):
+        time.sleep(SLEEP_BETWEEN_AUDITS)
+
+    def run_forever(self, *args, **kwargs):
+        """Run the object audit until stopped."""
+        # zero byte only command line option
+        zbo_fps = kwargs.get('zero_byte_fps', 0)
+        if zbo_fps:
+            # only start parent
+            parent = True
+        else:
+            parent = os.fork()  # child gets parent = 0
+        kwargs = {'mode': 'forever'}
+        if parent:
+            kwargs['zero_byte_fps'] = zbo_fps or self.conf_zero_byte_fps
+        while True:
+            self.run_once(**kwargs)
+            self._sleep()
+
+    def run_once(self, *args, **kwargs):
+        """Run the object audit once."""
+        mode = kwargs.get('mode', 'once')
+        zero_byte_only_at_fps = kwargs.get('zero_byte_fps', 0)
+        worker = AuditorWorker(self.conf,
+                               zero_byte_only_at_fps=zero_byte_only_at_fps)
+        worker.audit_all_objects(mode=mode)

=== modified file 'swift/obj/replicator.py'
--- swift/obj/replicator.py	2011-02-02 21:39:08 +0000
+++ swift/obj/replicator.py	2011-03-03 21:53:01 +0000
@@ -547,7 +547,7 @@
             lockup_detector.kill()
             self.stats_line()
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         start = time.time()
         self.logger.info(_("Running object replicator in script mode."))
         self.replicate()
@@ -555,7 +555,7 @@
         self.logger.info(
             _("Object replication complete. (%.02f minutes)"), total)
 
-    def run_forever(self):
+    def run_forever(self, *args, **kwargs):
         self.logger.info("Starting object replicator in daemon mode.")
         # Run the replicator continually
         while True:

=== modified file 'swift/obj/updater.py'
--- swift/obj/updater.py	2011-02-02 21:39:08 +0000
+++ swift/obj/updater.py	2011-03-03 21:53:01 +0000
@@ -58,7 +58,7 @@
             self.container_ring = Ring(self.container_ring_path)
         return self.container_ring
 
-    def run_forever(self):   # pragma: no cover
+    def run_forever(self, *args, **kwargs):
         """Run the updater continuously."""
         time.sleep(random() * self.interval)
         while True:
@@ -100,7 +100,7 @@
             if elapsed < self.interval:
                 time.sleep(self.interval - elapsed)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         """Run the updater once"""
         self.logger.info(_('Begin object update single threaded sweep'))
         begin = time.time()

=== modified file 'swift/stats/account_stats.py'
--- swift/stats/account_stats.py	2011-02-02 21:39:08 +0000
+++ swift/stats/account_stats.py	2011-03-03 21:53:01 +0000
@@ -51,7 +51,7 @@
         self.logger = \
             get_logger(stats_conf, log_route='account-stats')
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         self.logger.info(_("Gathering account stats"))
         start = time.time()
         self.find_and_process()

=== modified file 'swift/stats/log_processor.py'
--- swift/stats/log_processor.py	2011-03-01 18:27:19 +0000
+++ swift/stats/log_processor.py	2011-03-03 21:53:01 +0000
@@ -235,7 +235,7 @@
                                              'log_processing_data')
         self.worker_count = int(c.get('worker_count', '1'))
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         self.logger.info(_("Beginning log processing"))
         start = time.time()
         if self.lookback_hours == 0:

=== modified file 'swift/stats/log_uploader.py'
--- swift/stats/log_uploader.py	2011-02-17 20:42:14 +0000
+++ swift/stats/log_uploader.py	2011-03-03 21:53:01 +0000
@@ -68,7 +68,7 @@
         self.logger = utils.get_logger(uploader_conf, log_name,
                                        log_route=plugin_name)
 
-    def run_once(self):
+    def run_once(self, *args, **kwargs):
         self.logger.info(_("Uploading logs"))
         start = time.time()
         self.upload_all_logs()

=== modified file 'test/unit/common/test_utils.py'
--- test/unit/common/test_utils.py	2011-02-12 07:55:57 +0000
+++ test/unit/common/test_utils.py	2011-03-03 21:53:01 +0000
@@ -275,26 +275,16 @@
         stde = StringIO()
         utils.sys.stdout = stdo
         utils.sys.stderr = stde
-        err_msg = """Usage: test usage
-
-Error: missing config file argument
-"""
-        test_args = []
-        self.assertRaises(SystemExit, utils.parse_options, 'test usage', True,
-                          test_args)
-        self.assertEquals(stdo.getvalue(), err_msg)
+        self.assertRaises(SystemExit, utils.parse_options, once=True,
+                          test_args=[])
+        self.assert_('missing config file' in stdo.getvalue())
 
         # verify conf file must exist, context manager will delete temp file
         with NamedTemporaryFile() as f:
             conf_file = f.name
-        err_msg += """Usage: test usage
-
-Error: unable to locate %s
-""" % conf_file
-        test_args = [conf_file]
-        self.assertRaises(SystemExit, utils.parse_options, 'test usage', True,
-                          test_args)
-        self.assertEquals(stdo.getvalue(), err_msg)
+        self.assertRaises(SystemExit, utils.parse_options, once=True,
+                          test_args=[conf_file])
+        self.assert_('unable to locate' in stdo.getvalue())
 
         # reset stdio
         utils.sys.stdout = orig_stdout

=== modified file 'test/unit/obj/test_auditor.py'
--- test/unit/obj/test_auditor.py	2011-02-10 17:57:51 +0000
+++ test/unit/obj/test_auditor.py	2011-03-03 21:53:01 +0000
@@ -24,8 +24,9 @@
 from tempfile import mkdtemp
 from swift.obj import auditor
 from swift.obj import server as object_server
-from swift.obj.server import DiskFile, write_metadata
-from swift.common.utils import hash_path, mkdirs, normalize_timestamp, renamer
+from swift.obj.server import DiskFile, write_metadata, DATADIR
+from swift.common.utils import hash_path, mkdirs, normalize_timestamp, \
+    renamer, storage_directory
 from swift.obj.replicator import invalidate_hash
 from swift.common.exceptions import AuditException
 
@@ -60,7 +61,7 @@
         unit.xattr_data = {}
 
     def test_object_audit_extra_data(self):
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         cur_part = '0'
         disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
         data = '0' * 1024
@@ -90,7 +91,7 @@
             self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
 
     def test_object_audit_diff_data(self):
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         cur_part = '0'
         disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
         data = '0' * 1024
@@ -133,7 +134,7 @@
         fp.write('0' * 1024)
         fp.close()
         invalidate_hash(os.path.dirname(disk_file.datadir))
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         pre_quarantines = self.auditor.quarantines
         self.auditor.object_audit(
             os.path.join(disk_file.datadir, timestamp + '.data'),
@@ -141,7 +142,7 @@
         self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
 
     def test_object_audit_bad_args(self):
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         pre_errors = self.auditor.errors
         self.auditor.object_audit(5, 'sda', '0')
         self.assertEquals(self.auditor.errors, pre_errors + 1)
@@ -150,7 +151,7 @@
         self.assertEquals(self.auditor.errors, pre_errors)  # just returns
 
     def test_object_run_once_pass(self):
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         self.auditor.log_time = 0
         cur_part = '0'
         timestamp = str(normalize_timestamp(time.time()))
@@ -169,11 +170,11 @@
             }
             disk_file.put(fd, tmppath, metadata)
             disk_file.close()
-        self.auditor.run_once()
+        self.auditor.audit_all_objects()
         self.assertEquals(self.auditor.quarantines, pre_quarantines)
 
     def test_object_run_once_no_sda(self):
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         cur_part = '0'
         timestamp = str(normalize_timestamp(time.time()))
         pre_quarantines = self.auditor.quarantines
@@ -192,11 +193,11 @@
             disk_file.put(fd, tmppath, metadata)
             disk_file.close()
             os.write(fd, 'extra_data')
-        self.auditor.run_once()
+        self.auditor.audit_all_objects()
         self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
 
     def test_object_run_once_multi_devices(self):
-        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor = auditor.AuditorWorker(self.conf)
         cur_part = '0'
         timestamp = str(normalize_timestamp(time.time()))
         pre_quarantines = self.auditor.quarantines
@@ -214,7 +215,7 @@
             }
             disk_file.put(fd, tmppath, metadata)
             disk_file.close()
-        self.auditor.run_once()
+        self.auditor.audit_all_objects()
         disk_file = DiskFile(self.devices, 'sdb', cur_part, 'a', 'c', 'ob')
         data = '1' * 10
         etag = md5()
@@ -230,9 +231,150 @@
             disk_file.put(fd, tmppath, metadata)
             disk_file.close()
             os.write(fd, 'extra_data')
-        self.auditor.run_once()
+        self.auditor.audit_all_objects()
         self.assertEquals(self.auditor.quarantines, pre_quarantines + 1)
 
+    def test_object_run_fast_track_non_zero(self):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor.log_time = 0
+        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()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': str(normalize_timestamp(time.time())),
+                'Content-Length': str(os.fstat(fd).st_size),
+            }
+            disk_file.put(fd, tmppath, metadata)
+            etag = md5()
+            etag.update('1' + '0' * 1023)
+            etag = etag.hexdigest()
+            metadata['ETag'] = etag
+            write_metadata(fd, metadata)
+
+        quarantine_path = os.path.join(self.devices,
+                                       'sda', 'quarantined', 'objects')
+        self.auditor.run_once(zero_byte_fps=50)
+        self.assertFalse(os.path.isdir(quarantine_path))
+        self.auditor.run_once()
+        self.assertTrue(os.path.isdir(quarantine_path))
+
+    def setup_bad_zero_byte(self, with_ts=False):
+        self.auditor = auditor.ObjectAuditor(self.conf)
+        self.auditor.log_time = 0
+        cur_part = '0'
+        ts_file_path = ''
+        if with_ts:
+
+            name_hash = hash_path('a', 'c', 'o')
+            dir_path = os.path.join(self.devices, 'sda',
+                               storage_directory(DATADIR, cur_part, name_hash))
+            ts_file_path = os.path.join(dir_path, '99999.ts')
+            if not os.path.exists(dir_path):
+                mkdirs(dir_path)
+            fp = open(ts_file_path, 'w')
+            fp.close()
+
+
+        disk_file = DiskFile(self.devices, 'sda', cur_part, 'a', 'c', 'o')
+        etag = md5()
+        with disk_file.mkstemp() as (fd, tmppath):
+            etag = etag.hexdigest()
+            metadata = {
+                'ETag': etag,
+                'X-Timestamp': str(normalize_timestamp(time.time())),
+                'Content-Length': 10,
+            }
+            disk_file.put(fd, tmppath, metadata)
+            etag = md5()
+            etag = etag.hexdigest()
+            metadata['ETag'] = etag
+            write_metadata(fd, metadata)
+        if disk_file.data_file:
+            return disk_file.data_file
+        return ts_file_path
+
+    def test_object_run_fast_track_all(self):
+        self.setup_bad_zero_byte()
+        self.auditor.run_once()
+        quarantine_path = os.path.join(self.devices,
+                                       'sda', 'quarantined', 'objects')
+        self.assertTrue(os.path.isdir(quarantine_path))
+
+    def test_object_run_fast_track_zero(self):
+        self.setup_bad_zero_byte()
+        self.auditor.run_once(zero_byte_fps=50)
+        quarantine_path = os.path.join(self.devices,
+                                       'sda', 'quarantined', 'objects')
+        self.assertTrue(os.path.isdir(quarantine_path))
+
+    def test_with_tombstone(self):
+        ts_file_path = self.setup_bad_zero_byte(with_ts=True)
+        self.auditor.run_once()
+        quarantine_path = os.path.join(self.devices,
+                                       'sda', 'quarantined', 'objects')
+        self.assertTrue(ts_file_path.endswith('ts'))
+        self.assertTrue(os.path.exists(ts_file_path))
+
+    def test_sleeper(self):
+        auditor.SLEEP_BETWEEN_AUDITS = 0.01
+        my_auditor = auditor.ObjectAuditor(self.conf)
+        start = time.time()
+        my_auditor._sleep()
+        self.assertEquals(round(time.time() - start, 2), 0.01)
+
+    def test_run_forever(self):
+
+        class StopForever(Exception):
+            pass
+
+        class ObjectAuditorMock(object):
+            check_args = ()
+            check_kwargs = {}
+            fork_called = 0
+            fork_res = 0
+
+            def mock_run(self, *args, **kwargs):
+                self.check_args = args
+                self.check_kwargs = kwargs
+
+            def mock_sleep(self):
+                raise StopForever('stop')
+
+            def mock_fork(self):
+                self.fork_called += 1
+                return self.fork_res
+
+        my_auditor = auditor.ObjectAuditor(dict(devices=self.devices,
+                                                mount_check='false',
+                                                zero_byte_files_per_second=89))
+        mocker = ObjectAuditorMock()
+        my_auditor.run_once = mocker.mock_run
+        my_auditor._sleep = mocker.mock_sleep
+        was_fork = os.fork
+        try:
+            os.fork = mocker.mock_fork
+            self.assertRaises(StopForever,
+                              my_auditor.run_forever, zero_byte_fps=50)
+            self.assertEquals(mocker.check_kwargs['zero_byte_fps'], 50)
+            self.assertEquals(mocker.fork_called, 0)
+
+            self.assertRaises(StopForever, my_auditor.run_forever)
+            self.assertEquals(mocker.fork_called, 1)
+            self.assertEquals(mocker.check_args, ())
+
+            mocker.fork_res = 1
+            self.assertRaises(StopForever, my_auditor.run_forever)
+            self.assertEquals(mocker.fork_called, 2)
+            self.assertEquals(mocker.check_kwargs['zero_byte_fps'], 89)
+
+        finally:
+            os.fork = was_fork
 
 if __name__ == '__main__':
     unittest.main()