OpenStack使用Ceph存储，Ceph到底做了什么?

@capabilities.check
def add(self, image_id, image_file, image_size, context=None,
        verifier=None):
    checksum = hashlib.md5()
    image_name = str(image_id)
    with self.get_connection(conffile=self.conf_file,
                             rados_id=self.user) as conn:
        fsid = None
        if hasattr(conn, 'get_fsid'):
            fsid = conn.get_fsid()
        with conn.open_ioctx(self.pool) as ioctx:
            order = int(math.log(self.WRITE_CHUNKSIZE, 2))
            try:
                loc = self._create_image(fsid, conn, ioctx, image_name,
                                         image_size, order)
            except rbd.ImageExists:
                msg = _('RBD image %s already exists') % image_id
                raise exceptions.Duplicate(message=msg)
                ...

def _create_image(self, fsid, conn, ioctx, image_name,
                  size, order, context=None):
    librbd = rbd.RBD()
    features = conn.conf_get('rbd_default_features')
    librbd.create(ioctx, image_name, size, order, old_format=False,
                  features=int(features))
    return StoreLocation({
        'fsid': fsid,
        'pool': self.pool,
        'image': image_name,
        'snapshot': DEFAULT_SNAPNAME,
    }, self.conf)

with rbd.Image(ioctx, image_name) as image:
    bytes_written = 0
    offset = 0
    chunks = utils.chunkreadable(image_file,
                                 self.WRITE_CHUNKSIZE)
    for chunk in chunks:
        offset += image.write(chunk, offset)
        checksum.update(chunk)

if loc.snapshot:
    image.create_snap(loc.snapshot)
    image.protect_snap(loc.snapshot)

try:
    self._unprotect_snapshot(image, snapshot_name)
    image.remove_snap(snapshot_name)
except rbd.ImageBusy as exc:
    raise exceptions.InUseByStore()
rbd.RBD().remove(ioctx, image_name)

def spawn(self, context, instance, image_meta, injected_files,
          admin_password, network_info=None, block_device_info=None):
    ...
    self._create_image(context, instance, disk_info['mapping'],
                       injection_info=injection_info,
                       block_device_info=block_device_info)
    ...

def _create_image(self, context, instance,
                  disk_mapping, injection_info=None, suffix='',
                  disk_images=None, block_device_info=None,
                  fallback_from_host=None,
                  ignore_bdi_for_swap=False):
    booted_from_volume = self._is_booted_from_volume(block_device_info)
    ...
    # ensure directories exist and are writable
    fileutils.ensure_tree(libvirt_utils.get_instance_path(instance))
    ...
    self._create_and_inject_local_root(context, instance,
                                       booted_from_volume, suffix,
                                       disk_images, injection_info,
                                       fallback_from_host)
    ...

def _create_and_inject_local_root(self, context, instance,
                                  booted_from_volume, suffix, disk_images,
                                  injection_info, fallback_from_host):
    ...
    if not booted_from_volume:
        root_fname = imagecache.get_cache_fname(disk_images['image_id'])
        size = instance.flavor.root_gb * units.Gi
        backend = self.image_backend.by_name(instance, 'disk' + suffix,
                                             CONF.libvirt.images_type)
        if backend.SUPPORTS_CLONE:
            def clone_fallback_to_fetch(*args, **kwargs):
                try:
                    backend.clone(context, disk_images['image_id'])
                except exception.ImageUnacceptable:
                    libvirt_utils.fetch_image(*args, **kwargs)
            fetch_func = clone_fallback_to_fetch
        else:
            fetch_func = libvirt_utils.fetch_image
        self._try_fetch_image_cache(backend, fetch_func, context,
                                    root_fname, disk_images['image_id'],
                                    instance, size, fallback_from_host)
        ...

def clone(self, context, image_id_or_uri):
    ...
    for location in locations:
        if self.driver.is_cloneable(location, image_meta):
            LOG.debug('Selected location: %(loc)s', {'loc': location})
            return self.driver.clone(location, self.rbd_name)
    ...

 def is_cloneable(self, image_location, image_meta):
        url = image_location['url']
        try:
            fsid, pool, image, snapshot = self.parse_url(url)
        except exception.ImageUnacceptable as e:
            return False
        if self.get_fsid() != fsid:
            return False
        if image_meta.get('disk_format') != 'raw':
            return False
        # check that we can read the image
        try:
            return self.exists(image, pool=pool, snapshot=snapshot)
        except rbd.Error as e:
            LOG.debug('Unable to open image %(loc)s: %(err)s',
                      dict(loc=url, err=e))
            return False

