def HEAD(self, req):
     """Handle HTTP HEAD request."""
     drive, part, account, container, obj = split_and_validate_path(
         req, 4, 5, True)
     out_content_type = listing_formats.get_listing_content_type(req)
     try:
         check_drive(self.root, drive, self.mount_check)
     except ValueError:
         return HTTPInsufficientStorage(drive=drive, request=req)
     broker = self._get_container_broker(drive, part, account, container,
                                         pending_timeout=0.1,
                                         stale_reads_ok=True)
     info, is_deleted = broker.get_info_is_deleted()
     headers = gen_resp_headers(info, is_deleted=is_deleted)
     if is_deleted:
         return HTTPNotFound(request=req, headers=headers)
     headers.update(
         (str_to_wsgi(key), str_to_wsgi(value))
         for key, (value, timestamp) in broker.metadata.items()
         if value != '' and (key.lower() in self.save_headers or
                             is_sys_or_user_meta('container', key)))
     headers['Content-Type'] = out_content_type
     resp = HTTPNoContent(request=req, headers=headers, charset='utf-8')
     resp.last_modified = math.ceil(float(headers['X-PUT-Timestamp']))
     return resp

    def GET(self, req):
        """Handle HTTP GET request."""
        drive, part, account = split_and_validate_path(req, 3)
        prefix = get_param(req, 'prefix')
        delimiter = get_param(req, 'delimiter')
        if delimiter and (len(delimiter) > 1 or ord(delimiter) > 254):
            # delimiters can be made more flexible later
            return HTTPPreconditionFailed(body='Bad delimiter')
        limit = constraints.ACCOUNT_LISTING_LIMIT
        given_limit = get_param(req, 'limit')
        reverse = config_true_value(get_param(req, 'reverse'))
        if given_limit and given_limit.isdigit():
            limit = int(given_limit)
            if limit > constraints.ACCOUNT_LISTING_LIMIT:
                return HTTPPreconditionFailed(
                    request=req,
                    body='Maximum limit is %d' %
                    constraints.ACCOUNT_LISTING_LIMIT)
        marker = get_param(req, 'marker', '')
        end_marker = get_param(req, 'end_marker')
        out_content_type = listing_formats.get_listing_content_type(req)

        try:
            check_drive(self.root, drive, self.mount_check)
        except ValueError:
            return HTTPInsufficientStorage(drive=drive, request=req)
        broker = self._get_account_broker(drive, part, account,
                                          pending_timeout=0.1,
                                          stale_reads_ok=True)
        if broker.is_deleted():
            return self._deleted_response(broker, req, HTTPNotFound)
        return account_listing_response(account, req, out_content_type, broker,
                                        limit, marker, end_marker, prefix,
                                        delimiter, reverse)

    def test_check_drive_isdir(self):
        root = '/srv'
        path = 'sdb2'
        with mock_check_drive(isdir=True) as mocks:
            self.assertEqual('/srv/sdb2', constraints.check_dir(root, path))
            self.assertEqual('/srv/sdb2', constraints.check_drive(
                root, path, False))
            self.assertEqual([mock.call('/srv/sdb2'), mock.call('/srv/sdb2')],
                             mocks['isdir'].call_args_list)
            self.assertEqual([], mocks['ismount'].call_args_list)

        with mock_check_drive(isdir=True) as mocks:
            with self.assertRaises(ValueError) as exc_mgr:
                constraints.check_mount(root, path)
            self.assertEqual(str(exc_mgr.exception),
                             '/srv/sdb2 is not mounted')

            with self.assertRaises(ValueError) as exc_mgr:
                constraints.check_drive(root, path, True)
            self.assertEqual(str(exc_mgr.exception),
                             '/srv/sdb2 is not mounted')

            self.assertEqual([], mocks['isdir'].call_args_list)
            self.assertEqual([mock.call('/srv/sdb2'), mock.call('/srv/sdb2')],
                             mocks['ismount'].call_args_list)

    def test_check_drive_invalid_path(self):
        root = '/srv/'
        with mock_check_drive() as mocks:
            drive = 'foo?bar'
            with self.assertRaises(ValueError) as exc_mgr:
                constraints.check_dir(root, drive)
            self.assertEqual(str(exc_mgr.exception),
                             '%s is not a valid drive name' % drive)

            drive = 'foo bar'
            with self.assertRaises(ValueError) as exc_mgr:
                constraints.check_mount(root, drive)
            self.assertEqual(str(exc_mgr.exception),
                             '%s is not a valid drive name' % drive)

            drive = 'foo/bar'
            with self.assertRaises(ValueError) as exc_mgr:
                constraints.check_drive(root, drive, True)
            self.assertEqual(str(exc_mgr.exception),
                             '%s is not a valid drive name' % drive)

            drive = 'foo%bar'
            with self.assertRaises(ValueError) as exc_mgr:
                constraints.check_drive(root, drive, False)
            self.assertEqual(str(exc_mgr.exception),
                             '%s is not a valid drive name' % drive)
        self.assertEqual([], mocks['isdir'].call_args_list)
        self.assertEqual([], mocks['ismount'].call_args_list)

 def test_check_drive_invalid_path(self):
     root = '/srv/'
     with mock_check_drive() as mocks:
         self.assertIsNone(constraints.check_dir(root, 'foo?bar'))
         self.assertIsNone(constraints.check_mount(root, 'foo bar'))
         self.assertIsNone(constraints.check_drive(root, 'foo/bar', True))
         self.assertIsNone(constraints.check_drive(root, 'foo%bar', False))
     self.assertEqual([], mocks['isdir'].call_args_list)
     self.assertEqual([], mocks['ismount'].call_args_list)

 def DELETE(self, req):
     """Handle HTTP DELETE request."""
     drive, part, account = split_and_validate_path(req, 3)
     try:
         check_drive(self.root, drive, self.mount_check)
     except ValueError:
         return HTTPInsufficientStorage(drive=drive, request=req)
     req_timestamp = valid_timestamp(req)
     broker = self._get_account_broker(drive, part, account)
     if broker.is_deleted():
         return self._deleted_response(broker, req, HTTPNotFound)
     broker.delete_db(req_timestamp.internal)
     return self._deleted_response(broker, req, HTTPNoContent)

 def POST(self, req):
     """Handle HTTP POST request."""
     drive, part, account, container = split_and_validate_path(req, 4)
     req_timestamp = valid_timestamp(req)
     if 'x-container-sync-to' in req.headers:
         err, sync_to, realm, realm_key = validate_sync_to(
             req.headers['x-container-sync-to'], self.allowed_sync_hosts,
             self.realms_conf)
         if err:
             return HTTPBadRequest(err)
     if not check_drive(self.root, drive, self.mount_check):
         return HTTPInsufficientStorage(drive=drive, request=req)
     broker = self._get_container_broker(drive, part, account, container)
     if broker.is_deleted():
         return HTTPNotFound(request=req)
     broker.update_put_timestamp(req_timestamp.internal)
     metadata = {}
     metadata.update(
         (key, (value, req_timestamp.internal))
         for key, value in req.headers.items()
         if key.lower() in self.save_headers or
         is_sys_or_user_meta('container', key))
     if metadata:
         if 'X-Container-Sync-To' in metadata:
             if 'X-Container-Sync-To' not in broker.metadata or \
                     metadata['X-Container-Sync-To'][0] != \
                     broker.metadata['X-Container-Sync-To'][0]:
                 broker.set_x_container_sync_points(-1, -1)
         broker.update_metadata(metadata, validate_metadata=True)
         self._update_sync_store(broker, 'POST')
     return HTTPNoContent(request=req)

 def test_check_drive_isdir(self):
     root = '/srv'
     path = 'sdb2'
     with mock_check_drive(isdir=True) as mocks:
         self.assertEqual('/srv/sdb2', constraints.check_dir(root, path))
         self.assertEqual('/srv/sdb2', constraints.check_drive(
             root, path, False))
         self.assertEqual([mock.call('/srv/sdb2'), mock.call('/srv/sdb2')],
                          mocks['isdir'].call_args_list)
         self.assertEqual([], mocks['ismount'].call_args_list)
     with mock_check_drive(isdir=True) as mocks:
         self.assertIsNone(constraints.check_mount(root, path))
         self.assertIsNone(constraints.check_drive(root, path, True))
         self.assertEqual([], mocks['isdir'].call_args_list)
         self.assertEqual([mock.call('/srv/sdb2'), mock.call('/srv/sdb2')],
                          mocks['ismount'].call_args_list)

 def REPLICATE(self, req):
     """
     Handle HTTP REPLICATE request (json-encoded RPC calls for replication.)
     """
     post_args = split_and_validate_path(req, 3)
     drive, partition, hash = post_args
     try:
         check_drive(self.root, drive, self.mount_check)
     except ValueError:
         return HTTPInsufficientStorage(drive=drive, request=req)
     if not self.check_free_space(drive):
         return HTTPInsufficientStorage(drive=drive, request=req)
     try:
         args = json.load(req.environ['wsgi.input'])
     except ValueError as err:
         return HTTPBadRequest(body=str(err), content_type='text/plain')
     ret = self.replicator_rpc.dispatch(post_args, args)
     ret.request = req
     return ret

 def POST(self, req):
     """Handle HTTP POST request."""
     drive, part, account = split_and_validate_path(req, 3)
     req_timestamp = valid_timestamp(req)
     try:
         check_drive(self.root, drive, self.mount_check)
     except ValueError:
         return HTTPInsufficientStorage(drive=drive, request=req)
     if not self.check_free_space(drive):
         return HTTPInsufficientStorage(drive=drive, request=req)
     broker = self._get_account_broker(drive, part, account)
     if broker.is_deleted():
         return self._deleted_response(broker, req, HTTPNotFound)
     metadata = {}
     metadata.update((key, (value, req_timestamp.internal))
                     for key, value in req.headers.items()
                     if is_sys_or_user_meta('account', key))
     if metadata:
         broker.update_metadata(metadata, validate_metadata=True)
     return HTTPNoContent(request=req)

    def UPDATE(self, req):
        """
        Handle HTTP UPDATE request (merge_items RPCs coming from the proxy.)
        """
        drive, part, account, container = split_and_validate_path(req, 4)
        req_timestamp = valid_timestamp(req)
        try:
            check_drive(self.root, drive, self.mount_check)
        except ValueError:
            return HTTPInsufficientStorage(drive=drive, request=req)
        if not self.check_free_space(drive):
            return HTTPInsufficientStorage(drive=drive, request=req)

        requested_policy_index = self.get_and_validate_policy_index(req)
        broker = self._get_container_broker(drive, part, account, container)
        self._maybe_autocreate(broker, req_timestamp, account,
                               requested_policy_index)
        try:
            objs = json.load(req.environ['wsgi.input'])
        except ValueError as err:
            return HTTPBadRequest(body=str(err), content_type='text/plain')
        broker.merge_items(objs)
        return HTTPAccepted(request=req)

 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
     repeatedly by :func:`run_forever`. This will call :func:`reap_device`
     once for each device on the server.
     """
     self.logger.debug('Begin devices pass: %s', self.devices)
     begin = time()
     try:
         for device in os.listdir(self.devices):
             try:
                 check_drive(self.devices, device, self.mount_check)
             except ValueError as err:
                 self.logger.increment('errors')
                 self.logger.debug('Skipping: %s', err)
                 continue
             self.reap_device(device)
     except (Exception, Timeout):
         self.logger.exception(_("Exception in top-level account reaper "
                                 "loop"))
     elapsed = time() - begin
     self.logger.info(_('Devices pass completed: %.02fs'), elapsed)

 def run_forever(self, *args, **kwargs):
     """Run the updater continuously."""
     time.sleep(random() * self.interval)
     while True:
         self.logger.info(_('Begin object update sweep'))
         begin = time.time()
         pids = []
         # read from container ring to ensure it's fresh
         self.get_container_ring().get_nodes('')
         for device in self._listdir(self.devices):
             if not check_drive(self.devices, device, self.mount_check):
                 # We don't count this as an error. The occasional
                 # unmounted drive is part of normal cluster operations,
                 # so a simple warning is sufficient.
                 self.logger.warning(
                     _('Skipping %s as it is not mounted'), device)
                 continue
             while len(pids) >= self.concurrency:
                 pids.remove(os.wait()[0])
             pid = os.fork()
             if pid:
                 pids.append(pid)
             else:
                 signal.signal(signal.SIGTERM, signal.SIG_DFL)
                 eventlet_monkey_patch()
                 self.stats.reset()
                 forkbegin = time.time()
                 self.object_sweep(os.path.join(self.devices, device))
                 elapsed = time.time() - forkbegin
                 self.logger.info(
                     ('Object update sweep of %(device)s '
                      'completed: %(elapsed).02fs, %(stats)s'),
                     {'device': device, 'elapsed': elapsed,
                      'stats': self.stats})
                 sys.exit()
         while pids:
             pids.remove(os.wait()[0])
         elapsed = time.time() - begin
         self.logger.info(_('Object update sweep completed: %.02fs'),
                          elapsed)
         dump_recon_cache({'object_updater_sweep': elapsed},
                          self.rcache, self.logger)
         if elapsed < self.interval:
             time.sleep(self.interval - elapsed)

 def run_forever(self, *args, **kwargs):
     """Run the updater continuously."""
     time.sleep(random() * self.interval)
     while True:
         self.logger.info(_('Begin object update sweep'))
         begin = time.time()
         pids = []
         # read from container ring to ensure it's fresh
         self.get_container_ring().get_nodes('')
         for device in self._listdir(self.devices):
             if not check_drive(self.devices, device, self.mount_check):
                 self.logger.increment('errors')
                 self.logger.warning(
                     _('Skipping %s as it is not mounted'), device)
                 continue
             while len(pids) >= self.concurrency:
                 pids.remove(os.wait()[0])
             pid = os.fork()
             if pid:
                 pids.append(pid)
             else:
                 signal.signal(signal.SIGTERM, signal.SIG_DFL)
                 eventlet_monkey_patch()
                 self.successes = 0
                 self.failures = 0
                 forkbegin = time.time()
                 self.object_sweep(os.path.join(self.devices, device))
                 elapsed = time.time() - forkbegin
                 self.logger.info(
                     _('Object update sweep of %(device)s'
                       ' completed: %(elapsed).02fs, %(success)s successes'
                       ', %(fail)s failures'),
                     {'device': device, 'elapsed': elapsed,
                      'success': self.successes, 'fail': self.failures})
                 sys.exit()
         while pids:
             pids.remove(os.wait()[0])
         elapsed = time.time() - begin
         self.logger.info(_('Object update sweep completed: %.02fs'),
                          elapsed)
         dump_recon_cache({'object_updater_sweep': elapsed},
                          self.rcache, self.logger)
         if elapsed < self.interval:
             time.sleep(self.interval - elapsed)

    def get_paths(self):
        """
        Get paths to all of the partitions on each drive to be processed.

        :returns: a list of paths
        """
        paths = []
        for device in self._listdir(self.devices):
            dev_path = check_drive(self.devices, device, self.mount_check)
            if not dev_path:
                self.logger.warning(_('%s is not mounted'), device)
                continue
            con_path = os.path.join(dev_path, DATADIR)
            if not os.path.exists(con_path):
                continue
            for partition in self._listdir(con_path):
                paths.append(os.path.join(con_path, partition))
        shuffle(paths)
        return paths

 def run_once(self, *args, **kwargs):
     """Run a replication pass once."""
     self._zero_stats()
     dirs = []
     ips = whataremyips(self.bind_ip)
     if not ips:
         self.logger.error(_('ERROR Failed to get my own IPs?'))
         return
     self._local_device_ids = set()
     found_local = False
     for node in self.ring.devs:
         if node and is_local_device(ips, self.port,
                                     node['replication_ip'],
                                     node['replication_port']):
             found_local = True
             if not check_drive(self.root, node['device'],
                                self.mount_check):
                 self._add_failure_stats(
                     [(failure_dev['replication_ip'],
                       failure_dev['device'])
                      for failure_dev in self.ring.devs if failure_dev])
                 self.logger.warning(
                     _('Skipping %(device)s as it is not mounted') % node)
                 continue
             unlink_older_than(
                 os.path.join(self.root, node['device'], 'tmp'),
                 time.time() - self.reclaim_age)
             datadir = os.path.join(self.root, node['device'], self.datadir)
             if os.path.isdir(datadir):
                 self._local_device_ids.add(node['id'])
                 dirs.append((datadir, node['id']))
     if not found_local:
         self.logger.error("Can't find itself %s with port %s in ring "
                           "file, not replicating",
                           ", ".join(ips), self.port)
     self.logger.info(_('Beginning replication run'))
     for part, object_file, node_id in roundrobin_datadirs(dirs):
         self.cpool.spawn_n(
             self._replicate_object, part, object_file, node_id)
     self.cpool.waitall()
     self.logger.info(_('Replication run OVER'))
     self._report_stats()

 def dispatch(self, replicate_args, args):
     if not hasattr(args, 'pop'):
         return HTTPBadRequest(body='Invalid object type')
     op = args.pop(0)
     drive, partition, hsh = replicate_args
     if not check_drive(self.root, drive, self.mount_check):
         return Response(status='507 %s is not mounted' % drive)
     db_file = os.path.join(self.root, drive,
                            storage_directory(self.datadir, partition, hsh),
                            hsh + '.db')
     if op == 'rsync_then_merge':
         return self.rsync_then_merge(drive, db_file, args)
     if op == 'complete_rsync':
         return self.complete_rsync(drive, db_file, args)
     else:
         # someone might be about to rsync a db to us,
         # make sure there's a tmp dir to receive it.
         mkdirs(os.path.join(self.root, drive, 'tmp'))
         if not os.path.exists(db_file):
             return HTTPNotFound()
         return getattr(self, op)(self.broker_class(db_file), args)

 def run_once(self, *args, **kwargs):
     """Run the updater once."""
     self.logger.info(_('Begin object update single threaded sweep'))
     begin = time.time()
     self.stats.reset()
     for device in self._listdir(self.devices):
         if not check_drive(self.devices, device, self.mount_check):
             # We don't count this as an error. The occasional unmounted
             # drive is part of normal cluster operations, so a simple
             # warning is sufficient.
             self.logger.warning(
                 _('Skipping %s as it is not mounted'), device)
             continue
         self.object_sweep(os.path.join(self.devices, device))
     elapsed = time.time() - begin
     self.logger.info(
         ('Object update single-threaded sweep completed: '
          '%(elapsed).02fs, %(stats)s'),
         {'elapsed': elapsed, 'stats': self.stats})
     dump_recon_cache({'object_updater_sweep': elapsed},
                      self.rcache, self.logger)

 def run_once(self, *args, **kwargs):
     """Run the updater once."""
     self.logger.info(_('Begin object update single threaded sweep'))
     begin = time.time()
     self.successes = 0
     self.failures = 0
     for device in self._listdir(self.devices):
         if not check_drive(self.devices, device, self.mount_check):
             self.logger.increment('errors')
             self.logger.warning(
                 _('Skipping %s as it is not mounted'), device)
             continue
         self.object_sweep(os.path.join(self.devices, device))
     elapsed = time.time() - begin
     self.logger.info(
         _('Object update single threaded sweep completed: '
           '%(elapsed).02fs, %(success)s successes, %(fail)s failures'),
         {'elapsed': elapsed, 'success': self.successes,
          'fail': self.failures})
     dump_recon_cache({'object_updater_sweep': elapsed},
                      self.rcache, self.logger)

 def GET(self, req):
     """Handle HTTP GET request."""
     drive, part, account, container, obj = split_and_validate_path(
         req, 4, 5, True)
     path = get_param(req, 'path')
     prefix = get_param(req, 'prefix')
     delimiter = get_param(req, 'delimiter')
     if delimiter and (len(delimiter) > 1 or ord(delimiter) > 254):
         # delimiters can be made more flexible later
         return HTTPPreconditionFailed(body='Bad delimiter')
     marker = get_param(req, 'marker', '')
     end_marker = get_param(req, 'end_marker')
     limit = constraints.CONTAINER_LISTING_LIMIT
     given_limit = get_param(req, 'limit')
     reverse = config_true_value(get_param(req, 'reverse'))
     if given_limit and given_limit.isdigit():
         limit = int(given_limit)
         if limit > constraints.CONTAINER_LISTING_LIMIT:
             return HTTPPreconditionFailed(
                 request=req,
                 body='Maximum limit is %d'
                 % constraints.CONTAINER_LISTING_LIMIT)
     out_content_type = listing_formats.get_listing_content_type(req)
     if not check_drive(self.root, drive, self.mount_check):
         return HTTPInsufficientStorage(drive=drive, request=req)
     broker = self._get_container_broker(drive, part, account, container,
                                         pending_timeout=0.1,
                                         stale_reads_ok=True)
     info, is_deleted = broker.get_info_is_deleted()
     resp_headers = gen_resp_headers(info, is_deleted=is_deleted)
     if is_deleted:
         return HTTPNotFound(request=req, headers=resp_headers)
     container_list = broker.list_objects_iter(
         limit, marker, end_marker, prefix, delimiter, path,
         storage_policy_index=info['storage_policy_index'], reverse=reverse)
     return self.create_listing(req, out_content_type, info, resp_headers,
                                broker.metadata, container_list, container)