diff --git a/drivers/LVMSR.py b/drivers/LVMSR.py index fc72defc9..aa6cf2996 100755 --- a/drivers/LVMSR.py +++ b/drivers/LVMSR.py @@ -17,8 +17,10 @@ # # LVMSR: VHD and QCOW2 on LVM storage repository # +import contextlib +from contextlib import contextmanager -from sm_typing import Dict, List, override +from sm_typing import Dict, List, override, Optional, Tuple, Union, Collection, Any import SR from SR import deviceCheck @@ -37,10 +39,11 @@ import cleanup import blktap2 from journaler import Journaler +from jutils import BaseLogEntry from refcounter import RefCounter from ipc import IPCFlag from constants import NS_PREFIX_LVM, VG_LOCATION, VG_PREFIX, CBT_BLOCK_SIZE -from cowutil import CowUtil, getCowUtil, getImageStringFromVdiType, getVdiTypeFromImageFormat +from cowutil import CowUtil, getCowUtil, getImageStringFromVdiType from lvmcowutil import LV_PREFIX, LvmCowUtil from lvmanager import LVActivator from vditype import VdiType @@ -51,7 +54,6 @@ READ_ONLY_TAG, MANAGED_TAG, SNAPSHOT_TIME_TAG, METADATA_OF_POOL_TAG, \ LVMMetadataHandler, METADATA_OBJECT_TYPE_VDI, \ METADATA_OBJECT_TYPE_SR, METADATA_UPDATE_OBJECT_TYPE_TAG -from metadata import retrieveXMLfromFile, _parseXML from xmlrpc.client import DateTime import glob from constants import CBTLOG_TAG @@ -63,7 +65,7 @@ "VDI_CREATE", "VDI_DELETE", "VDI_ATTACH", "VDI_DETACH", "VDI_MIRROR", "VDI_CLONE", "VDI_SNAPSHOT", "VDI_RESIZE", "ATOMIC_PAUSE", "VDI_RESET_ON_BOOT/2", "VDI_UPDATE", "VDI_CONFIG_CBT", - "VDI_ACTIVATE", "VDI_DEACTIVATE"] + "VDI_ACTIVATE", "VDI_DEACTIVATE", "VDI_REVERT"] CONFIGURATION = [['device', 'local device path (required) (e.g. /dev/sda3)']] @@ -82,11 +84,123 @@ OPS_EXCLUSIVE = [ "sr_create", "sr_delete", "sr_attach", "sr_detach", "sr_scan", "sr_update", "vdi_create", "vdi_delete", "vdi_resize", "vdi_snapshot", - "vdi_clone"] + "vdi_clone", "vdi_revert"] # Log if snapshot pauses VM for more than this many seconds LONG_SNAPTIME = 60 +class RevertLogDestinationVDI: + def __init__( + self, + uuid: str, + lvname: str, + backup_lvname: str, + *, + is_cbt_enabled: bool, + ): + self.uuid = uuid + self.lvname = lvname + self.backup_lvname = backup_lvname + self.is_cbt_enabled = is_cbt_enabled + + def to_dict(self) -> Dict[str, Union[str, bool]]: + return { + "uuid": self.uuid, + "lvname": self.lvname, + "backup_lvname": self.backup_lvname, + "is_cbt_enabled": self.is_cbt_enabled, + } + + @classmethod + def from_dict(cls, data: Dict[str, Union[str, bool]]) -> "RevertLogDestinationVDI": + return cls( + uuid=str(data["uuid"]), + lvname=str(data["lvname"]), + backup_lvname=str(data["backup_lvname"]), + is_cbt_enabled=bool(data["is_cbt_enabled"]), + ) + + +class RevertLogSourceVDI: + def __init__( + self, + uuid: str, + lvname: str, + *, + is_cbt_enabled: bool, + ): + self.uuid = uuid + self.lvname = lvname + self.is_cbt_enabled = is_cbt_enabled + + def to_dict(self) -> Dict[str, Union[str, bool]]: + return { + "uuid": self.uuid, + "lvname": self.lvname, + "is_cbt_enabled": self.is_cbt_enabled, + } + + @classmethod + def from_dict(cls, data: Dict[str, Union[str, bool]]) -> "RevertLogSourceVDI": + return cls( + uuid=str(data["uuid"]), + lvname=str(data["lvname"]), + is_cbt_enabled=bool(data["is_cbt_enabled"]), + ) + + +class RevertLogInsertedVDI: + def __init__(self, uuid: str, lvname: str): + self.uuid = uuid + self.lvname = lvname + + def to_dict(self) -> Dict[str, Collection[str]]: + return { + "uuid": self.uuid, + "lvname": self.lvname, + } + + @classmethod + def from_dict(cls, data: Dict[str, str]) -> "RevertLogInsertedVDI": + return cls( + uuid=data["uuid"], + lvname=data["lvname"], + ) + + +class RevertLogEntry(BaseLogEntry): + """Journal entry used to rollback failed revert operations""" + + CURRENT_VERSION = "1.0" + JRN_KEY = Journaler.JRN_REVERT + + def __init__( + self, + dest: RevertLogDestinationVDI, + src: RevertLogSourceVDI, + inserted: RevertLogInsertedVDI, + ): + self.dest = dest + self.src = src + self.inserted = inserted + + @override + def to_dict(self) -> Dict[str, Collection[str]]: + return { + "dest": self.dest.to_dict(), + "src": self.src.to_dict(), + "inserted": self.inserted.to_dict(), + } + + @override + @classmethod + def from_dict(cls, data: Dict[str, Union[Dict[str, str], str]]) -> "RevertLogEntry": + return cls( + dest=RevertLogDestinationVDI.from_dict(data["dest"]), # type: ignore # Version checked + src=RevertLogSourceVDI.from_dict(data["src"]), # type: ignore # Version checked + inserted=RevertLogInsertedVDI.from_dict(data["inserted"]) #type: ignore # Version checked + ) + class LVMSR(SR.SR): DRIVER_TYPE = 'lvhd' @@ -130,6 +244,12 @@ class LVMSR(SR.SR): TEST_MODE_VHD_FAIL_RESIZE_END: "VHD_UTIL_TEST_FAIL_RESIZE_END" } + + # Journals that should prevent VMs from booting while pending + _CRITICAL_JOURNALS = [ + RevertLogEntry.JRN_KEY, + ] + testMode = "" legacyMode = True @@ -209,8 +329,22 @@ def load(self, sr_uuid) -> None: self.legacyMode = False if lvutil._checkVG(self.vgname): - if self.isMaster and not self.cmd in ["vdi_attach", "vdi_detach", - "vdi_activate", "vdi_deactivate"]: + # Disable SR on slaves when mounting VDIs if + # a critical journal is pending + # and force journal undo if on master + if self._has_critical_journals(): + if not self.isMaster: + raise xs_errors.XenError( + "SRUnavailable", opterr="Critical journals are pending. A scan is required." + ) + self._undoAllJournals() + + if self.isMaster and not self.cmd in [ + "vdi_attach", + "vdi_detach", + "vdi_activate", + "vdi_deactivate", + ]: self._undoAllJournals() if not self.cmd in ["sr_attach", "sr_probe"]: self._checkMetadataVolume() @@ -234,10 +368,8 @@ def load(self, sr_uuid) -> None: break # check if metadata volume exists - try: + with contextlib.suppress(Exception): self.mdexists = self.lvmCache.checkLV(self.MDVOLUME_NAME) - except: - pass @override def cleanup(self) -> None: @@ -359,7 +491,7 @@ def syncMetadataAndXapi(self): .updateMetadata(update_map) except Exception as e: raise xs_errors.XenError('MetadataError', \ - opterr='Error synching SR Metadata and XAPI: %s' % str(e)) + opterr='Error syncing SR Metadata and XAPI: %s' % str(e)) def _checkMetadataVolume(self): util.SMlog("Entering _checkMetadataVolume") @@ -384,7 +516,7 @@ def _checkMetadataVolume(self): self.legacyMode = False def _synchSmConfigWithMetaData(self): - util.SMlog("Synching sm-config with metadata volume") + util.SMlog("Syncing sm-config with metadata volume") try: # get SR info from metadata @@ -908,6 +1040,12 @@ def _loadvdis(self): util.SMlog("Scan found hidden leaf (%s), ignoring" % uuid) del self.vdis[uuid] + def _has_critical_journals(self) -> bool: + for key in self._CRITICAL_JOURNALS: + if any(self.journaler.getAll(key)): + return True + return False + def _ensureSpaceAvailable(self, amount_needed): space_available = lvutil._getVGstats(self.vgname)['freespace'] if (space_available < amount_needed): @@ -1174,12 +1312,98 @@ def _undoAllJournals(self): try: self._undoAllInflateJournals() self._undoAllCowJournals() + self._undo_revert_journals() self._handleInterruptedCloneOps() self._handleInterruptedCoalesceLeaf() finally: self.lock.release() self.cleanup() + def _undo_revert_journals(self): + for journal_id, content in self.journaler.getAll(RevertLogEntry.JRN_KEY).items(): + entry = RevertLogEntry.from_journal(journal_id, content) + util.SMlog(f"Reverting {entry.dest.uuid}") + + self._rollback_revert_vdi(entry) + + self.journaler.remove(RevertLogEntry.JRN_KEY, journal_id) + + def _rollback_revert_vdi(self, entry: RevertLogEntry): + util.SMlog( + f"Reverting vdi {entry.dest.uuid} from backup {entry.dest.backup_lvname}" + ) + if not self.lvmCache.checkLV(entry.dest.backup_lvname): + util.SMlog(f"Could not find backup {entry.dest.backup_lvname}, skipping.") + return + + util.SMlog(f"Restoring src vdi {entry.src.uuid}") + if not self.lvmCache.checkLV(entry.src.lvname): + util.SMlog(f"Restoring src vdi {entry.src.uuid} from {entry.inserted.uuid}") + self.lvmCache.rename(entry.inserted.lvname, entry.src.lvname) + + if self.lvmCache.getLVInfo(entry.src.lvname)[entry.src.lvname].readonly: + self.lvmCache.setReadonly(entry.src.lvname, False) + + util.SMlog(f"Restoring {entry.src.uuid}") + src: "LVMVDI" = self.vdi(entry.src.uuid) # type: ignore[assignment] + src_ref = src.session.xenapi.VDI.get_by_uuid(src.uuid) + src.sm_config = src.session.xenapi.VDI.get_sm_config(src_ref) + src._loadThis() + with src.activated(False): + src.cowutil.setHidden(str(src.path), False) + src.update_from_disk() + src._db_update() + src.disable_leaf_on_secondary(src.uuid) + + # CBT cannot be added by accident on src, we can skip checking if it exists + util.SMlog(f"CBT rollback: restoring cbtlog for src {entry.src.uuid}") + if entry.src.is_cbt_enabled: + src._reset_cbt_log() + src.session.xenapi.VDI.set_cbt_enabled(src_ref, True) + + self.vdis[src.uuid] = src + + # Remove any existing vdi to allow renaming the backup with this name + if self.lvmCache.checkLV(entry.dest.lvname): + self.lvmCache.remove(entry.dest.lvname) + + # Ensure that inserted node is hidden so that it doesn't leak + if self.lvmCache.checkLV(entry.inserted.lvname): + inserted: "LVMVDI" = self.vdi(entry.inserted.uuid) # type: ignore[assignment] + with inserted.activated(False): + inserted.cowutil.setHidden(str(inserted.path), True) + inserted.update_from_disk() + inserted_ref = inserted._db_update_or_introduce() + + inserted.session.xenapi.VDI.set_managed(inserted_ref, False) + inserted.disable_leaf_on_secondary(inserted.uuid) + + # Once we move the backup to it's old name, the journal can't be run again + # If something fails from now, we might lose some info, most of them + # will be back from a simple scan + # Worst case scenario, CBT is disabled or incoherent (which will be solved by tapdisk) + self.lvmCache.rename(entry.dest.backup_lvname, entry.dest.lvname) + + dest: "LVMVDI" = self.vdi(entry.dest.uuid) # type: ignore[assignment] + dest_ref = dest.session.xenapi.VDI.get_by_uuid(dest.uuid) + dest.sm_config = dest.session.xenapi.VDI.get_sm_config(dest_ref) + + with dest.activated(False): + dest.update_from_disk() + dest._db_update() + + self.vdis[dest.uuid] = dest + + # CBT might have been added when not needed or removed when needed + util.SMlog(f"CBT rollback: restoring cbtlog for dest {entry.dest.uuid}") + if entry.dest.is_cbt_enabled: + dest._reset_cbt_log() + dest.session.xenapi.VDI.set_cbt_enabled(dest_ref, True) + else: + dest._delete_cbt_log() + dest.session.xenapi.VDI.set_cbt_enabled(dest_ref, False) + + def _undoAllInflateJournals(self): entries = self.journaler.getAll(LvmCowUtil.JOURNAL_INFLATE) if len(entries) == 0: @@ -1354,7 +1578,6 @@ def load(self, vdi_uuid) -> None: else: self.sm_config_override['vhd-parent'] = None return - # scan() didn't run: determine the type of the VDI manually if self._determineType(): return @@ -1700,6 +1923,186 @@ def _detach(self): self._chainSetActive(False, True) self.attached = False + @override + def _do_revert( + self, + dest: "LVMVDI", # type: ignore # self and dest are the same type + src_cbtlog: Optional[str], + dest_cbtlog: Optional[str], + ): + # Sanity checks + if not self.sr.isMaster: + raise xs_errors.XenError('LVMMaster') + if self.sr.legacyMode: + raise xs_errors.XenError('Unimplemented', opterr='In legacy mode') + + if not VdiType.isCowImage(self.vdi_type): + raise xs_errors.XenError('Unimplemented', opterr='RAW formats not supported') + + # Ensure that every involved VDIs aren't RAW + for vdi in [self, dest]: + vdi._loadThis() + if not util.pathexists(vdi.path): + raise xs_errors.XenError( + "VDIUnavailable", opterr=f"VDI unavailable: {self.path}" + ) + + # We need to activate the whole vdi chain to check if everything exists + with self.tap_pause(), dest.tap_pause(), self._activated_chain( + False + ), dest._activated_chain(False): + self._revert(dest, src_cbtlog, dest_cbtlog) + + def _revert( + self, + dest: "LVMVDI", + src_cbtlog: Optional[str], + dest_cbtlog: Optional[str], + ): + """This assumes that self, dest and their parents VDIs has been loaded and activated""" + self._ensure_not_max_depth() + + # Compute required size + ## Size for new destination vdi, it will be full size if thick + thick_destination_size, thin_destination_size = self._provisionning_sizes( + self.session.xenapi.VDI.get_by_uuid(self.uuid) + ) + destination_size = thick_destination_size if self.sr.provision == "thick" else thin_destination_size + + ## Size for the inserted base copy + inserted_size = util.roundup( + lvutil.LVM_SIZE_INCREMENT, self.cowutil.getSizePhys(self.path) + ) + + ## the space required must include a revert journal LV + ## It will also include a newly created destination disk and an inserted base copy + ## source image used for the revert + size_req = self.sr.journaler.LV_SIZE + destination_size + inserted_size + + inserted_uuid = util.gen_uuid() + inserted_lvname = LV_PREFIX[self.vdi_type] + inserted_uuid + + self.sr._ensureSpaceAvailable(size_req) + + # We create a valid backup VDI so that it properly protect + # it's parent chain in case of a coalesce + dest_backup_name = LV_PREFIX[self.vdi_type] + util.gen_uuid() + + # Create journal + + journal_id, value = RevertLogEntry( + dest=RevertLogDestinationVDI( + dest.uuid, + dest.lvname, + dest_backup_name, + is_cbt_enabled=bool(dest_cbtlog), + ), + src=RevertLogSourceVDI( + self.uuid, + self.lvname, + is_cbt_enabled=bool(src_cbtlog), + ), + inserted=RevertLogInsertedVDI(inserted_uuid, inserted_lvname), + ).to_journal() + self.sr.journaler.create(RevertLogEntry.JRN_KEY, journal_id, value) + + # Backup + ## First move the old vdi to a backup location to allow rolling back + ## This has to be done first because it's used as a signal to the rollback + ## algorithm to know if there is cleanup work to do + ## We deactivate it so that in case of an error we don't crash during + ## The chain deactivation on the finalize + dest.sr.lvActivator.deactivate(dest.uuid, False) + dest.sr.lvmCache.rename(dest.lvname, dest_backup_name) + + # Create the base copy by renaming the src snapshot + ## Since src is a snapshot, we don't need to deflate it + ## We deactivate it so that in case of an error we don't crash during + ## The chain deactivation on the finalize + self.sr.lvActivator.deactivate(self.uuid, False) + self.sr.lvmCache.rename(self.lvname, inserted_lvname) + + util.fistpoint.activate("LVM_revert_create_insert", self.sr.uuid) + + inserted = LVMVDI(self.sr, inserted_uuid) + inserted.label = "base copy" + inserted.read_only = False + inserted.location = inserted_uuid + inserted.sm_config = {} + inserted.sm_config["image-format"] = getImageStringFromVdiType(self.vdi_type) + if "key_hash" in self.sm_config: + inserted.sm_config["key_hash"] = self.sm_config["key_hash"] + inserted.cbt_enabled = False # Base copies don't have cbt + + ## Protect the new parent from being coalesced by the GC + inserted.sr.lvActivator.activate(inserted.uuid, inserted.lvname, False) + inserted.cowutil.setHidden(inserted.path, False) + inserted_ref = inserted._db_introduce() + + util.fistpoint.activate("LVM_revert_create_src", self.sr.uuid) + + inserted.disable_leaf_on_secondary(inserted.uuid, True) + + # Recreate src + ## Since src is a snapshot, we use the minimal size on it + _ = inserted._createSnap( + self.uuid, self.vdi_type, inserted_size, False, False + ) + + # Update src + self.sr.lvActivator.activate(self.uuid, self.lvname, False) + self.update_from_disk() + self._db_update() + + # Create snapshot + _ = inserted._createSnap( + dest.uuid, inserted.vdi_type, destination_size, False, False + ) + + util.fistpoint.activate("LVM_revert_create_dest", self.sr.uuid) + + dest.update_from_disk() + dest._db_update() + + ## The new parent can now be set hidden and don't need to be protected by the GC + inserted.update_from_disk() + inserted.read_only = True + inserted._db_update() + inserted.cowutil.setHidden(inserted.path, True) + inserted.sr.lvmCache.setReadonly(inserted.lvname, True) + self.session.xenapi.VDI.set_managed(inserted_ref, False) + + if src_cbtlog: + self._revert_cbt(dest) + elif dest_cbtlog: + util.SMlog( + f"Reverting {dest.uuid} to non-CBT snapshot, disabling CBT on dest" + ) + self._disable_cbt_on_vdi( + dest, + "VDI_CBT_REVERT_DISABLE", + f"CBT disabled on {dest.uuid}: reverted to non-CBT snapshot", + ) + + # Cleanup + ## Remove backup and journal + dest.sr.lvmCache.remove(dest_backup_name) + self.sr.journaler.remove(RevertLogEntry.JRN_KEY, journal_id) + + def update_from_disk(self): + """Update properties based on the disk content""" + if not VdiType.isCowImage(self.vdi_type): + raise xs_errors.XenError('Unimplemented', opterr='Can update from COW images') + + image_info = self.cowutil.getInfo(self.path, LvmCowUtil.extractUuid, False) + self.utilisation = image_info.sizePhys + self.size = image_info.sizeVirt + self.parent = image_info.parentUuid + self.hidden = image_info.hidden + self.read_only = self.sr.lvmCache.getLVInfo(self.lvname)[self.lvname].readonly + if self.parent: + self.sm_config['vhd-parent'] = self.parent + @override def _do_snapshot(self, sr_uuid, vdi_uuid, snapType, cloneOp=False, secondary=None, cbtlog=None, is_mirror_destination=False) -> str: @@ -1737,75 +2140,79 @@ def _do_snapshot(self, sr_uuid, vdi_uuid, snapType, (unpause_time - pause_time)) return snapResult - def _snapshot(self, snapType, cloneOp=False, cbtlog=None, cbt_consistency=None, is_mirror_destination=False): + def _snapshot( + self, + snapType, + cloneOp=False, + cbtlog=None, + cbt_consistency=None, + is_mirror_destination=False, + ): util.SMlog("LVMVDI._snapshot for %s (type %s)" % (self.uuid, snapType)) if not self.sr.isMaster: - raise xs_errors.XenError('LVMMaster') + raise xs_errors.XenError("LVMMaster") if self.sr.legacyMode: - raise xs_errors.XenError('Unimplemented', opterr='In legacy mode') + raise xs_errors.XenError("Unimplemented", opterr="In legacy mode") self._loadThis() if self.hidden: - raise xs_errors.XenError('VDISnapshot', opterr='hidden VDI') + raise xs_errors.XenError("VDISnapshot", opterr="hidden VDI") snapVdiType = self.sr._get_snap_vdi_type(self.vdi_type, self.size) - self.sm_config = self.session.xenapi.VDI.get_sm_config( \ - self.sr.srcmd.params['vdi_ref']) - if "type" in self.sm_config and self.sm_config['type'] == 'raw': + self.sm_config = self.session.xenapi.VDI.get_sm_config( + self.sr.srcmd.params["vdi_ref"] + ) + if "type" in self.sm_config and self.sm_config["type"] == "raw": if not util.fistpoint.is_active("testsm_clone_allow_raw"): - raise xs_errors.XenError('Unimplemented', \ - opterr='Raw VDI, snapshot or clone not permitted') + raise xs_errors.XenError( + "Unimplemented", opterr="Raw VDI, snapshot or clone not permitted" + ) # we must activate the entire image chain because the real parent could # theoretically be anywhere in the chain if all images under it are empty self._chainSetActive(True, False) if not util.pathexists(self.path): - raise xs_errors.XenError('VDIUnavailable', \ - opterr='VDI unavailable: %s' % (self.path)) + raise xs_errors.XenError( + "VDIUnavailable", opterr="VDI unavailable: %s" % (self.path) + ) - if VdiType.isCowImage(self.vdi_type): - depth = self.cowutil.getDepth(self.path) - if depth == -1: - raise xs_errors.XenError('VDIUnavailable', \ - opterr='failed to get COW depth') - elif depth >= self.cowutil.getMaxChainLength(): - raise xs_errors.XenError('SnapshotChainTooLong') + self._ensure_not_max_depth() - self.issnap = self.session.xenapi.VDI.get_is_a_snapshot( \ - self.sr.srcmd.params['vdi_ref']) + fullpr, thinpr = self._provisionning_sizes(self.sr.srcmd.params["vdi_ref"]) - fullpr = self.lvmcowutil.calcVolumeSize(self.size) - thinpr = util.roundup( - lvutil.LVM_SIZE_INCREMENT, - self.cowutil.calcOverheadEmpty(max(self.size, self.cowutil.getDefaultPreallocationSizeVirt())) - ) lvSizeOrig = thinpr lvSizeClon = thinpr hostRefs = [] + # If we do a snapshot, it might mean that the VDI is mounted somewhere + # If it's on another machine, we need to copy all data locally if self.sr.cmd == "vdi_snapshot": hostRefs = util.get_hosts_attached_on(self.session, [self.uuid]) if hostRefs: lvSizeOrig = fullpr + + # If we have a thick provisionned SR and we're doing a snapshot that won't be writable + # We take the minimum possible size for "archiving", it doesn't matter since it'll be read only if self.sr.provision == "thick": if not self.issnap: lvSizeOrig = fullpr if self.sr.cmd != "vdi_snapshot": lvSizeClon = fullpr - if (snapType == VDI.SNAPSHOT_SINGLE or - snapType == VDI.SNAPSHOT_INTERNAL): + if snapType == VDI.SNAPSHOT_SINGLE or snapType == VDI.SNAPSHOT_INTERNAL: lvSizeClon = 0 # the space required must include 2 journal LVs: a clone journal and an - # inflate journal (for the failure handling + # inflate journal (for the failure handling) size_req = lvSizeOrig + lvSizeClon + 2 * self.sr.journaler.LV_SIZE lvSizeBase = self.size if VdiType.isCowImage(self.vdi_type): - lvSizeBase = util.roundup(lvutil.LVM_SIZE_INCREMENT, self.cowutil.getSizePhys(self.path)) - size_req -= (self.utilisation - lvSizeBase) + lvSizeBase = util.roundup( + lvutil.LVM_SIZE_INCREMENT, self.cowutil.getSizePhys(self.path) + ) + size_req -= self.utilisation - lvSizeBase self.sr._ensureSpaceAvailable(size_req) if hostRefs: @@ -2090,6 +2497,17 @@ def _initFromVDIInfo(self, vdiInfo): self.sm_config_override = {'vdi_type': self.vdi_type} self.loaded = True + def _ensure_not_max_depth(self): + if not VdiType.isCowImage(self.vdi_type): + return # There is no depth concept outside of cow images + + depth = self.cowutil.getDepth(self.path) + if depth == -1: + raise xs_errors.XenError('VDIUnavailable', \ + opterr='failed to get COW depth') + elif depth >= self.cowutil.getMaxChainLength(): + raise xs_errors.XenError('SnapshotChainTooLong') + def _initFromLVInfo(self, lvInfo): self._setType(lvInfo.vdiType) self.lvname = lvInfo.name @@ -2120,8 +2538,13 @@ def _determineType(self): """ Determine whether this is a RAW or a COW VDI. """ - if "vdi_ref" in self.sr.srcmd.params: - vdi_ref = self.sr.srcmd.params["vdi_ref"] + + try: + vdi_ref = self.session.xenapi.VDI.get_by_uuid(self.uuid) + except XenAPI.Failure: + vdi_ref = None + + if vdi_ref: sm_config = self.session.xenapi.VDI.get_sm_config(vdi_ref) if sm_config.get("vdi_type"): self._setType(sm_config["vdi_type"]) @@ -2207,6 +2630,34 @@ def _chainSetActive(self, active, binary, persistent=False): # operation self.sr.lvActivator.add(uuid, lvName, binaryParam) + def _provisionning_sizes(self, vdi_ref: str) -> Tuple[int, int]: + self.issnap = self.session.xenapi.VDI.get_is_a_snapshot(vdi_ref) + + full_provisioning = self.lvmcowutil.calcVolumeSize(self.size) + thin_provisioning = util.roundup( + lvutil.LVM_SIZE_INCREMENT, + self.cowutil.calcOverheadEmpty( + max(self.size, self.cowutil.getDefaultPreallocationSizeVirt()) + ), + ) + return full_provisioning, thin_provisioning + + @contextmanager + def activated(self, binary, persistent=False): + self.sr.lvActivator.activate(self.uuid, self.lvname, binary, persistent) + try: + yield + finally: + self.sr.lvActivator.deactivate(self.uuid, binary, persistent) + + @contextmanager + def _activated_chain(self, binary, persistent=False): + self._chainSetActive(True, binary, persistent) + try: + yield + finally: + self.sr.cleanup() + def _failClone(self, uuid, jval, msg): try: self.sr._handleInterruptedCloneOp(uuid, jval, True) @@ -2379,6 +2830,15 @@ def _deactivate_cbt_log(self, lv_name) -> None: def _cbt_log_exists(self, logpath) -> bool: return lvutil.exists(logpath) + @override + def _create_cbt_log_with_size(self, size: int) -> str: + log_path = self._get_cbt_logpath(self.uuid) + logsize = max(util.roundup(CBT_BLOCK_SIZE, self.size//CBT_BLOCK_SIZE), self.sr.journaler.LV_SIZE) + # We choose 4MiB as the minimum for the log size to maintain the old behavior and compute the correct amount + # if we need a bigger LV for the CBT (can happen with big QCOW2) + self.sr.lvmCache.create(log_path, logsize, CBTLOG_TAG) + return super()._create_cbt_log_with_size(size) + if __name__ == '__main__': SRCommand.run(LVMSR, DRIVER_INFO) else: diff --git a/drivers/LVMoFCoESR.py b/drivers/LVMoFCoESR.py index e5995c6be..4b6469795 100755 --- a/drivers/LVMoFCoESR.py +++ b/drivers/LVMoFCoESR.py @@ -34,7 +34,7 @@ "VDI_GENERATE_CONFIG", "VDI_SNAPSHOT", "VDI_CLONE", "VDI_RESIZE", "ATOMIC_PAUSE", "VDI_RESET_ON_BOOT/2", "VDI_UPDATE", "VDI_MIRROR", "VDI_CONFIG_CBT", "VDI_ACTIVATE", - "VDI_DEACTIVATE"] + "VDI_DEACTIVATE", "VDI_REVERT"] CONFIGURATION = [['SCSIid', 'The scsi_id of the destination LUN'], ['allocation', 'Valid values are thick or thin(optional,\ diff --git a/drivers/LVMoHBASR.py b/drivers/LVMoHBASR.py index 38cb4ae7b..8617f32bd 100755 --- a/drivers/LVMoHBASR.py +++ b/drivers/LVMoHBASR.py @@ -41,7 +41,7 @@ "VDI_CREATE", "VDI_DELETE", "VDI_ATTACH", "VDI_DETACH", "VDI_GENERATE_CONFIG", "VDI_SNAPSHOT", "VDI_CLONE", "VDI_MIRROR", "VDI_RESIZE", "ATOMIC_PAUSE", "VDI_RESET_ON_BOOT/2", - "VDI_UPDATE", "VDI_CONFIG_CBT", "VDI_ACTIVATE", "VDI_DEACTIVATE"] + "VDI_UPDATE", "VDI_CONFIG_CBT", "VDI_ACTIVATE", "VDI_DEACTIVATE", "VDI_REVERT"] CONFIGURATION = [['SCSIid', 'The scsi_id of the destination LUN'], \ ['allocation', 'Valid values are thick or thin (optional, defaults to thick)']] diff --git a/drivers/LVMoISCSISR.py b/drivers/LVMoISCSISR.py index 9c7066e8c..f6d7da5ce 100755 --- a/drivers/LVMoISCSISR.py +++ b/drivers/LVMoISCSISR.py @@ -46,7 +46,7 @@ "VDI_GENERATE_CONFIG", "VDI_CLONE", "VDI_SNAPSHOT", "VDI_RESIZE", "ATOMIC_PAUSE", "VDI_RESET_ON_BOOT/2", "VDI_UPDATE", "VDI_MIRROR", "VDI_CONFIG_CBT", - "VDI_ACTIVATE", "VDI_DEACTIVATE"] + "VDI_ACTIVATE", "VDI_DEACTIVATE", "VDI_REVERT"] CONFIGURATION = [['SCSIid', 'The scsi_id of the destination LUN'], \ ['target', 'IP address or hostname of the iSCSI target'], \ diff --git a/drivers/journaler.py b/drivers/journaler.py index 45b8fe4a8..82f4ea1bd 100644 --- a/drivers/journaler.py +++ b/drivers/journaler.py @@ -19,7 +19,7 @@ import util import xs_errors -from srmetadata import open_file, file_read_wrapper, file_write_wrapper +from srmetadata import open_file, file_read_wrapper, file_write_wrapper, align_data_to_file LVM_MAX_NAME_LEN = 127 @@ -34,6 +34,16 @@ class Journaler: SEPARATOR = "_" JRN_CLONE = "clone" JRN_LEAF = "leaf" + JRN_REVERT = "revert" + + @classmethod + def has_additional_data(cls, type: str) -> bool: + """Return True if journal type contains additional data inside of it""" + return type in [ + cls.JRN_CLONE, + cls.JRN_LEAF, + cls.JRN_REVERT, + ] def __init__(self, lvmCache): self.vgName = lvmCache.vgName @@ -42,54 +52,63 @@ def __init__(self, lvmCache): def create(self, type, id, val): """Create an entry of type "type" for "id" with the value "val". Error if such an entry already exists.""" - valExisting = self.get(type, id) - writeData = False - if valExisting or util.fistpoint.is_active("LVM_journaler_exists"): - raise xs_errors.XenError('LVMCreate', opterr="Journal already exists for '%s:%s': %s" % (type, id, valExisting)) - lvName = self._getNameLV(type, id, val) + to_write = None + journal_exists = self.get(type, id) + if journal_exists or util.fistpoint.is_active("LVM_journaler_exists"): + raise xs_errors.XenError('LVMCreate', opterr=f"Journal already exists for '{type}:{id}': {journal_exists}") + lv_name = self._getNameLV(type, id, val) + + mapper_device = self._getLVMapperName(lv_name) + if len(mapper_device) > LVM_MAX_NAME_LEN: + lv_name = self._getNameLV(type, id) + mapper_device = self._getLVMapperName(lv_name) + assert len(mapper_device) <= LVM_MAX_NAME_LEN - mapperDevice = self._getLVMapperName(lvName) - if len(mapperDevice) > LVM_MAX_NAME_LEN: - lvName = self._getNameLV(type, id) - writeData = True - mapperDevice = self._getLVMapperName(lvName) - assert len(mapperDevice) <= LVM_MAX_NAME_LEN + try: + to_write = ("%d %s" % (len(val), val)).encode() + except UnicodeEncodeError as e: + util.logException("journaler.create") + raise xs_errors.XenError("LVMWrite", opterr=f"Failed to encode data for journal {lv_name}") from e + used_size = len(align_data_to_file(to_write)) + if used_size > self.LV_SIZE: + raise xs_errors.XenError("LVMWrite", opterr=f"Size of encoded journal {used_size} exceeds limit of {self.LV_SIZE}") + + + self.lvmCache.create(lv_name, self.LV_SIZE, self.LV_TAG) - self.lvmCache.create(lvName, self.LV_SIZE, self.LV_TAG) + if not to_write: + return - if writeData: - fullPath = self.lvmCache._getPath(lvName) - journal_file = open_file(fullPath, True) + full_path = self.lvmCache._getPath(lv_name) + journal_file = open_file(full_path, True) + try: + raised_exception = None try: - e = None + file_write_wrapper(journal_file, 0, to_write) + if util.fistpoint.is_active("LVM_journaler_writefail"): + raise ValueError("LVM_journaler_writefail FistPoint active") + except Exception as e: + raised_exception = e + raise + finally: try: - data = ("%d %s" % (len(val), val)).encode() - file_write_wrapper(journal_file, 0, data) - if util.fistpoint.is_active("LVM_journaler_writefail"): - raise ValueError("LVM_journaler_writefail FistPoint active") + journal_file.close() + self.lvmCache.deactivateNoRefcount(lv_name) except Exception as e: - raise - finally: - try: - journal_file.close() - self.lvmCache.deactivateNoRefcount(lvName) - except Exception as e2: - msg = 'failed to close/deactivate %s: %s' \ - % (lvName, e2) - if not e: - util.SMlog(msg) - raise e2 - else: - util.SMlog('WARNING: %s (error ignored)' % msg) - - except: - util.logException("journaler.create") - try: - self.lvmCache.remove(lvName) - except Exception as e: - util.SMlog('WARNING: failed to clean up failed journal ' \ - ' creation: %s (error ignored)' % e) - raise xs_errors.XenError('LVMWrite', opterr="Failed to write to journal %s" % lvName) + msg = f"failed to close/deactivate {lv_name}: {e}" + if not raised_exception: + util.SMlog(msg) + raise e + else: + util.SMlog(f"WARNING: {msg} (error ignored)") + + except: + util.logException("journaler.create") + try: + self.lvmCache.remove(lv_name) + except Exception as e: + util.SMlog(f"WARNING: failed to clean up failed journal creation: {e} (error ignored)") + raise xs_errors.XenError('LVMWrite', opterr=f"Failed to write to journal {lv_name}") def remove(self, type, id): """Remove the entry of type "type" for "id". Error if the entry doesn't @@ -139,26 +158,23 @@ def _getAllEntries(self, readFile=True): if len(parts) != 3 or util.fistpoint.is_active("LVM_journaler_badname"): raise xs_errors.XenError('LVMNoVolume', opterr="Bad LV name: %s" % lvName) type, id, val = parts - if readFile: - # For clone and leaf journals, additional - # data is written inside file - # TODO: Remove dependency on journal type - if type == self.JRN_CLONE or type == self.JRN_LEAF: - fullPath = self.lvmCache._getPath(lvName) - self.lvmCache.activateNoRefcount(lvName, False) - journal_file = open_file(fullPath) + # TODO: Remove dependency on journal type + if readFile and self.has_additional_data(type): + fullPath = self.lvmCache._getPath(lvName) + self.lvmCache.activateNoRefcount(lvName, False) + journal_file = open_file(fullPath) + try: try: - try: - data = file_read_wrapper(journal_file, 0) - length, val = data.decode().split(" ", 1) - val = val[:int(length)] - if util.fistpoint.is_active("LVM_journaler_readfail"): - raise ValueError("LVM_journaler_readfail FistPoint active") - except: - raise xs_errors.XenError('LVMRead', opterr="Failed to read from journal %s" % lvName) - finally: - journal_file.close() - self.lvmCache.deactivateNoRefcount(lvName) + data = file_read_wrapper(journal_file, 0, -1) + length, val = data.decode().split(" ", 1) + val = val[:int(length)] + if util.fistpoint.is_active("LVM_journaler_readfail"): + raise ValueError("LVM_journaler_readfail FistPoint active") + except: + raise xs_errors.XenError('LVMRead', opterr="Failed to read from journal %s" % lvName) + finally: + journal_file.close() + self.lvmCache.deactivateNoRefcount(lvName) if not entries.get(type): entries[type] = dict() entries[type][id] = val diff --git a/drivers/srmetadata.py b/drivers/srmetadata.py index 57a4f801c..7a4e0dddd 100755 --- a/drivers/srmetadata.py +++ b/drivers/srmetadata.py @@ -102,6 +102,14 @@ def open_file(path, write=False): (path, e.errno)) return file_p +def align_data_to_file(data: bytes) -> bytes: + """Add space padding to data to ensure that a complete blocks can be written.""" + blocksize = METADATA_BLK_SIZE + length = len(data) + newlength = length + if length % blocksize: + newlength = length + (blocksize - length % blocksize) + return data + b' ' * (newlength - length) def file_write_wrapper(fd, offset, data): """ @@ -109,14 +117,10 @@ def file_write_wrapper(fd, offset, data): may be written out after the given data to ensure that complete blocks are written. """ + blocksize = METADATA_BLK_SIZE try: - blocksize = METADATA_BLK_SIZE - length = len(data) - newlength = length - if length % blocksize: - newlength = length + (blocksize - length % blocksize) + to_write = align_data_to_file(data) fd.seek(offset, SEEK_SET) - to_write = data + b' ' * (newlength - length) return fd.write(to_write) except OSError as e: raise OSError( @@ -127,10 +131,16 @@ def file_write_wrapper(fd, offset, data): def file_read_wrapper(fd, offset, bytesToRead=METADATA_BLK_SIZE): """ Reads data from a file at a given offset. If not specified, the amount of - data to read defaults to one block. + data to read defaults to one block. Using -1 reads the whole file. """ try: fd.seek(offset, SEEK_SET) + if bytesToRead == -1: + # The most reliable solution I could find to read the whole file + # is to read it to the end to find the size + # This isn't super efficient but in most case where this is used, the file + # is quite short and that shouldn't be an issue + bytesToRead = len(fd.peek()) return fd.read(bytesToRead) except OSError as e: raise OSError( diff --git a/drivers/util.py b/drivers/util.py index a5b9ea7c3..057a4e3db 100755 --- a/drivers/util.py +++ b/drivers/util.py @@ -1549,6 +1549,9 @@ def list_find(f, seq): "FileSR_revert_create_insert", "FileSR_revert_create_src", "FileSR_revert_create_dest", + "LVM_revert_create_insert", + "LVM_revert_create_src", + "LVM_revert_create_dest", "LVM_journaler_exists", "LVM_journaler_none", "LVM_journaler_badname", diff --git a/tests/test_LVMSR.py b/tests/test_LVMSR.py index 0151e2f29..256315da7 100644 --- a/tests/test_LVMSR.py +++ b/tests/test_LVMSR.py @@ -1,3 +1,4 @@ +import XenAPI from sm_typing import override import copy @@ -698,6 +699,10 @@ def test_update_slaves_on_cbt_disable(self, mock_xenapi, mock_lock): Ensure we tell the supporter host when we disable CBT for one of its VMs """ # Arrange + def xenapi_failure(_): + raise XenAPI.Failure("No uuid") + + mock_xenapi.xapi_local.return_value.xenapi.VDI.get_by_uuid.side_effect = xenapi_failure xapi_session = mock_xenapi.xapi_local.return_value vdi_uuid = str(uuid.uuid4)