def clone(self, image_location, dest_name, dest_pool=None):
    _fsid, pool, image, snapshot = self.parse_url(
            image_location['url'])
    with RADOSClient(self, str(pool)) as src_client:
        with RADOSClient(self, dest_pool) as dest_client:
            try:
                RbdProxy().clone(src_client.ioctx,
                                 image,
                                 snapshot,
                                 dest_client.ioctx,
                                 str(dest_name),
                                 features=src_client.features)
            except rbd.PermissionError:
                raise exception.Forbidden(_('no write permission on '
                                            'storage pool %s') % dest_pool)

def snapshot(self, context, instance, image_id, update_task_state):
    ...
    root_disk = self.image_backend.by_libvirt_path(
        instance, disk_path, image_type=source_type)
    try:
        update_task_state(task_state=task_states.IMAGE_UPLOADING,
                          expected_state=task_states.IMAGE_PENDING_UPLOAD)
        metadata['location'] = root_disk.direct_snapshot(
            context, snapshot_name, image_format, image_id,
            instance.image_ref)
        self._snapshot_domain(context, live_snapshot, virt_dom, state,
                              instance)
        self._image_api.update(context, image_id, metadata,
                               purge_props=False)
    except (NotImplementedError, exception.ImageUnacceptable) as e:
        ...

def direct_snapshot(self, context, snapshot_name, image_format,
                    image_id, base_image_id):
    fsid = self.driver.get_fsid()
    parent_pool = self._get_parent_pool(context, base_image_id, fsid)

    self.driver.create_snap(self.rbd_name, snapshot_name, protect=True)
    location = {'url': 'rbd://%(fsid)s/%(pool)s/%(image)s/%(snap)s' %
                       dict(fsid=fsid,
                            pool=self.pool,
                            image=self.rbd_name,
                            snap=snapshot_name)}
    try:
        self.driver.clone(location, image_id, dest_pool=parent_pool)
        self.driver.flatten(image_id, pool=parent_pool)
    finally:
        self.cleanup_direct_snapshot(location)
    self.driver.create_snap(image_id, 'snap', pool=parent_pool,
                            protect=True)
    return ('rbd://%(fsid)s/%(pool)s/%(image)s/snap' %
            dict(fsid=fsid, pool=parent_pool, image=image_id))

def destroy(self, context, instance, network_info, block_device_info=None,
                destroy_disks=True):
    self._destroy(instance)
    self.cleanup(context, instance, network_info, block_device_info,
                 destroy_disks)

...
if destroy_disks:
    # NOTE(haomai): destroy volumes if needed
    if CONF.libvirt.images_type == 'lvm':
        self._cleanup_lvm(instance, block_device_info)
    if CONF.libvirt.images_type == 'rbd':
        self._cleanup_rbd(instance)
...

def _cleanup_rbd(self, instance):
    if instance.task_state == task_states.RESIZE_REVERTING:
        filter_fn = lambda disk: (disk.startswith(instance.uuid) and
                                  disk.endswith('disk.local'))
    else:
        filter_fn = lambda disk: disk.startswith(instance.uuid)
    LibvirtDriver._get_rbd_driver().cleanup_volumes(filter_fn)

def cleanup_volumes(self, filter_fn):
    with RADOSClient(self, self.pool) as client:
        volumes = RbdProxy().list(client.ioctx)
        for volume in filter(filter_fn, volumes):
            self._destroy_volume(client, volume)

