# Copyright (c) 2010-2012 OpenStack Foundation # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or # implied. # See the License for the specific language governing permissions and # limitations under the License. import copy import errno import os import time import json from hashlib import md5 import logging import traceback from os.path import basename, dirname, join from random import shuffle from contextlib import contextmanager from collections import defaultdict from eventlet import Timeout, tpool import six from swift import gettext_ as _ from swift.common.constraints import check_drive from swift.common.request_helpers import is_sys_meta from swift.common.utils import fdatasync, \ config_true_value, listdir, split_path, lock_path from swift.common.exceptions import DiskFileQuarantined, DiskFileNotExist, \ DiskFileCollision, DiskFileNoSpace, DiskFileDeviceUnavailable, \ DiskFileError, PathNotDir, \ DiskFileExpired, DiskFileXattrNotSupported, \ DiskFileBadMetadataChecksum from swift.common.storage_policy import ( split_policy_string, POLICIES, REPL_POLICY, EC_POLICY) from swift.obj import vfile from swift.obj.diskfile import BaseDiskFileManager, DiskFileManager, \ ECDiskFileManager, BaseDiskFile, DiskFile, DiskFileReader, DiskFileWriter,\ BaseDiskFileReader, BaseDiskFileWriter, ECDiskFile, ECDiskFileReader, \ ECDiskFileWriter, AuditLocation, RESERVED_DATAFILE_META, \ DATAFILE_SYSTEM_META, strip_self, DEFAULT_RECLAIM_AGE, _encode_metadata, \ get_part_path, get_data_dir, update_auditor_status, extract_policy, \ HASH_FILE, read_hashes, write_hashes, consolidate_hashes, invalidate_hash def quarantine_vrenamer(device_path, corrupted_file_path): """ In the case that a file is corrupted, move it to a quarantined area to allow replication to fix it. :params device_path: The path to the device the corrupted file is on. :params corrupted_file_path: The path to the file you want quarantined. :returns: path (str) of directory the file was moved to :raises OSError: re-raises non errno.EEXIST / errno.ENOTEMPTY exceptions from rename """ policy = extract_policy(corrupted_file_path) if policy is None: # TODO: support a quarantine-unknown location policy = POLICIES.legacy # rename key in KV return vfile.quarantine_ohash(dirname(corrupted_file_path), policy) def object_audit_location_generator(devices, datadir, mount_check=True, logger=None, device_dirs=None, auditor_type="ALL"): """ Given a devices path (e.g. "/srv/node"), yield an AuditLocation for all objects stored under that directory for the given datadir (policy), if device_dirs isn't set. If device_dirs is set, only yield AuditLocation for the objects under the entries in device_dirs. The AuditLocation only knows the path to the hash directory, not to the .data file therein (if any). This is to avoid a double listdir(hash_dir); the DiskFile object will always do one, so we don't. :param devices: parent directory of the devices to be audited :param datadir: objects directory :param mount_check: flag to check if a mount check should be performed on devices :param logger: a logger object :param device_dirs: a list of directories under devices to traverse :param auditor_type: either ALL or ZBF """ if not device_dirs: device_dirs = listdir(devices) else: # remove bogus devices and duplicates from device_dirs device_dirs = list( set(listdir(devices)).intersection(set(device_dirs))) # randomize devices in case of process restart before sweep completed shuffle(device_dirs) base, policy = split_policy_string(datadir) for device in device_dirs: if not check_drive(devices, device, mount_check): if logger: logger.debug( 'Skipping %s as it is not %s', device, 'mounted' if mount_check else 'a dir') continue datadir_path = os.path.join(devices, device, datadir) if not os.path.exists(datadir_path): continue partitions = get_auditor_status(datadir_path, logger, auditor_type) for pos, partition in enumerate(partitions): update_auditor_status(datadir_path, logger, partitions[pos:], auditor_type) part_path = os.path.join(datadir_path, partition) try: suffixes = vfile.listdir(part_path) except OSError as e: if e.errno != errno.ENOTDIR: raise continue for asuffix in suffixes: suff_path = os.path.join(part_path, asuffix) try: hashes = vfile.listdir(suff_path) except OSError as e: if e.errno != errno.ENOTDIR: raise continue for hsh in hashes: hsh_path = os.path.join(suff_path, hsh) yield AuditLocation(hsh_path, device, partition, policy) update_auditor_status(datadir_path, logger, [], auditor_type) def get_auditor_status(datadir_path, logger, auditor_type): auditor_status = os.path.join( datadir_path, "auditor_status_%s.json" % auditor_type) status = {} try: if six.PY3: statusfile = open(auditor_status, encoding='utf8') else: statusfile = open(auditor_status, 'rb') with statusfile: status = statusfile.read() except (OSError, IOError) as e: if e.errno != errno.ENOENT and logger: logger.warning(_('Cannot read %(auditor_status)s (%(err)s)') % {'auditor_status': auditor_status, 'err': e}) return vfile.listdir(datadir_path) try: status = json.loads(status) except ValueError as e: logger.warning(_('Loading JSON from %(auditor_status)s failed' ' (%(err)s)') % {'auditor_status': auditor_status, 'err': e}) return vfile.listdir(datadir_path) return status['partitions'] class BaseKVFile(BaseDiskFile): # Todo: we may want a separate __init__ to define KV specific attribute def open(self, modernize=False, current_time=None): """ Open the object. This implementation opens the data file representing the object, reads the associated metadata in the extended attributes, additionally combining metadata from fast-POST `.meta` files. :param modernize: if set, update this diskfile to the latest format. Currently, this means adding metadata checksums if none are present. :param current_time: Unix time used in checking expiration. If not present, the current time will be used. .. note:: An implementation is allowed to raise any of the following exceptions, but is only required to raise `DiskFileNotExist` when the object representation does not exist. :raises DiskFileCollision: on name mis-match with metadata :raises DiskFileNotExist: if the object does not exist :raises DiskFileDeleted: if the object was previously deleted :raises DiskFileQuarantined: if while reading metadata of the file some data did pass cross checks :returns: itself for use as a context manager """ try: files = vfile.listdir(self._datadir) except (OSError, vfile.VFileException) as err: raise DiskFileError( "Error listing directory %s: %s" % (self._datadir, err)) # gather info about the valid files to use to open the DiskFile file_info = self._get_ondisk_files(files) self._data_file = file_info.get('data_file') if not self._data_file: raise self._construct_exception_from_ts_file(**file_info) self._vfr = self._construct_from_data_file( current_time=current_time, modernize=modernize, **file_info) # This method must populate the internal _metadata attribute. self._metadata = self._metadata or {} return self def _verify_data_file(self, data_file, data_vfr, current_time): """ Verify the metadata's name value matches what we think the object is named. :param data_file: data file name being consider, used when quarantines occur :param fp: open file pointer so that we can `fstat()` the file to verify the on-disk size with Content-Length metadata value :param current_time: Unix time used in checking expiration :raises DiskFileCollision: if the metadata stored name does not match the referenced name of the file :raises DiskFileExpired: if the object has expired :raises DiskFileQuarantined: if data inconsistencies were detected between the metadata and the file-system metadata """ try: mname = self._metadata['name'] except KeyError: raise self._quarantine(data_file, "missing name metadata") else: if mname != self._name: self._logger.error( _('Client path %(client)s does not match ' 'path stored in object metadata %(meta)s'), {'client': self._name, 'meta': mname}) raise DiskFileCollision('Client path does not match path ' 'stored in object metadata') try: x_delete_at = int(self._metadata['X-Delete-At']) except KeyError: pass except ValueError: # Quarantine, the x-delete-at key is present but not an # integer. raise self._quarantine( data_file, "bad metadata x-delete-at value %s" % ( self._metadata['X-Delete-At'])) else: if current_time is None: current_time = time.time() if x_delete_at <= current_time and not self._open_expired: raise DiskFileExpired(metadata=self._metadata) try: metadata_size = int(self._metadata['Content-Length']) except KeyError: raise self._quarantine( data_file, "missing content-length in metadata") except ValueError: # Quarantine, the content-length key is present but not an # integer. raise self._quarantine( data_file, "bad metadata content-length value %s" % ( self._metadata['Content-Length'])) obj_size = data_vfr.data_size if obj_size != metadata_size: raise self._quarantine( data_file, "metadata content-length %s does" " not match actual object size %s" % ( metadata_size, obj_size)) self._content_length = obj_size return obj_size def _failsafe_read_metadata(self, source, quarantine_filename=None, add_missing_checksum=False): """ Read metadata from source object file. In case of failure, quarantine the file. Takes source and filename separately so we can read from an open file if we have one. :param source: file descriptor or filename to load the metadata from :param quarantine_filename: full path of file to load the metadata from :param add_missing_checksum: ignored """ try: vfr = vfile.VFileReader.get_vfile(source, self._logger) vfr_metadata = vfr.metadata vfr.close() return vfr_metadata except (DiskFileXattrNotSupported, DiskFileNotExist): raise except DiskFileBadMetadataChecksum as err: raise self._quarantine(quarantine_filename, str(err)) except Exception as err: raise self._quarantine( quarantine_filename, "Exception reading metadata: %s" % err) # This could be left unchanged now that _failsafe_read_metadata() is # patched. Test it. def _merge_content_type_metadata(self, ctype_file): """ When a second .meta file is providing the most recent Content-Type metadata then merge it into the metafile_metadata. :param ctype_file: An on-disk .meta file """ try: ctype_vfr = vfile.VFileReader.get_vfile(ctype_file, self._logger) except IOError as e: if e.errno == errno.ENOENT: raise DiskFileNotExist() raise ctypefile_metadata = ctype_vfr.metadata ctype_vfr.close() if ('Content-Type' in ctypefile_metadata and (ctypefile_metadata.get('Content-Type-Timestamp') > self._metafile_metadata.get('Content-Type-Timestamp')) and (ctypefile_metadata.get('Content-Type-Timestamp') > self.data_timestamp)): self._metafile_metadata['Content-Type'] = \ ctypefile_metadata['Content-Type'] self._metafile_metadata['Content-Type-Timestamp'] = \ ctypefile_metadata.get('Content-Type-Timestamp') def _construct_from_data_file(self, data_file, meta_file, ctype_file, current_time, modernize=False, **kwargs): """ Open the `.data` file to fetch its metadata, and fetch the metadata from fast-POST `.meta` files as well if any exist, merging them properly. :param data_file: on-disk `.data` file being considered :param meta_file: on-disk fast-POST `.meta` file being considered :param ctype_file: on-disk fast-POST `.meta` file being considered that contains content-type and content-type timestamp :param current_time: Unix time used in checking expiration :param modernize: ignored :returns: an opened data file pointer :raises DiskFileError: various exceptions from :func:`swift.obj.diskfile.DiskFile._verify_data_file` """ # TODO: need to catch exception, check if ENOENT (see in diskfile) try: data_vfr = vfile.VFileReader.get_vfile(data_file, self._logger) except IOError as e: if e.errno == errno.ENOENT: raise DiskFileNotExist() raise self._datafile_metadata = data_vfr.metadata self._metadata = {} if meta_file: self._metafile_metadata = self._failsafe_read_metadata( meta_file, meta_file) if ctype_file and ctype_file != meta_file: self._merge_content_type_metadata(ctype_file) sys_metadata = dict( [(key, val) for key, val in self._datafile_metadata.items() if key.lower() in (RESERVED_DATAFILE_META | DATAFILE_SYSTEM_META) or is_sys_meta('object', key)]) self._metadata.update(self._metafile_metadata) self._metadata.update(sys_metadata) # diskfile writer added 'name' to metafile, so remove it here self._metafile_metadata.pop('name', None) # TODO: the check for Content-Type is only here for tests that # create .data files without Content-Type if ('Content-Type' in self._datafile_metadata and (self.data_timestamp > self._metafile_metadata.get('Content-Type-Timestamp'))): self._metadata['Content-Type'] = \ self._datafile_metadata['Content-Type'] self._metadata.pop('Content-Type-Timestamp', None) else: self._metadata.update(self._datafile_metadata) if self._name is None: # If we don't know our name, we were just given a hash dir at # instantiation, so we'd better validate that the name hashes back # to us self._name = self._metadata['name'] self._verify_name_matches_hash(data_file) self._verify_data_file(data_file, data_vfr, current_time) return data_vfr def reader(self, keep_cache=False, _quarantine_hook=lambda m: None): """ Return a :class:`swift.common.swob.Response` class compatible "`app_iter`" object as defined by :class:`swift.obj.diskfile.DiskFileReader`. For this implementation, the responsibility of closing the open file is passed to the :class:`swift.obj.diskfile.DiskFileReader` object. :param keep_cache: caller's preference for keeping data read in the OS buffer cache :param _quarantine_hook: 1-arg callable called when obj quarantined; the arg is the reason for quarantine. Default is to ignore it. Not needed by the REST layer. :returns: a :class:`swift.obj.diskfile.DiskFileReader` object """ dr = self.reader_cls( self._vfr, self._data_file, int(self._metadata['Content-Length']), self._metadata['ETag'], self._disk_chunk_size, self._manager.keep_cache_size, self._device_path, self._logger, use_splice=self._use_splice, quarantine_hook=_quarantine_hook, pipe_size=self._pipe_size, diskfile=self, keep_cache=keep_cache) # At this point the reader object is now responsible for closing # the file pointer. # self._fp = None self._vfr = None return dr def writer(self, size=None): return self.writer_cls(self._manager, self._name, self._datadir, size, self._bytes_per_sync, self, self._logger) def _get_tempfile(self): raise Exception("_get_tempfile called, shouldn't happen") @contextmanager def create(self, size=None, extension=None): """ Context manager to create a vfile. It could create separate volumes based on the extension. Currently no caller will set the extension. The easiest would be to patch server.py, in DELETE(), add an extension=".ts" parameter to the self.get_diskfile() call, as kwargs is passed all the way down to here. It's also possible to have the writer_cls handle the volume creation later: at the first write() call, if any, assume it's not a tombstone, and in put(), check self._extension. :param size: optional initial size of file to explicitly allocate on disk :param extension: file extension, with dot ('.ts') :raises DiskFileNoSpace: if a size is specified and allocation fails """ dfw = self.writer(size) try: yield dfw.open() finally: dfw.close() class BaseKVFileReader(BaseDiskFileReader): def __init__(self, vfr, data_file, obj_size, etag, disk_chunk_size, keep_cache_size, device_path, logger, quarantine_hook, use_splice, pipe_size, diskfile, keep_cache=False): # Parameter tracking self._vfr = vfr self._data_file = data_file self._obj_size = obj_size self._etag = etag self._diskfile = diskfile self._disk_chunk_size = disk_chunk_size self._device_path = device_path self._logger = logger self._quarantine_hook = quarantine_hook self._use_splice = use_splice self._pipe_size = pipe_size if keep_cache: # Caller suggests we keep this in cache, only do it if the # object's size is less than the maximum. self._keep_cache = obj_size < keep_cache_size else: self._keep_cache = False # Internal Attributes self._iter_etag = None self._bytes_read = 0 self._started_at_0 = False self._read_to_eof = False self._md5_of_sent_bytes = None self._suppress_file_closing = False self._quarantined_dir = None def _init_checks(self): if self._vfr.tell() == 0: self._started_at_0 = True self._iter_etag = md5() def __iter__(self): """Returns an iterator over the data file.""" try: self._bytes_read = 0 self._started_at_0 = False self._read_to_eof = False self._init_checks() while True: chunk = self._vfr.read(self._disk_chunk_size) if chunk: self._update_checks(chunk) self._bytes_read += len(chunk) yield chunk else: self._read_to_eof = True break finally: if not self._suppress_file_closing: self.close() def can_zero_copy_send(self): return False def app_iter_range(self, start, stop): """ Returns an iterator over the data file for range (start, stop) """ if start or start == 0: self._vfr.seek(start) if stop is not None: length = stop - start else: length = None try: for chunk in self: if length is not None: length -= len(chunk) if length < 0: # Chop off the extra: yield chunk[:length] break yield chunk finally: if not self._suppress_file_closing: self.close() def close(self): """ Close the open file handle if present. For this specific implementation, this method will handle quarantining the file if necessary. """ if self._vfr: try: if self._started_at_0 and self._read_to_eof: self._handle_close_quarantine() except DiskFileQuarantined: raise except (Exception, Timeout) as e: self._logger.error(_( 'ERROR DiskFile %(data_file)s' ' close failure: %(exc)s : %(stack)s'), {'exc': e, 'stack': ''.join(traceback.format_stack()), 'data_file': self._data_file}) finally: vfr, self._vfr = self._vfr, None vfr.close() class BaseKVFileWriter(BaseDiskFileWriter): def __init__(self, mgr, name, datadir, size, bytes_per_sync, diskfile, logger): # Parameter tracking self._manager = mgr self._name = name self._datadir = datadir self._vfile_writer = None self._tmppath = None self._size = size self._chunks_etag = md5() self._bytes_per_sync = bytes_per_sync self._diskfile = diskfile self._logger = logger # Internal attributes self._upload_size = 0 self._last_sync = 0 self._extension = '.data' self._put_succeeded = False def open(self): if self._vfile_writer is not None: raise ValueError('DiskFileWriter is already open') try: # TODO: support extension self._vfile_writer = vfile.VFileWriter.create( self._datadir, self._size, self._manager.vfile_conf, self._logger, extension=None) except OSError as err: if err.errno in (errno.ENOSPC, errno.EDQUOT): # No more inodes in filesystem raise DiskFileNoSpace(err.strerror) raise return self def close(self): if self._vfile_writer: try: os.close(self._vfile_writer.fd) os.close(self._vfile_writer.lock_fd) except OSError: pass self._vfile_writer = None def write(self, chunk): """ Write a chunk of data to disk. All invocations of this method must come before invoking the :func: :param chunk: the chunk of data to write as a string object :returns: the total number of bytes written to an object """ if not self._vfile_writer: raise ValueError('Writer is not open') self._chunks_etag.update(chunk) while chunk: written = os.write(self._vfile_writer.fd, chunk) self._upload_size += written chunk = chunk[written:] # For large files sync every 512MB (by default) written diff = self._upload_size - self._last_sync if diff >= self._bytes_per_sync: tpool.execute(fdatasync, self._vfile_writer.fd) # drop_buffer_cache(self._vfile_writer.fd, self._last_sync, diff) self._last_sync = self._upload_size return self._upload_size def _finalize_put(self, metadata, target_path, cleanup): filename = basename(target_path) # write metadata and sync self._vfile_writer.commit(filename, _encode_metadata(metadata)) self._put_succeeded = True if cleanup: try: self.manager.cleanup_ondisk_files(self._datadir)['files'] except OSError: logging.exception(_('Problem cleaning up %s'), self._datadir) class KVFileReader(BaseKVFileReader, DiskFileReader): pass class KVFileWriter(BaseKVFileWriter, DiskFileWriter): def put(self, metadata): """ Finalize writing the file on disk. :param metadata: dictionary of metadata to be associated with the object """ super(KVFileWriter, self)._put(metadata, True) class KVFile(BaseKVFile, DiskFile): reader_cls = KVFileReader writer_cls = KVFileWriter class BaseKVFileManager(BaseDiskFileManager): diskfile_cls = None # must be set by subclasses invalidate_hash = strip_self(invalidate_hash) consolidate_hashes = strip_self(consolidate_hashes) quarantine_renamer = strip_self(quarantine_vrenamer) def __init__(self, conf, logger): self.logger = logger self.devices = conf.get('devices', '/srv/node') self.disk_chunk_size = int(conf.get('disk_chunk_size', 65536)) self.keep_cache_size = int(conf.get('keep_cache_size', 5242880)) self.bytes_per_sync = int(conf.get('mb_per_sync', 512)) * 1024 * 1024 # vfile specific config self.vfile_conf = {} self.vfile_conf['volume_alloc_chunk_size'] = int( conf.get('volume_alloc_chunk_size', 16 * 1024)) self.vfile_conf['volume_low_free_space'] = int( conf.get('volume_low_free_space', 8 * 1024)) self.vfile_conf['metadata_reserve'] = int( conf.get('metadata_reserve', 500)) self.vfile_conf['max_volume_count'] = int( conf.get('max_volume_count', 1000)) self.vfile_conf['max_volume_size'] = int( conf.get('max_volume_size', 10 * 1024 * 1024 * 1024)) # end vfile specific config self.mount_check = config_true_value(conf.get('mount_check', 'true')) self.reclaim_age = int(conf.get('reclaim_age', DEFAULT_RECLAIM_AGE)) replication_concurrency_per_device = conf.get( 'replication_concurrency_per_device') replication_one_per_device = conf.get('replication_one_per_device') if replication_concurrency_per_device is None \ and replication_one_per_device is not None: self.logger.warning('Option replication_one_per_device is ' 'deprecated and will be removed in a future ' 'version. Update your configuration to use ' 'option replication_concurrency_per_device.') if config_true_value(replication_one_per_device): replication_concurrency_per_device = 1 else: replication_concurrency_per_device = 0 elif replication_one_per_device is not None: self.logger.warning('Option replication_one_per_device ignored as ' 'replication_concurrency_per_device is ' 'defined.') if replication_concurrency_per_device is None: self.replication_concurrency_per_device = 1 else: self.replication_concurrency_per_device = int( replication_concurrency_per_device) self.replication_lock_timeout = int(conf.get( 'replication_lock_timeout', 15)) self.use_splice = False self.pipe_size = None def cleanup_ondisk_files(self, hsh_path, **kwargs): """ Clean up on-disk files that are obsolete and gather the set of valid on-disk files for an object. :param hsh_path: object hash path :param frag_index: if set, search for a specific fragment index .data file, otherwise accept the first valid .data file :returns: a dict that may contain: valid on disk files keyed by their filename extension; a list of obsolete files stored under the key 'obsolete'; a list of files remaining in the directory, reverse sorted, stored under the key 'files'. """ def is_reclaimable(timestamp): return (time.time() - float(timestamp)) > self.reclaim_age files = vfile.listdir(hsh_path) files.sort(reverse=True) results = self.get_ondisk_files( files, hsh_path, verify=False, **kwargs) if 'ts_info' in results and is_reclaimable( results['ts_info']['timestamp']): remove_vfile(join(hsh_path, results['ts_info']['filename'])) files.remove(results.pop('ts_info')['filename']) for file_info in results.get('possible_reclaim', []): # stray files are not deleted until reclaim-age if is_reclaimable(file_info['timestamp']): results.setdefault('obsolete', []).append(file_info) for file_info in results.get('obsolete', []): remove_vfile(join(hsh_path, file_info['filename'])) files.remove(file_info['filename']) results['files'] = files return results def object_audit_location_generator(self, policy, device_dirs=None, auditor_type="ALL"): """ Yield an AuditLocation for all objects stored under device_dirs. :param policy: the StoragePolicy instance :param device_dirs: directory of target device :param auditor_type: either ALL or ZBF """ datadir = get_data_dir(policy) return object_audit_location_generator(self.devices, datadir, self.mount_check, self.logger, device_dirs, auditor_type) def _hash_suffix_dir(self, path, policy): """ :param path: full path to directory :param policy: storage policy used """ if six.PY2: hashes = defaultdict(md5) else: class shim(object): def __init__(self): self.md5 = md5() def update(self, s): if isinstance(s, str): self.md5.update(s.encode('utf-8')) else: self.md5.update(s) def hexdigest(self): return self.md5.hexdigest() hashes = defaultdict(shim) try: path_contents = sorted(vfile.listdir(path)) except OSError as err: if err.errno in (errno.ENOTDIR, errno.ENOENT): raise PathNotDir() raise for hsh in path_contents: hsh_path = os.path.join(path, hsh) try: ondisk_info = self.cleanup_ondisk_files( hsh_path, policy=policy) except OSError as err: if err.errno == errno.ENOTDIR: partition_path = os.path.dirname(path) objects_path = os.path.dirname(partition_path) device_path = os.path.dirname(objects_path) # The made-up filename is so that the eventual dirpath() # will result in this object directory that we care about. # Some failures will result in an object directory # becoming a file, thus causing the parent directory to # be qarantined. quar_path = quarantine_vrenamer( device_path, os.path.join( hsh_path, "made-up-filename")) logging.exception( _('Quarantined %(hsh_path)s to %(quar_path)s because ' 'it is not a directory'), {'hsh_path': hsh_path, 'quar_path': quar_path}) continue raise if not ondisk_info['files']: continue # ondisk_info has info dicts containing timestamps for those # files that could determine the state of the diskfile if it were # to be opened. We update the suffix hash with the concatenation of # each file's timestamp and extension. The extension is added to # guarantee distinct hash values from two object dirs that have # different file types at the same timestamp(s). # # Files that may be in the object dir but would have no effect on # the state of the diskfile are not used to update the hash. for key in (k for k in ('meta_info', 'ts_info') if k in ondisk_info): info = ondisk_info[key] hashes[None].update(info['timestamp'].internal + info['ext']) # delegate to subclass for data file related updates... self._update_suffix_hashes(hashes, ondisk_info) if 'ctype_info' in ondisk_info: # We have a distinct content-type timestamp so update the # hash. As a precaution, append '_ctype' to differentiate this # value from any other timestamp value that might included in # the hash in future. There is no .ctype file so use _ctype to # avoid any confusion. info = ondisk_info['ctype_info'] hashes[None].update(info['ctype_timestamp'].internal + '_ctype') return hashes def _get_hashes(self, *args, **kwargs): hashed, hashes = self.__get_hashes(*args, **kwargs) hashes.pop('updated', None) hashes.pop('valid', None) return hashed, hashes def __get_hashes(self, device, partition, policy, recalculate=None, do_listdir=False): """ Get hashes for each suffix dir in a partition. do_listdir causes it to mistrust the hash cache for suffix existence at the (unexpectedly high) cost of a listdir. :param device: name of target device :param partition: partition on the device in which the object lives :param policy: the StoragePolicy instance :param recalculate: list of suffixes which should be recalculated when got :param do_listdir: force existence check for all hashes in the partition :returns: tuple of (number of suffix dirs hashed, dictionary of hashes) """ hashed = 0 dev_path = self.get_dev_path(device) partition_path = get_part_path(dev_path, policy, partition) hashes_file = os.path.join(partition_path, HASH_FILE) modified = False orig_hashes = {'valid': False} if recalculate is None: recalculate = [] try: orig_hashes = self.consolidate_hashes(partition_path) except Exception: self.logger.warning('Unable to read %r', hashes_file, exc_info=True) if not orig_hashes['valid']: # This is the only path to a valid hashes from invalid read (e.g. # does not exist, corrupt, etc.). Moreover, in order to write this # valid hashes we must read *the exact same* invalid state or we'll # trigger race detection. do_listdir = True hashes = {'valid': True} # If the exception handling around consolidate_hashes fired we're # going to do a full rehash regardless; but we need to avoid # needless recursion if the on-disk hashes.pkl is actually readable # (worst case is consolidate_hashes keeps raising exceptions and we # eventually run out of stack). # N.B. orig_hashes invalid only effects new parts and error/edge # conditions - so try not to get overly caught up trying to # optimize it out unless you manage to convince yourself there's a # bad behavior. orig_hashes = read_hashes(partition_path) else: hashes = copy.deepcopy(orig_hashes) if do_listdir: for suff in vfile.listdir(partition_path): if len(suff) == 3: hashes.setdefault(suff, None) modified = True hashes.update((suffix, None) for suffix in recalculate) for suffix, hash_ in list(hashes.items()): if not hash_: suffix_dir = os.path.join(partition_path, suffix) try: hashes[suffix] = self._hash_suffix( suffix_dir, policy=policy) hashed += 1 except PathNotDir: del hashes[suffix] except OSError: logging.exception(_('Error hashing suffix')) modified = True if modified: with lock_path(partition_path): if read_hashes(partition_path) == orig_hashes: write_hashes(partition_path, hashes) return hashed, hashes return self.__get_hashes(device, partition, policy, recalculate=recalculate, do_listdir=do_listdir) else: return hashed, hashes def get_diskfile_from_hash(self, device, partition, object_hash, policy, **kwargs): """ Returns a DiskFile instance for an object at the given object_hash. Just in case someone thinks of refactoring, be sure DiskFileDeleted is *not* raised, but the DiskFile instance representing the tombstoned object is returned instead. :param device: name of target device :param partition: partition on the device in which the object lives :param object_hash: the hash of an object path :param policy: the StoragePolicy instance :raises DiskFileNotExist: if the object does not exist :returns: an instance of BaseDiskFile """ dev_path = self.get_dev_path(device) if not dev_path: raise DiskFileDeviceUnavailable() object_path = join( dev_path, get_data_dir(policy), str(partition), object_hash[-3:], object_hash) try: filenames = self.cleanup_ondisk_files(object_path)['files'] except OSError as err: if err.errno == errno.ENOTDIR: quar_path = self.quarantine_renamer(dev_path, object_path) logging.exception( _('Quarantined %(object_path)s to %(quar_path)s because ' 'it is not a directory'), {'object_path': object_path, 'quar_path': quar_path}) raise DiskFileNotExist() if err.errno != errno.ENOENT: raise raise DiskFileNotExist() if not filenames: raise DiskFileNotExist() try: vf = vfile.VFileReader.get_vfile(join(object_path, filenames[-1]), self.logger) metadata = vf.metadata vf.close() except EOFError: raise DiskFileNotExist() try: account, container, obj = split_path( metadata.get('name', ''), 3, 3, True) except ValueError: raise DiskFileNotExist() return self.diskfile_cls(self, dev_path, partition, account, container, obj, policy=policy, **kwargs) def _listdir(self, path): """ :param path: full path to directory """ try: return vfile.listdir(path) except OSError as err: if err.errno != errno.ENOENT: self.logger.error( 'ERROR: Skipping %r due to error with listdir attempt: %s', path, err) return [] def exists(self, path): """ :param path: full path to directory """ return vfile.exists(path) def mkdirs(self, path): """ :param path: full path to directory """ return vfile.mkdirs(path) def listdir(self, path): """ :param path: full path to directory """ return vfile.listdir(path) def rmtree(self, path, ignore_errors=False): vfile.rmtree(path) def remove_file(self, path): """ similar to utils.remove_file :param path: full path to directory """ try: return vfile.delete_vfile_from_path(path) except (OSError, vfile.VFileException): pass def remove(self, path): """ :param path: full path to directory """ return vfile.delete_vfile_from_path(path) def isdir(self, path): """ :param path: full path to directory """ return vfile.isdir(path) def isfile(self, path): """ :param path: full path to directory """ return vfile.isfile(path) def rmdir(self, path): """ :param path: full path to directory """ pass class KVFileManager(BaseKVFileManager, DiskFileManager): diskfile_cls = KVFile policy_type = REPL_POLICY class ECKVFileReader(BaseKVFileReader, ECDiskFileReader): def __init__(self, vfr, data_file, obj_size, etag, disk_chunk_size, keep_cache_size, device_path, logger, quarantine_hook, use_splice, pipe_size, diskfile, keep_cache=False): super(ECKVFileReader, self).__init__( vfr, data_file, obj_size, etag, disk_chunk_size, keep_cache_size, device_path, logger, quarantine_hook, use_splice, pipe_size, diskfile, keep_cache) self.frag_buf = None self.frag_offset = 0 self.frag_size = self._diskfile.policy.fragment_size def _init_checks(self): super(ECKVFileReader, self)._init_checks() # for a multi-range GET this will be called at the start of each range; # only initialise the frag_buf for reads starting at 0. # TODO: reset frag buf to '' if tell() shows that start is on a frag # boundary so that we check frags selected by a range not starting at 0 # ECDOING - check _started_at_0 is defined correctly if self._started_at_0: self.frag_buf = '' else: self.frag_buf = None def _update_checks(self, chunk): # super(ECKVFileReader, self)._update_checks(chunk) # Because of python's MRO, this will call # ECDiskFileReader._update_checks, and blow up. # rather than mess with the class's mro() function, explicitely call # the one we want. BaseDiskFileReader._update_checks(self, chunk) if self.frag_buf is not None: self.frag_buf += chunk cursor = 0 while len(self.frag_buf) >= cursor + self.frag_size: self._check_frag(self.frag_buf[cursor:cursor + self.frag_size]) cursor += self.frag_size self.frag_offset += self.frag_size if cursor: self.frag_buf = self.frag_buf[cursor:] def _handle_close_quarantine(self): # super(ECKVFileReader, self)._handle_close_quarantine() BaseDiskFileReader._handle_close_quarantine(self) self._check_frag(self.frag_buf) class ECKVFileWriter(BaseKVFileWriter, ECDiskFileWriter): # TODO: this needs to be updated wrt. next_part_power, and other changes # in diskfile.py def _finalize_durable(self, data_file_path, durable_data_file_path, timestamp): exc = None try: try: vfile.rename_vfile(data_file_path, durable_data_file_path, self._diskfile._logger) except (OSError, IOError) as err: if err.errno == errno.ENOENT: files = vfile.listdir(self._datadir) results = self.manager.get_ondisk_files( files, self._datadir, frag_index=self._diskfile._frag_index, policy=self._diskfile.policy) # We "succeeded" if another writer cleaned up our data ts_info = results.get('ts_info') durables = results.get('durable_frag_set', []) if ts_info and ts_info['timestamp'] > timestamp: return elif any(frag_set['timestamp'] > timestamp for frag_set in durables): return if err.errno not in (errno.ENOSPC, errno.EDQUOT): # re-raise to catch all handler raise params = {'file': durable_data_file_path, 'err': err} self.manager.logger.exception( _('No space left on device for %(file)s (%(err)s)'), params) exc = DiskFileNoSpace( 'No space left on device for %(file)s (%(err)s)' % params) else: try: self.manager.cleanup_ondisk_files(self._datadir)['files'] except OSError as os_err: self.manager.logger.exception( _('Problem cleaning up %(datadir)s (%(err)s)'), {'datadir': self._datadir, 'err': os_err}) except Exception as err: params = {'file': durable_data_file_path, 'err': err} self.manager.logger.exception( _('Problem making data file durable %(file)s (%(err)s)'), params) exc = DiskFileError( 'Problem making data file durable %(file)s (%(err)s)' % params) if exc: raise exc def put(self, metadata): """ The only difference between this method and the replication policy DiskFileWriter method is adding the frag index to the metadata. :param metadata: dictionary of metadata to be associated with object """ fi = None cleanup = True if self._extension == '.data': # generally we treat the fragment index provided in metadata as # canon, but if it's unavailable (e.g. tests) it's reasonable to # use the frag_index provided at instantiation. Either way make # sure that the fragment index is included in object sysmeta. fi = metadata.setdefault('X-Object-Sysmeta-Ec-Frag-Index', self._diskfile._frag_index) fi = self.manager.validate_fragment_index(fi) self._diskfile._frag_index = fi # defer cleanup until commit() writes makes diskfile durable cleanup = False super(ECKVFileWriter, self)._put(metadata, cleanup, frag_index=fi) class ECKVFile(BaseKVFile, ECDiskFile): reader_cls = ECKVFileReader writer_cls = ECKVFileWriter def purge(self, timestamp, frag_index): """ Remove a tombstone file matching the specified timestamp or datafile matching the specified timestamp and fragment index from the object directory. This provides the EC reconstructor/ssync process with a way to remove a tombstone or fragment from a handoff node after reverting it to its primary node. The hash will be invalidated, and if empty or invalid the hsh_path will be removed on next cleanup_ondisk_files. :param timestamp: the object timestamp, an instance of :class:`~swift.common.utils.Timestamp` :param frag_index: fragment archive index, must be a whole number or None. """ purge_file = self.manager.make_on_disk_filename( timestamp, ext='.ts') remove_vfile(os.path.join(self._datadir, purge_file)) if frag_index is not None: # data file may or may not be durable so try removing both filename # possibilities purge_file = self.manager.make_on_disk_filename( timestamp, ext='.data', frag_index=frag_index) remove_vfile(os.path.join(self._datadir, purge_file)) purge_file = self.manager.make_on_disk_filename( timestamp, ext='.data', frag_index=frag_index, durable=True) remove_vfile(os.path.join(self._datadir, purge_file)) # we don't use hashes.pkl files # self.manager.invalidate_hash(dirname(self._datadir)) class ECKVFileManager(BaseKVFileManager, ECDiskFileManager): diskfile_cls = ECKVFile policy_type = EC_POLICY def remove_vfile(filepath): try: vfile.delete_vfile_from_path(filepath) except OSError: pass