def _destroy_volume(self, client, volume, pool=None):
    """Destroy an RBD volume, retrying as needed.
    """
    def _cleanup_vol(ioctx, volume, retryctx):
        try:
            RbdProxy().remove(ioctx, volume)
            raise loopingcall.LoopingCallDone(retvalue=False)
        except rbd.ImageHasSnapshots:
            self.remove_snap(volume, libvirt_utils.RESIZE_SNAPSHOT_NAME,
                             ignore_errors=True)
        except (rbd.ImageBusy, rbd.ImageHasSnapshots):
            LOG.warning('rbd remove %(volume)s in pool %(pool)s failed',
                        {'volume': volume, 'pool': self.pool})
        retryctx['retries'] -= 1
        if retryctx['retries'] <= 0:
            raise loopingcall.LoopingCallDone()

    # NOTE(danms): We let it go for ten seconds
    retryctx = {'retries': 10}
    timer = loopingcall.FixedIntervalLoopingCall(
        _cleanup_vol, client.ioctx, volume, retryctx)
    timed_out = timer.start(interval=1).wait()
    if timed_out:
        # NOTE(danms): Run this again to propagate the error, but
        # if it succeeds, don't raise the loopingcall exception
        try:
            _cleanup_vol(client.ioctx, volume, retryctx)
        except loopingcall.LoopingCallDone:
            pass

def create_volume(self, context, volume, request_spec=None,
                  filter_properties=None, allow_reschedule=True):
    ...              
    try:
        # NOTE(flaper87): Driver initialization is
        # verified by the task itself.
        flow_engine = create_volume.get_flow(
            context_elevated,
            self,
            self.db,
            self.driver,
            self.scheduler_rpcapi,
            self.host,
            volume,
            allow_reschedule,
            context,
            request_spec,
            filter_properties,
            image_volume_cache=self.image_volume_cache,
        )
    except Exception:
        msg = _("Create manager volume flow failed.")
        LOG.exception(msg, resource={'type': 'volume', 'id': volume.id})
        raise exception.CinderException(msg)
...        

def execute(self, context, volume, volume_spec):
    ...
    if create_type == 'raw':
        model_update = self._create_raw_volume(volume, **volume_spec)
    elif create_type == 'snap':
        model_update = self._create_from_snapshot(context, volume,
                                                  **volume_spec)
    elif create_type == 'source_vol':
        model_update = self._create_from_source_volume(
            context, volume, **volume_spec)
    elif create_type == 'image':
        model_update = self._create_from_image(context,
                                               volume,
                                               **volume_spec)
    else:
        raise exception.VolumeTypeNotFound(volume_type_id=create_type)
    ...    

def _create_raw_volume(self, volume, **kwargs):
    ret = self.driver.create_volume(volume)
    ...

def create_volume(self, volume):
    with RADOSClient(self) as client:
        self.RBDProxy().create(client.ioctx,
                               vol_name,
                               size,
                               order,
                               old_format=False,
                               features=client.features)

        try:
            volume_update = self._enable_replication_if_needed(volume)
        except Exception:
            self.RBDProxy().remove(client.ioctx, vol_name)
            err_msg = (_('Failed to enable image replication'))
            raise exception.ReplicationError(reason=err_msg,
                                             volume_id=volume.id)

def create_volume_from_snapshot(self, volume, snapshot):
    """Creates a volume from a snapshot."""
    volume_update = self._clone(volume, self.configuration.rbd_pool,
                                snapshot.volume_name, snapshot.name)
    if self.configuration.rbd_flatten_volume_from_snapshot:
        self._flatten(self.configuration.rbd_pool, volume.name)
    if int(volume.size):
        self._resize(volume)
    return volume_update

def _create_from_source_volume(self, context, volume, source_volid,
                               **kwargs):
    # NOTE(harlowja): if the source volume has disappeared this will be our
    # detection of that since this database call should fail.
    #
    # NOTE(harlowja): likely this is not the best place for this to happen
    # and we should have proper locks on the source volume while actions
    # that use the source volume are underway.
    srcvol_ref = objects.Volume.get_by_id(context, source_volid)
    model_update = self.driver.create_cloned_volume(volume, srcvol_ref)

# Do full copy if requested
if self.configuration.rbd_max_clone_depth <= 0:
    with RBDVolumeProxy(self, src_name, read_only=True) as vol:
        vol.copy(vol.ioctx, dest_name)
        self._extend_if_required(volume, src_vref)
    return

# Otherwise do COW clone.
with RADOSClient(self) as client:
    src_volume = self.rbd.Image(client.ioctx, src_name)
    LOG.debug("creating snapshot='%s'", clone_snap)
    try:
        # Create new snapshot of source volume
        src_volume.create_snap(clone_snap)
        src_volume.protect_snap(clone_snap)
        # Now clone source volume snapshot
        LOG.debug("cloning '%(src_vol)s@%(src_snap)s' to "
                  "'%(dest)s'",
                  {'src_vol': src_name, 'src_snap': clone_snap,
                   'dest': dest_name})
        self.RBDProxy().clone(client.ioctx, src_name, clone_snap,
                              client.ioctx, dest_name,
                              features=client.features)

depth = self._get_clone_depth(client, src_name)
if depth >= self.configuration.rbd_max_clone_depth:
        dest_volume = self.rbd.Image(client.ioctx, dest_name)
        try:
            dest_volume.flatten()
        except Exception as e:
            ...

        try:
            src_volume.unprotect_snap(clone_snap)
            src_volume.remove_snap(clone_snap)
        except Exception as e:
            ...           

def _create_from_image(self, context, volume,
                       image_location, image_id, image_meta,
                       image_service, **kwargs):
    ...
    model_update, cloned = self.driver.clone_image(
        context,
        volume,
        image_location,
        image_meta,
        image_service)
   ...

def clone_image(self, context, volume,
                image_location, image_meta,
                image_service):
    # iterate all locations to look for a cloneable one.
    for url_location in url_locations:
        if url_location and self._is_cloneable(
                url_location, image_meta):
            _prefix, pool, image, snapshot = \
                self._parse_location(url_location)
            volume_update = self._clone(volume, pool, image, snapshot)
            volume_update['provider_location'] = None
            self._resize(volume)
            return volume_update, True
    return ({}, False)

...
try:
    utils.require_driver_initialized(self.driver)
    snapshot.context = context
    model_update = self.driver.create_snapshot(snapshot)
    ...
except Exception:
    ...

def create_snapshot(self, snapshot):
    """Creates an rbd snapshot."""
    with RBDVolumeProxy(self, snapshot.volume_name) as volume:
        snap = utils.convert_str(snapshot.name)
        volume.create_snap(snap)
        volume.protect_snap(snap)

if self._file_is_rbd(volume_file):
    # If volume an RBD, attempt incremental backup.
    LOG.debug("Volume file is RBD: attempting incremental backup.")
    try:
        updates = self._backup_rbd(backup, volume_file,
                                   volume.name, length)
    except exception.BackupRBDOperationFailed:
        LOG.debug("Forcing full backup of volume %s.", volume.id)
        do_full_backup = True

from_snap = self._get_most_recent_snap(source_rbd_image)
base_name = self._get_backup_base_name(volume_id, diff_format=True)
image_created = False
with rbd_driver.RADOSClient(self, backup.container) as client:
    if base_name not in self.rbd.RBD().list(ioctx=client.ioctx):
        ...
        # Create new base image
        self._create_base_image(base_name, length, client)
        image_created = True
    else:
        ...

def _create_base_image(self, name, size, rados_client):
    old_format, features = self._get_rbd_support()
    self.rbd.RBD().create(ioctx=rados_client.ioctx,
                          name=name,
                          size=size,
                          old_format=old_format,
                          features=features,
                          stripe_unit=self.rbd_stripe_unit,
                          stripe_count=self.rbd_stripe_count)

def _backup_rbd(self, backup, volume_file, volume_name, length):
    ...
    new_snap = self._get_new_snap_name(backup.id)
    LOG.debug("Creating backup snapshot='%s'", new_snap)
    source_rbd_image.create_snap(new_snap)

    try:
        self._rbd_diff_transfer(volume_name, rbd_pool, base_name,
                                backup.container,
                                src_user=rbd_user,
                                src_conf=rbd_conf,
                                dest_user=self._ceph_backup_user,
                                dest_conf=self._ceph_backup_conf,
                                src_snap=new_snap,
                                from_snap=from_snap)
                            
def _get_new_snap_name(self, backup_id):
    return utils.convert_str("backup.%s.snap.%s"
                             % (backup_id, time.time()))

def _rbd_diff_transfer(self, src_name, src_pool, dest_name, dest_pool,
                       src_user, src_conf, dest_user, dest_conf,
                       src_snap=None, from_snap=None):
    src_ceph_args = self._ceph_args(src_user, src_conf, pool=src_pool)
    dest_ceph_args = self._ceph_args(dest_user, dest_conf, pool=dest_pool)

    cmd1 = ['rbd', 'export-diff'] + src_ceph_args
    if from_snap is not None:
        cmd1.extend(['--from-snap', from_snap])
    if src_snap:
        path = utils.convert_str("%s/%s@%s"
                                 % (src_pool, src_name, src_snap))
    else:
        path = utils.convert_str("%s/%s" % (src_pool, src_name))
    cmd1.extend([path, '-'])

    cmd2 = ['rbd', 'import-diff'] + dest_ceph_args
    rbd_path = utils.convert_str("%s/%s" % (dest_pool, dest_name))
    cmd2.extend(['-', rbd_path])

    ret, stderr = self._piped_execute(cmd1, cmd2)
    if ret:
        msg = (_("RBD diff op failed - (ret=%(ret)s stderr=%(stderr)s)") %
               {'ret': ret, 'stderr': stderr})
        LOG.info(msg)
        raise exception.BackupRBDOperationFailed(msg)

from_snap = self._get_most_recent_snap(source_rbd_image)
with rbd_driver.RADOSClient(self, backup.container) as client:
    if base_name not in self.rbd.RBD().list(ioctx=client.ioctx):
        ...
    else:
        if not self._snap_exists(base_name, from_snap, client):
            errmsg = (_("Snapshot='%(snap)s' does not exist in base "
                        "image='%(base)s' - aborting incremental "
                        "backup") %
                      {'snap': from_snap, 'base': base_name})
            LOG.info(errmsg)
            raise exception.BackupRBDOperationFailed(errmsg)

new_snap = self._get_new_snap_name(backup.id)
source_rbd_image.create_snap(new_snap)

try:
    before = time.time()
    self._rbd_diff_transfer(volume_name, rbd_pool, base_name,
                            backup.container,
                            src_user=rbd_user,
                            src_conf=rbd_conf,
                            dest_user=self._ceph_backup_user,
                            dest_conf=self._ceph_backup_conf,
                            src_snap=new_snap,
                            from_snap=from_snap)
    if from_snap:
        source_rbd_image.remove_snap(from_snap)

def _restore_volume(self, backup, volume, volume_file):
    length = int(volume.size) * units.Gi

    base_name = self._get_backup_base_name(backup.volume_id,
                                           diff_format=True)
    with rbd_driver.RADOSClient(self, backup.container) as client:
        diff_allowed, restore_point = \
            self._diff_restore_allowed(base_name, backup, volume,
                                       volume_file, client)

def _diff_restore_allowed(self, base_name, backup, volume, volume_file,
                          rados_client):
    rbd_exists, base_name = self._rbd_image_exists(base_name,
                                                   backup.volume_id,
                                                   rados_client)
    if not rbd_exists:
        return False, None
    restore_point = self._get_restore_point(base_name, backup.id)
    if restore_point:
        if self._file_is_rbd(volume_file):
            if volume.id == backup.volume_id:
                return False, restore_point
            if self._rbd_has_extents(volume_file.rbd_image):
                return False, restore_point
            return True, restore_point

def _diff_restore_rbd(self, backup, restore_file, restore_name,
                      restore_point, restore_length):
    rbd_user = restore_file.rbd_user
    rbd_pool = restore_file.rbd_pool
    rbd_conf = restore_file.rbd_conf
    base_name = self._get_backup_base_name(backup.volume_id,
                                           diff_format=True)
    before = time.time()
    try:
        self._rbd_diff_transfer(base_name, backup.container,
                                restore_name, rbd_pool,
                                src_user=self._ceph_backup_user,
                                src_conf=self._ceph_backup_conf,
                                dest_user=rbd_user, dest_conf=rbd_conf,
                                src_snap=restore_point)
    except exception.BackupRBDOperationFailed:
        raise
    self._check_restore_vol_size(backup, restore_name, restore_length,
                                 rbd_pool)

def _transfer_data(self, src, src_name, dest, dest_name, length):
    chunks = int(length / self.chunk_size)
    for chunk in range(0, chunks):
        before = time.time()
        data = src.read(self.chunk_size)
        dest.write(data)
        dest.flush()
        delta = (time.time() - before)
        rate = (self.chunk_size / delta) / 1024
        # yield to any other pending backups
        eventlet.sleep(0)
    rem = int(length % self.chunk_size)
    if rem:
        dest.write(data)
        dest.flush()
        # yield to any other pending backups
        eventlet.sleep(0)