diff --git a/conf/springConfigXml/VolumeManager.xml b/conf/springConfigXml/VolumeManager.xml index 99c93fefa1b..6b001b19d80 100644 --- a/conf/springConfigXml/VolumeManager.xml +++ b/conf/springConfigXml/VolumeManager.xml @@ -100,6 +100,8 @@ + + diff --git a/header/src/main/java/org/zstack/header/volume/VolumeBackupEncryptionConversionContext.java b/header/src/main/java/org/zstack/header/volume/VolumeBackupEncryptionConversionContext.java new file mode 100644 index 00000000000..b848847a859 --- /dev/null +++ b/header/src/main/java/org/zstack/header/volume/VolumeBackupEncryptionConversionContext.java @@ -0,0 +1,31 @@ +package org.zstack.header.volume; + +public class VolumeBackupEncryptionConversionContext { + private String volumeUuid; + private boolean targetEncrypted; + private Object extensionData; + + public String getVolumeUuid() { + return volumeUuid; + } + + public void setVolumeUuid(String volumeUuid) { + this.volumeUuid = volumeUuid; + } + + public boolean isTargetEncrypted() { + return targetEncrypted; + } + + public void setTargetEncrypted(boolean targetEncrypted) { + this.targetEncrypted = targetEncrypted; + } + + public Object getExtensionData() { + return extensionData; + } + + public void setExtensionData(Object extensionData) { + this.extensionData = extensionData; + } +} diff --git a/header/src/main/java/org/zstack/header/volume/VolumeBackupEncryptionConversionExtensionPoint.java b/header/src/main/java/org/zstack/header/volume/VolumeBackupEncryptionConversionExtensionPoint.java new file mode 100644 index 00000000000..69b26eb6ad8 --- /dev/null +++ b/header/src/main/java/org/zstack/header/volume/VolumeBackupEncryptionConversionExtensionPoint.java @@ -0,0 +1,14 @@ +package org.zstack.header.volume; + +import org.zstack.header.core.ReturnValueCompletion; + +public interface VolumeBackupEncryptionConversionExtensionPoint { + void prepareVolumeBackupEncryptionConversion(VolumeInventory volume, boolean encrypted, + ReturnValueCompletion completion); + + void commitVolumeBackupEncryptionConversion(VolumeBackupEncryptionConversionContext context); + + void rollbackVolumeBackupEncryptionConversion(VolumeBackupEncryptionConversionContext context); + + void afterVolumeBackupEncryptionConversionCommitted(VolumeBackupEncryptionConversionContext context); +} diff --git a/plugin/ceph/src/main/java/org/zstack/storage/ceph/primary/CephPrimaryStorageBase.java b/plugin/ceph/src/main/java/org/zstack/storage/ceph/primary/CephPrimaryStorageBase.java index 47ffbda6611..2447ccc6b8a 100755 --- a/plugin/ceph/src/main/java/org/zstack/storage/ceph/primary/CephPrimaryStorageBase.java +++ b/plugin/ceph/src/main/java/org/zstack/storage/ceph/primary/CephPrimaryStorageBase.java @@ -3566,7 +3566,7 @@ protected void handle(ConvertVolumeEncryptionOnPrimaryStorageMsg msg) { String hostUuid; try { - hostUuid = findConnectedHostForCephLuks(); + hostUuid = resolveHostForCephLuks(msg.getHostUuid()); } catch (OperationFailureException e) { reply.setError(e.getErrorCode()); bus.reply(msg, reply); @@ -3600,6 +3600,33 @@ public void success(KVMHostLuksRsp ret) { }); } + private String resolveHostForCephLuks(String hostUuid) { + if (StringUtils.isBlank(hostUuid)) { + throw new OperationFailureException(operr( + "convert volume encryption on ceph primary storage[uuid:%s] requires hostUuid", + self.getUuid())); + } + + boolean hostAvailable = Q.New(HostVO.class) + .eq(HostVO_.uuid, hostUuid) + .eq(HostVO_.hypervisorType, KVMConstant.KVM_HYPERVISOR_TYPE) + .eq(HostVO_.status, HostStatus.Connected) + .eq(HostVO_.state, HostState.Enabled) + .isExists(); + PrimaryStorageHostStatus refStatus = Q.New(PrimaryStorageHostRefVO.class) + .eq(PrimaryStorageHostRefVO_.primaryStorageUuid, self.getUuid()) + .eq(PrimaryStorageHostRefVO_.hostUuid, hostUuid) + .select(PrimaryStorageHostRefVO_.status) + .findValue(); + if (!hostAvailable || refStatus != PrimaryStorageHostStatus.Connected) { + throw new OperationFailureException(operr( + "host[uuid:%s] is not an enabled connected KVM host attached to ceph primary storage[uuid:%s]", + hostUuid, self.getUuid())); + } + + return hostUuid; + } + @Override protected void handle(final DownloadIsoToPrimaryStorageMsg msg) { final DownloadIsoToPrimaryStorageReply reply = new DownloadIsoToPrimaryStorageReply(); diff --git a/plugin/localstorage/src/main/java/org/zstack/storage/primary/local/LocalStorageBase.java b/plugin/localstorage/src/main/java/org/zstack/storage/primary/local/LocalStorageBase.java index 5e4831fc1de..d16d264d37a 100755 --- a/plugin/localstorage/src/main/java/org/zstack/storage/primary/local/LocalStorageBase.java +++ b/plugin/localstorage/src/main/java/org/zstack/storage/primary/local/LocalStorageBase.java @@ -926,20 +926,11 @@ protected void handle(ConvertVolumeEncryptionOnPrimaryStorageMsg msg) { return; } - String hostUuid = StringUtils.isNotBlank(msg.getHostUuid()) ? - msg.getHostUuid() : null; - if (StringUtils.isBlank(hostUuid)) { - try { - hostUuid = getHostUuidByResourceUuid(msg.getVolume().getUuid()); - } catch (OperationFailureException e) { - reply.setError(e.getErrorCode()); - bus.reply(msg, reply); - return; - } - } - if (StringUtils.isBlank(hostUuid)) { - reply.setError(operr("cannot determine host for converting volume[uuid:%s] encryption on local primary storage[uuid:%s]", - msg.getVolume().getUuid(), self.getUuid())); + String hostUuid; + try { + hostUuid = resolveHostForConvertVolumeEncryption(msg); + } catch (OperationFailureException e) { + reply.setError(e.getErrorCode()); bus.reply(msg, reply); return; } @@ -959,6 +950,40 @@ public void fail(ErrorCode errorCode) { }); } + private String resolveHostForConvertVolumeEncryption(ConvertVolumeEncryptionOnPrimaryStorageMsg msg) { + String hostUuid = msg.getHostUuid(); + if (StringUtils.isBlank(hostUuid)) { + throw new OperationFailureException(operr( + "convert volume encryption on local primary storage[uuid:%s] requires hostUuid for volume[uuid:%s]", + self.getUuid(), msg.getVolume().getUuid())); + } + + String actualHostUuid = Q.New(LocalStorageResourceRefVO.class) + .eq(LocalStorageResourceRefVO_.resourceUuid, msg.getVolume().getUuid()) + .eq(LocalStorageResourceRefVO_.resourceType, VolumeVO.class.getSimpleName()) + .eq(LocalStorageResourceRefVO_.primaryStorageUuid, self.getUuid()) + .select(LocalStorageResourceRefVO_.hostUuid) + .findValue(); + if (StringUtils.isBlank(actualHostUuid) || !hostUuid.equals(actualHostUuid)) { + throw new OperationFailureException(operr( + "volume[uuid:%s] on local primary storage[uuid:%s] belongs to host[uuid:%s], cannot convert encryption on host[uuid:%s]", + msg.getVolume().getUuid(), self.getUuid(), actualHostUuid, hostUuid)); + } + + boolean hostReady = Q.New(HostVO.class) + .eq(HostVO_.uuid, hostUuid) + .eq(HostVO_.status, HostStatus.Connected) + .eq(HostVO_.state, HostState.Enabled) + .isExists(); + if (!hostReady) { + throw new OperationFailureException(operr( + "host[uuid:%s] is not enabled and connected for converting volume[uuid:%s] encryption on local primary storage[uuid:%s]", + hostUuid, msg.getVolume().getUuid(), self.getUuid())); + } + + return hostUuid; + } + private void handle(EncryptVolumeBitsOnPrimaryStorageMsg msg) { if (StringUtils.isBlank(msg.getHostUuid()) || StringUtils.isBlank(msg.getVolumeUuid())) { EncryptVolumeBitsOnPrimaryStorageReply reply = new EncryptVolumeBitsOnPrimaryStorageReply(); diff --git a/plugin/nfsPrimaryStorage/src/main/java/org/zstack/storage/primary/nfs/NfsPrimaryStorageKVMBackend.java b/plugin/nfsPrimaryStorage/src/main/java/org/zstack/storage/primary/nfs/NfsPrimaryStorageKVMBackend.java index abb511dbd9c..7a2e3640736 100755 --- a/plugin/nfsPrimaryStorage/src/main/java/org/zstack/storage/primary/nfs/NfsPrimaryStorageKVMBackend.java +++ b/plugin/nfsPrimaryStorage/src/main/java/org/zstack/storage/primary/nfs/NfsPrimaryStorageKVMBackend.java @@ -2351,7 +2351,13 @@ public void run(MessageReply reply) { @Override public void handle(PrimaryStorageInventory inv, ConvertVolumeEncryptionOnPrimaryStorageMsg msg, ReturnValueCompletion completion) { - HostInventory host = nfsFactory.getConnectedHostForOperation(inv).get(0); + HostInventory host; + try { + host = resolveHostForConvertVolumeEncryption(inv, msg.getHostUuid()); + } catch (OperationFailureException e) { + completion.fail(e.getErrorCode()); + return; + } ConvertVolumeEncryptionCmd cmd = new ConvertVolumeEncryptionCmd(); cmd.setUuid(inv.getUuid()); cmd.volumeUuid = msg.getVolume().getUuid(); @@ -2387,6 +2393,38 @@ public void run(MessageReply reply) { }); } + private HostInventory resolveHostForConvertVolumeEncryption(PrimaryStorageInventory inv, String hostUuid) { + if (StringUtils.isBlank(hostUuid)) { + throw new OperationFailureException(operr( + "convert volume encryption on NFS primary storage[uuid:%s] requires hostUuid", + inv.getUuid())); + } + + HostVO host = Q.New(HostVO.class) + .eq(HostVO_.uuid, hostUuid) + .eq(HostVO_.status, HostStatus.Connected) + .eq(HostVO_.state, HostState.Enabled) + .find(); + if (host == null || !inv.getAttachedClusterUuids().contains(host.getClusterUuid())) { + throw new OperationFailureException(operr( + "host[uuid:%s] is not an enabled connected host attached to NFS primary storage[uuid:%s]", + hostUuid, inv.getUuid())); + } + + PrimaryStorageHostStatus status = Q.New(PrimaryStorageHostRefVO.class) + .eq(PrimaryStorageHostRefVO_.primaryStorageUuid, inv.getUuid()) + .eq(PrimaryStorageHostRefVO_.hostUuid, hostUuid) + .select(PrimaryStorageHostRefVO_.status) + .findValue(); + if (status != PrimaryStorageHostStatus.Connected) { + throw new OperationFailureException(operr( + "host[uuid:%s] is not connected to NFS primary storage[uuid:%s]", + hostUuid, inv.getUuid())); + } + + return HostInventory.valueOf(host); + } + private String prepareVolumeEncryptedDek(String hostUuid, VolumeInventory volume, boolean required) { if (!required) { return null; diff --git a/storage/src/main/java/org/zstack/storage/encrypt/VolumeBackupEncryptionConversionExtensionHelper.java b/storage/src/main/java/org/zstack/storage/encrypt/VolumeBackupEncryptionConversionExtensionHelper.java new file mode 100644 index 00000000000..0bd865d1e1f --- /dev/null +++ b/storage/src/main/java/org/zstack/storage/encrypt/VolumeBackupEncryptionConversionExtensionHelper.java @@ -0,0 +1,127 @@ +package org.zstack.storage.encrypt; + +import org.springframework.beans.factory.annotation.Autowired; +import org.zstack.core.asyncbatch.While; +import org.zstack.core.componentloader.PluginRegistry; +import org.zstack.header.core.ReturnValueCompletion; +import org.zstack.header.core.WhileCompletion; +import org.zstack.header.core.WhileDoneCompletion; +import org.zstack.header.errorcode.ErrorCode; +import org.zstack.header.errorcode.ErrorCodeList; +import org.zstack.header.volume.VolumeBackupEncryptionConversionContext; +import org.zstack.header.volume.VolumeBackupEncryptionConversionExtensionPoint; +import org.zstack.header.volume.VolumeInventory; +import org.zstack.utils.Utils; +import org.zstack.utils.logging.CLogger; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.zstack.core.Platform.operr; + +public class VolumeBackupEncryptionConversionExtensionHelper { + private static final CLogger logger = Utils.getLogger(VolumeBackupEncryptionConversionExtensionHelper.class); + @Autowired + private PluginRegistry pluginRgty; + + public static class Context { + private final VolumeBackupEncryptionConversionExtensionPoint extension; + private final VolumeBackupEncryptionConversionContext context; + + public Context(VolumeBackupEncryptionConversionExtensionPoint extension, + VolumeBackupEncryptionConversionContext context) { + this.extension = extension; + this.context = context; + } + } + + public void prepare(VolumeInventory volume, boolean targetEncrypted, ReturnValueCompletion> completion) { + List extensions = + pluginRgty.getExtensionList(VolumeBackupEncryptionConversionExtensionPoint.class); + List contexts = Collections.synchronizedList(new ArrayList<>()); + new While<>(extensions).each((ext, whileCompletion) -> + prepareExtension(ext, volume, targetEncrypted, contexts, whileCompletion) + ).run(new WhileDoneCompletion(completion) { + @Override + public void done(ErrorCodeList errorCodeList) { + if (errorCodeList.getCauses().isEmpty()) { + completion.success(new ArrayList<>(contexts)); + return; + } + + rollback(contexts, volume.getUuid()); + completion.fail(errorCodeList.getCauses().get(0)); + } + }); + } + + public void commit(List contexts) { + for (Context context : safeContexts(contexts)) { + context.extension.commitVolumeBackupEncryptionConversion(context.context); + } + } + + public void rollback(List contexts, String volumeUuid) { + List reversed = new ArrayList<>(safeContexts(contexts)); + Collections.reverse(reversed); + for (Context context : reversed) { + try { + context.extension.rollbackVolumeBackupEncryptionConversion(context.context); + } catch (RuntimeException e) { + logger.warn(String.format("failed to rollback volume backup encryption conversion for volume[uuid:%s]: %s", + volumeUuid, e.getMessage()), e); + } + } + } + + public void cleanupCommitted(List contexts, String volumeUuid) { + for (Context context : safeContexts(contexts)) { + try { + context.extension.afterVolumeBackupEncryptionConversionCommitted(context.context); + } catch (RuntimeException e) { + logger.warn(String.format("failed to run volume backup encryption conversion cleanup for volume[uuid:%s]: %s", + volumeUuid, e.getMessage()), e); + } + } + } + + private List safeContexts(List contexts) { + return contexts == null ? Collections.emptyList() : contexts; + } + + private Context makeContext(VolumeBackupEncryptionConversionExtensionPoint extension, + VolumeBackupEncryptionConversionContext context) { + return new Context(extension, context); + } + + private void prepareExtension(VolumeBackupEncryptionConversionExtensionPoint ext, + VolumeInventory volume, + boolean targetEncrypted, + List contexts, + WhileCompletion whileCompletion) { + try { + ext.prepareVolumeBackupEncryptionConversion(volume, targetEncrypted, + new ReturnValueCompletion(whileCompletion) { + @Override + public void success(VolumeBackupEncryptionConversionContext context) { + if (context != null) { + contexts.add(makeContext(ext, context)); + } + whileCompletion.done(); + } + + @Override + public void fail(ErrorCode errorCode) { + whileCompletion.addError(errorCode); + whileCompletion.allDone(); + } + } + ); + } catch (RuntimeException e) { + whileCompletion.addError(operr("failed to prepare volume backup encryption conversion for volume[uuid:%s]: %s", + volume.getUuid(), e.getMessage())); + whileCompletion.allDone(); + } + } +} diff --git a/storage/src/main/java/org/zstack/storage/encrypt/VolumeEncryptionConversionHostLeaseHelper.java b/storage/src/main/java/org/zstack/storage/encrypt/VolumeEncryptionConversionHostLeaseHelper.java new file mode 100644 index 00000000000..61e80581354 --- /dev/null +++ b/storage/src/main/java/org/zstack/storage/encrypt/VolumeEncryptionConversionHostLeaseHelper.java @@ -0,0 +1,569 @@ +package org.zstack.storage.encrypt; + +import org.apache.commons.lang.StringUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.zstack.core.thread.ThreadFacade; +import org.zstack.core.thread.ThreadFacadeImpl; +import org.zstack.core.db.DatabaseFacade; +import org.zstack.core.db.GLock; +import org.zstack.core.db.Q; +import org.zstack.core.db.SQL; +import org.zstack.core.db.SimpleQuery; +import org.zstack.header.core.workflow.FlowTrigger; +import org.zstack.core.jsonlabel.JsonLabelVO; +import org.zstack.core.jsonlabel.JsonLabelVO_; +import org.zstack.header.errorcode.OperationFailureException; +import org.zstack.header.host.HostState; +import org.zstack.header.host.HostStatus; +import org.zstack.header.host.HostVO; +import org.zstack.header.host.HostVO_; +import org.zstack.header.managementnode.ManagementNodeVO; +import org.zstack.header.managementnode.ManagementNodeVO_; +import org.zstack.header.storage.primary.PrimaryStorageClusterRefVO; +import org.zstack.header.storage.primary.PrimaryStorageClusterRefVO_; +import org.zstack.header.storage.primary.PrimaryStorageHostRefVO; +import org.zstack.header.storage.primary.PrimaryStorageHostRefVO_; +import org.zstack.header.storage.primary.PrimaryStorageHostStatus; +import org.zstack.header.storage.primary.PrimaryStorageVO; +import org.zstack.header.storage.primary.PrimaryStorageVO_; +import org.zstack.header.tag.SystemTagVO; +import org.zstack.header.tag.SystemTagVO_; +import org.zstack.header.vm.VmInstanceConstant; +import org.zstack.header.vm.VmInstanceVO; +import org.zstack.header.vm.VmInstanceVO_; +import org.zstack.header.volume.VolumeVO; +import org.zstack.storage.volume.VolumeSystemTags; +import org.zstack.utils.Utils; +import org.zstack.utils.logging.CLogger; + +import java.util.Collections; +import java.util.Date; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.zstack.core.Platform.getManagementServerId; +import static org.zstack.core.Platform.operr; + +public class VolumeEncryptionConversionHostLeaseHelper { + private static final CLogger logger = Utils.getLogger(VolumeEncryptionConversionHostLeaseHelper.class); + private static final String LEASE_KEY_PREFIX = "volumeEncryptionConversionHostLease::"; + private static final String LEASE_LOCK_PREFIX = "volEncConvHost::"; + private static final String LEASE_SEPARATOR = "::"; + private static final String LOCAL_STORAGE_HOST_TAG_PREFIX = "localStorage::hostUuid::"; + public static final long WAIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(3); + public static final long WAIT_TIMEOUT_MS = TimeUnit.HOURS.toMillis(2); + public static final long LEASE_TTL_MS = TimeUnit.MINUTES.toMillis(30); + public static final long LEASE_RENEW_INTERVAL_MS = TimeUnit.MINUTES.toMillis(5); + + @Autowired + private DatabaseFacade dbf; + @Autowired + private ThreadFacade thdf; + + public static class Lease { + private String hostUuid; + private String vmUuid; + private String volumeUuid; + private String labelValue; + private boolean released; + + public String getHostUuid() { + return hostUuid; + } + + public boolean isReleased() { + return released; + } + } + + public static class LeaseSession { + private Lease lease; + private ThreadFacadeImpl.TimeoutTaskReceipt renewReceipt; + + public String getHostUuid() { + return lease == null ? null : lease.getHostUuid(); + } + } + + public static class Reservation { + private Lease lease; + private String hostUuid; + private String vmUuid; + private boolean sameVmBusy; + + public Lease getLease() { + return lease; + } + + public String getHostUuid() { + return hostUuid; + } + + public String getVmUuid() { + return vmUuid; + } + + public boolean isSameVmBusy() { + return sameVmBusy; + } + } + + public String makeQueueSignature(String volumeUuid, String fallback) { + VolumeVO volume = dbf.findByUuid(volumeUuid, VolumeVO.class); + if (volume == null) { + return fallback; + } + + try { + String hostUuid = resolveHostUuid(volume); + if (StringUtils.isBlank(hostUuid)) { + return fallback; + } + + return String.format("volumeEncryptionConversion::%s::%s", hostUuid, resolveVmUuid(volume)); + } catch (RuntimeException e) { + logger.warn(String.format("failed to make volume encryption conversion queue signature for volume[uuid:%s]: %s", + volumeUuid, e.getMessage()), e); + return fallback; + } + } + + public Reservation tryReserve(VolumeVO volume) { + String hostUuid = resolveHostUuid(volume); + if (StringUtils.isBlank(hostUuid)) { + throw new OperationFailureException(operr( + "cannot find a connected KVM host to convert volume[uuid:%s] encryption on primary storage[uuid:%s]", + volume.getUuid(), volume.getPrimaryStorageUuid())); + } + + String vmUuid = resolveVmUuid(volume); + GLock lock = newLeaseLock(hostUuid); + try { + String labelKey = leaseKey(hostUuid); + JsonLabelVO label = Q.New(JsonLabelVO.class).eq(JsonLabelVO_.labelKey, labelKey).find(); + if (label == null) { + return reserved(createLease(labelKey, hostUuid, vmUuid, volume.getUuid()), hostUuid, vmUuid); + } + + if (leaseExpired(label) || ownerManagementNodeRestarted(label)) { + SQL.New(JsonLabelVO.class).eq(JsonLabelVO_.labelKey, labelKey).hardDelete(); + return reserved(createLease(labelKey, hostUuid, vmUuid, volume.getUuid()), hostUuid, vmUuid); + } + + String ownerVmUuid = leaseVmUuid(label.getLabelValue()); + if (StringUtils.isBlank(ownerVmUuid)) { + SQL.New(JsonLabelVO.class).eq(JsonLabelVO_.labelKey, labelKey).hardDelete(); + return reserved(createLease(labelKey, hostUuid, vmUuid, volume.getUuid()), hostUuid, vmUuid); + } + + if (!vmUuid.equals(ownerVmUuid)) { + throw new OperationFailureException(operr( + "host[uuid:%s] is converting volume encryption for vm[uuid:%s], cannot convert volume[uuid:%s] for vm[uuid:%s] at the same time", + hostUuid, ownerVmUuid, volume.getUuid(), vmUuid)); + } + + return sameVmBusy(hostUuid, vmUuid); + } finally { + lock.unlock(); + } + } + + public void reserve(VolumeVO volume, FlowTrigger trigger, AtomicReference sessionRef) { + reserve(volume, trigger, sessionRef, System.currentTimeMillis() + WAIT_TIMEOUT_MS); + } + + public void release(LeaseSession session) { + if (session == null) { + return; + } + + if (session.renewReceipt != null) { + session.renewReceipt.cancel(); + } + release(session.lease); + } + + public boolean renew(Lease lease) { + if (lease == null || lease.released) { + return false; + } + + GLock lock; + try { + lock = newLeaseLock(lease.hostUuid); + } catch (RuntimeException e) { + logger.warn(String.format("failed to lock host[uuid:%s] lease while renewing volume[uuid:%s] encryption conversion: %s", + lease.hostUuid, lease.volumeUuid, e.getMessage()), e); + return false; + } + + try { + JsonLabelVO label = Q.New(JsonLabelVO.class).eq(JsonLabelVO_.labelKey, leaseKey(lease.hostUuid)).find(); + if (label == null || !lease.labelValue.equals(label.getLabelValue())) { + return false; + } + + lease.labelValue = makeLeaseValue(lease.vmUuid, lease.volumeUuid, newLeaseExpireAt(), currentManagementNodeJoinDate()); + label.setLabelValue(lease.labelValue); + dbf.update(label); + return true; + } catch (RuntimeException e) { + logger.warn(String.format("failed to renew host[uuid:%s] lease for volume[uuid:%s] encryption conversion: %s", + lease.hostUuid, lease.volumeUuid, e.getMessage()), e); + return false; + } finally { + lock.unlock(); + } + } + + private void reserve(VolumeVO volume, FlowTrigger trigger, AtomicReference sessionRef, long deadline) { + try { + Reservation reservation = tryReserve(volume); + if (reservation.getLease() != null) { + LeaseSession session = new LeaseSession(); + session.lease = reservation.getLease(); + sessionRef.set(session); + scheduleRenew(session); + trigger.next(); + return; + } + + if (!reservation.isSameVmBusy()) { + trigger.fail(operr("failed to reserve host for converting volume[uuid:%s] encryption", volume.getUuid())); + return; + } + + if (System.currentTimeMillis() >= deadline) { + trigger.fail(operr( + "timeout waiting for vm[uuid:%s] volume encryption conversion on host[uuid:%s] before converting volume[uuid:%s]", + reservation.getVmUuid(), reservation.getHostUuid(), volume.getUuid())); + return; + } + + thdf.submitTimeoutTask(() -> reserve(volume, trigger, sessionRef, deadline), + TimeUnit.MILLISECONDS, WAIT_INTERVAL_MS); + } catch (OperationFailureException e) { + trigger.fail(e.getErrorCode()); + } catch (RuntimeException e) { + trigger.fail(operr("failed to reserve host for converting volume[uuid:%s] encryption: %s", + volume.getUuid(), e.getMessage())); + } + } + + private void scheduleRenew(LeaseSession session) { + if (session == null || session.lease == null || session.lease.isReleased()) { + return; + } + + session.renewReceipt = thdf.submitTimeoutTask(() -> { + if (renew(session.lease)) { + scheduleRenew(session); + } + }, TimeUnit.MILLISECONDS, LEASE_RENEW_INTERVAL_MS); + } + + public void release(Lease lease) { + if (lease == null) { + return; + } + if (lease.released) { + return; + } + lease.released = true; + + GLock lock; + try { + lock = newLeaseLock(lease.hostUuid); + } catch (RuntimeException e) { + logger.warn(String.format("failed to lock host[uuid:%s] lease while releasing volume[uuid:%s] encryption conversion: %s", + lease.hostUuid, lease.volumeUuid, e.getMessage()), e); + return; + } + + try { + String labelKey = leaseKey(lease.hostUuid); + JsonLabelVO label = Q.New(JsonLabelVO.class).eq(JsonLabelVO_.labelKey, labelKey).find(); + if (label == null) { + return; + } + + if (!lease.labelValue.equals(label.getLabelValue())) { + return; + } + + SQL.New(JsonLabelVO.class).eq(JsonLabelVO_.labelKey, labelKey).hardDelete(); + } catch (RuntimeException e) { + logger.warn(String.format("failed to release host[uuid:%s] lease for volume[uuid:%s] encryption conversion: %s", + lease.hostUuid, lease.volumeUuid, e.getMessage()), e); + } finally { + lock.unlock(); + } + } + + private Lease createLease(String labelKey, String hostUuid, String vmUuid, String volumeUuid) { + String labelValue = makeLeaseValue(vmUuid, volumeUuid, newLeaseExpireAt(), currentManagementNodeJoinDate()); + JsonLabelVO label = new JsonLabelVO(); + label.setLabelKey(labelKey); + label.setResourceUuid(hostUuid); + label.setLabelValue(labelValue); + dbf.persist(label); + + Lease lease = new Lease(); + lease.hostUuid = hostUuid; + lease.vmUuid = vmUuid; + lease.volumeUuid = volumeUuid; + lease.labelValue = labelValue; + return lease; + } + + private Reservation reserved(Lease lease, String hostUuid, String vmUuid) { + Reservation reservation = new Reservation(); + reservation.lease = lease; + reservation.hostUuid = hostUuid; + reservation.vmUuid = vmUuid; + return reservation; + } + + private Reservation sameVmBusy(String hostUuid, String vmUuid) { + Reservation reservation = new Reservation(); + reservation.hostUuid = hostUuid; + reservation.vmUuid = vmUuid; + reservation.sameVmBusy = true; + return reservation; + } + + private GLock newLeaseLock(String hostUuid) { + GLock lock = new GLock(LEASE_LOCK_PREFIX + hostUuid, TimeUnit.SECONDS.toSeconds(30)); + lock.lock(); + return lock; + } + + private String leaseKey(String hostUuid) { + return LEASE_KEY_PREFIX + hostUuid; + } + + private String leaseVmUuid(String labelValue) { + return StringUtils.substringBefore(labelValue, LEASE_SEPARATOR); + } + + private boolean ownerManagementNodeRestarted(JsonLabelVO label) { + String ownerUuid = leaseOwnerManagementNodeUuid(label.getLabelValue()); + Long ownerJoinDate = leaseOwnerManagementNodeJoinDate(label.getLabelValue()); + if (StringUtils.isBlank(ownerUuid) || ownerJoinDate == null) { + return false; + } + + Date currentOwnerJoinDate = Q.New(ManagementNodeVO.class) + .eq(ManagementNodeVO_.uuid, ownerUuid) + .select(ManagementNodeVO_.joinDate) + .findValue(); + return currentOwnerJoinDate == null || !ownerJoinDate.equals(currentOwnerJoinDate.getTime()); + } + + private boolean leaseExpired(JsonLabelVO label) { + Long expireAt = leaseExpireAt(label.getLabelValue()); + if (expireAt != null) { + return expireAt < System.currentTimeMillis(); + } + + Date lastAliveDate = label.getLastOpDate() == null ? label.getCreateDate() : label.getLastOpDate(); + return lastAliveDate == null || System.currentTimeMillis() - lastAliveDate.getTime() > LEASE_TTL_MS; + } + + private Long leaseExpireAt(String labelValue) { + return leaseLongToken(labelValue, 2); + } + + private Long leaseOwnerManagementNodeJoinDate(String labelValue) { + return leaseLongToken(labelValue, 4); + } + + private String leaseOwnerManagementNodeUuid(String labelValue) { + return leaseToken(labelValue, 3); + } + + private Long leaseLongToken(String labelValue, int index) { + String value = leaseToken(labelValue, index); + if (StringUtils.isBlank(value)) { + return null; + } + + try { + return Long.valueOf(value); + } catch (NumberFormatException e) { + return null; + } + } + + private String leaseToken(String labelValue, int index) { + if (StringUtils.isBlank(labelValue)) { + return null; + } + + int start = 0; + for (int i = 0; i < index; i++) { + int separator = labelValue.indexOf(LEASE_SEPARATOR, start); + if (separator < 0) { + return null; + } + start = separator + LEASE_SEPARATOR.length(); + } + + int end = labelValue.indexOf(LEASE_SEPARATOR, start); + return end < 0 ? labelValue.substring(start) : labelValue.substring(start, end); + } + + private long newLeaseExpireAt() { + return System.currentTimeMillis() + LEASE_TTL_MS; + } + + private String makeLeaseValue(String vmUuid, String volumeUuid, long expireAt, Long ownerManagementNodeJoinDate) { + return vmUuid + LEASE_SEPARATOR + volumeUuid + LEASE_SEPARATOR + expireAt + + LEASE_SEPARATOR + getManagementServerId() + + LEASE_SEPARATOR + (ownerManagementNodeJoinDate == null ? "" : ownerManagementNodeJoinDate); + } + + private Long currentManagementNodeJoinDate() { + Date joinDate = Q.New(ManagementNodeVO.class) + .eq(ManagementNodeVO_.uuid, getManagementServerId()) + .select(ManagementNodeVO_.joinDate) + .findValue(); + return joinDate == null ? null : joinDate.getTime(); + } + + private String resolveVmUuid(VolumeVO volume) { + String vmUuid = StringUtils.defaultIfBlank(volume.getVmInstanceUuid(), volume.getLastVmInstanceUuid()); + return StringUtils.defaultIfBlank(vmUuid, volume.getUuid()); + } + + private String resolveHostUuid(VolumeVO volume) { + String hostUuid = resolveVolumeSecretHostUuid(volume); + if (StringUtils.isNotBlank(hostUuid)) { + return hostUuid; + } + + hostUuid = resolveVmHostUuid(volume.getVmInstanceUuid()); + if (StringUtils.isNotBlank(hostUuid)) { + return hostUuid; + } + + hostUuid = resolveVmHostUuid(volume.getLastVmInstanceUuid()); + if (StringUtils.isNotBlank(hostUuid)) { + return hostUuid; + } + + hostUuid = resolveLocalStorageHostUuidFromVolumeTag(volume); + if (StringUtils.isNotBlank(hostUuid)) { + return hostUuid; + } + + hostUuid = resolveLocalStorageHostUuidFromResourceRef(volume); + if (StringUtils.isNotBlank(hostUuid)) { + return hostUuid; + } + + if (isLocalStoragePrimaryStorage(volume.getPrimaryStorageUuid())) { + return null; + } + + return findConnectedKvmHostAttachedToPrimaryStorage(volume.getPrimaryStorageUuid()); + } + + private String resolveVolumeSecretHostUuid(VolumeVO volume) { + if (volume == null) { + return null; + } + + List tags = VolumeSystemTags.VOLUME_LIBVIRT_SECRET_HOST.getTags(volume.getUuid(), VolumeVO.class); + if (tags != null && !tags.isEmpty()) { + return VolumeSystemTags.VOLUME_LIBVIRT_SECRET_HOST.getTokenByTag( + tags.get(0), VolumeSystemTags.VOLUME_LIBVIRT_SECRET_HOST_TOKEN); + } + + return resolveVmHostUuid(volume.getVmInstanceUuid()); + } + + private String resolveVmHostUuid(String vmUuid) { + if (StringUtils.isBlank(vmUuid)) { + return null; + } + + VmInstanceVO vm = Q.New(VmInstanceVO.class).eq(VmInstanceVO_.uuid, vmUuid).find(); + if (vm == null) { + return null; + } + + return StringUtils.defaultIfBlank(vm.getHostUuid(), vm.getLastHostUuid()); + } + + private String resolveLocalStorageHostUuidFromVolumeTag(VolumeVO volume) { + List tags = Q.New(SystemTagVO.class) + .eq(SystemTagVO_.resourceUuid, volume.getUuid()) + .like(SystemTagVO_.tag, LOCAL_STORAGE_HOST_TAG_PREFIX + "%") + .select(SystemTagVO_.tag) + .listValues(); + if (tags == null || tags.isEmpty()) { + return null; + } + + return StringUtils.substringAfter(tags.get(0), LOCAL_STORAGE_HOST_TAG_PREFIX); + } + + private String resolveLocalStorageHostUuidFromResourceRef(VolumeVO volume) { + if (!isLocalStoragePrimaryStorage(volume.getPrimaryStorageUuid())) { + return null; + } + + return SQL.New("select hostUuid from LocalStorageResourceRefVO where resourceUuid = :resourceUuid and primaryStorageUuid = :primaryStorageUuid and resourceType = :resourceType", String.class) + .param("resourceUuid", volume.getUuid()) + .param("primaryStorageUuid", volume.getPrimaryStorageUuid()) + .param("resourceType", VolumeVO.class.getSimpleName()) + .limit(1) + .find(); + } + + private boolean isLocalStoragePrimaryStorage(String primaryStorageUuid) { + String primaryStorageType = Q.New(PrimaryStorageVO.class) + .eq(PrimaryStorageVO_.uuid, primaryStorageUuid) + .select(PrimaryStorageVO_.type) + .findValue(); + return "LocalStorage".equals(primaryStorageType); + } + + private String findConnectedKvmHostAttachedToPrimaryStorage(String primaryStorageUuid) { + List clusterUuids = Q.New(PrimaryStorageClusterRefVO.class) + .eq(PrimaryStorageClusterRefVO_.primaryStorageUuid, primaryStorageUuid) + .select(PrimaryStorageClusterRefVO_.clusterUuid) + .listValues(); + if (clusterUuids.isEmpty()) { + return null; + } + + List connectedHostUuids = Q.New(PrimaryStorageHostRefVO.class) + .eq(PrimaryStorageHostRefVO_.primaryStorageUuid, primaryStorageUuid) + .eq(PrimaryStorageHostRefVO_.status, PrimaryStorageHostStatus.Connected) + .select(PrimaryStorageHostRefVO_.hostUuid) + .listValues(); + if (!Q.New(PrimaryStorageHostRefVO.class) + .eq(PrimaryStorageHostRefVO_.primaryStorageUuid, primaryStorageUuid) + .isExists()) { + connectedHostUuids = Collections.emptyList(); + } else if (connectedHostUuids.isEmpty()) { + return null; + } + + SimpleQuery hostQuery = dbf.createQuery(HostVO.class); + hostQuery.add(HostVO_.clusterUuid, SimpleQuery.Op.IN, clusterUuids); + hostQuery.add(HostVO_.hypervisorType, SimpleQuery.Op.EQ, VmInstanceConstant.KVM_HYPERVISOR_TYPE); + hostQuery.add(HostVO_.status, SimpleQuery.Op.EQ, HostStatus.Connected); + hostQuery.add(HostVO_.state, SimpleQuery.Op.EQ, HostState.Enabled); + if (!connectedHostUuids.isEmpty()) { + hostQuery.add(HostVO_.uuid, SimpleQuery.Op.IN, connectedHostUuids); + } + hostQuery.select(HostVO_.uuid); + hostQuery.orderBy(HostVO_.uuid, SimpleQuery.Od.ASC); + hostQuery.setLimit(1); + return hostQuery.findValue(); + } +} diff --git a/storage/src/main/java/org/zstack/storage/volume/VolumeBase.java b/storage/src/main/java/org/zstack/storage/volume/VolumeBase.java index c364e790fe9..c0131955143 100755 --- a/storage/src/main/java/org/zstack/storage/volume/VolumeBase.java +++ b/storage/src/main/java/org/zstack/storage/volume/VolumeBase.java @@ -52,6 +52,8 @@ import org.zstack.storage.primary.EstimateVolumeTemplateSizeOnPrimaryStorageMsg; import org.zstack.storage.primary.EstimateVolumeTemplateSizeOnPrimaryStorageReply; import org.zstack.storage.primary.PrimaryStorageGlobalConfig; +import org.zstack.storage.encrypt.VolumeBackupEncryptionConversionExtensionHelper; +import org.zstack.storage.encrypt.VolumeEncryptionConversionHostLeaseHelper; import org.zstack.storage.encrypt.VolumeEncryptedResourceKeyBackend; import org.zstack.storage.encrypt.VolumeEncryptedSecretHelper; import org.zstack.storage.snapshot.group.VolumeSnapshotGroupOperationValidator; @@ -68,6 +70,7 @@ import javax.persistence.TypedQuery; import java.io.File; import java.util.*; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -113,6 +116,10 @@ public class VolumeBase extends AbstractVolume implements Volume { private VolumeEncryptedResourceKeyBackend volumeEncryptedResourceKeyBackend; @Autowired private VolumeEncryptedSecretHelper volumeEncryptedSecretHelper; + @Autowired + private VolumeEncryptionConversionHostLeaseHelper volumeEncryptionConversionHostLeaseHelper; + @Autowired + private VolumeBackupEncryptionConversionExtensionHelper volumeBackupEncryptionConversionExtensionHelper; public VolumeBase(VolumeVO vo) { self = vo; @@ -3256,10 +3263,12 @@ public void run(MessageReply reply) { } private void handle(ChangeVolumeEncryptionMsg msg) { + String conversionSyncSignature = + volumeEncryptionConversionHostLeaseHelper.makeQueueSignature(msg.getVolumeUuid(), syncThreadId); thdf.chainSubmit(new ChainTask(msg) { @Override public String getSyncSignature() { - return syncThreadId; + return conversionSyncSignature; } @Override @@ -3320,6 +3329,10 @@ private void changeVolumeEncryption(boolean targetEncrypted, ReturnValueCompleti String sourceSecretHostUuid = sourceEncrypted ? resolveVolumeSecretHostUuid(self) : null; String sourceSecretVmUuid = StringUtils.defaultIfBlank(self.getVmInstanceUuid(), self.getUuid()); AtomicReference conversionContext = new AtomicReference<>(); + AtomicReference> backupConversionContexts = + new AtomicReference<>(Collections.emptyList()); + AtomicBoolean encryptionConversionCommitted = new AtomicBoolean(false); + AtomicReference hostLease = new AtomicReference<>(); FlowChain chain = FlowChainBuilder.newShareFlowChain(); chain.setName(String.format("change-volume-%s-encryption-to-%s", self.getUuid(), targetEncrypted)); @@ -3342,6 +3355,21 @@ public void run(FlowTrigger trigger, Map data) { } }); + flow(new Flow() { + String __name__ = "reserve-volume-encryption-conversion-host"; + + @Override + public void run(FlowTrigger trigger, Map data) { + volumeEncryptionConversionHostLeaseHelper.reserve(self, trigger, hostLease); + } + + @Override + public void rollback(FlowRollback trigger, Map data) { + volumeEncryptionConversionHostLeaseHelper.release(hostLease.get()); + trigger.rollback(); + } + }); + flow(new Flow() { String __name__ = "prepare-target-volume-key"; @@ -3379,6 +3407,7 @@ public void run(FlowTrigger trigger, Map data) { cmsg.setVolume(VolumeInventory.valueOf(self)); cmsg.setTargetEncrypted(targetEncrypted); cmsg.setItems(conversionContext.get().items); + cmsg.setHostUuid(hostLease.get().getHostUuid()); bus.makeTargetServiceIdByResourceUuid(cmsg, PrimaryStorageConstant.SERVICE_ID, self.getPrimaryStorageUuid()); bus.send(cmsg, new CloudBusCallBack(trigger) { @Override @@ -3396,7 +3425,38 @@ public void run(MessageReply reply) { @Override public void rollback(FlowRollback trigger, Map data) { - deleteConvertedVolumeEncryptionBits(conversionContext.get().items); + if (!encryptionConversionCommitted.get()) { + deleteConvertedVolumeEncryptionBits(conversionContext.get().items); + } + trigger.rollback(); + } + }); + + flow(new Flow() { + String __name__ = "prepare-volume-backup-encryption-conversion"; + + @Override + public void run(FlowTrigger trigger, Map data) { + volumeBackupEncryptionConversionExtensionHelper.prepare(getSelfInventory(), targetEncrypted, + new ReturnValueCompletion>(trigger) { + @Override + public void success(List contexts) { + backupConversionContexts.set(contexts); + trigger.next(); + } + + @Override + public void fail(ErrorCode errorCode) { + trigger.fail(errorCode); + } + }); + } + + @Override + public void rollback(FlowRollback trigger, Map data) { + if (!encryptionConversionCommitted.get()) { + volumeBackupEncryptionConversionExtensionHelper.rollback(backupConversionContexts.get(), self.getUuid()); + } trigger.rollback(); } }); @@ -3407,9 +3467,26 @@ public void rollback(FlowRollback trigger, Map data) { @Override public void run(FlowTrigger trigger, Map data) { VolumeEncryptionConversionContext ctx = conversionContext.get(); - updateVolumeEncryptionConversionInDb(targetEncrypted, ctx.snapshots, ctx.oldAndNewInstallPaths, - (Map) data.get("actualSizes")); - refreshVO(); + Map actualSizes = (Map) data.get("actualSizes"); + List backupContexts = backupConversionContexts.get(); + new SQLBatch() { + @Override + protected void scripts() { + updateVolumeEncryptionConversionInDb(targetEncrypted, ctx.snapshots, ctx.oldAndNewInstallPaths, + actualSizes, backupContexts); + } + }.execute(); + encryptionConversionCommitted.set(true); + try { + refreshVO(); + } catch (RuntimeException e) { + logger.warn(String.format("failed to refresh volume[uuid:%s] after encryption conversion DB update: %s", + self.getUuid(), e.getMessage()), e); + VolumeVO latest = dbf.findByUuid(self.getUuid(), VolumeVO.class); + if (latest != null) { + self = latest; + } + } trigger.next(); } @@ -3419,6 +3496,16 @@ public void rollback(FlowRollback trigger, Map data) { } }); + flow(new NoRollbackFlow() { + String __name__ = "cleanup-converted-volume-backup-bits"; + + @Override + public void run(FlowTrigger trigger, Map data) { + volumeBackupEncryptionConversionExtensionHelper.cleanupCommitted(backupConversionContexts.get(), self.getUuid()); + trigger.next(); + } + }); + flow(new NoRollbackFlow() { String __name__ = "cleanup-old-libvirt-secret"; @@ -3475,11 +3562,13 @@ public void run(FlowTrigger trigger, Map data) { }).done(new FlowDoneHandler(completion) { @Override public void handle(Map data) { + volumeEncryptionConversionHostLeaseHelper.release(hostLease.get()); completion.success(getSelfInventory()); } }).error(new FlowErrorHandler(completion) { @Override public void handle(ErrorCode errCode, Map data) { + volumeEncryptionConversionHostLeaseHelper.release(hostLease.get()); completion.fail(errCode); } }).start(); @@ -3749,10 +3838,10 @@ private void deleteConvertedVolumeEncryptionBits(List snapshots, Map oldAndNewInstallPaths, - Map actualSizes) { + Map actualSizes, + List backupConversionContexts) { VolumeSnapshotReferenceUtils.handleVolumeInstallUrlChange(self.getUuid(), oldAndNewInstallPaths); UpdateQuery q = SQL.New(VolumeVO.class) @@ -3777,6 +3866,8 @@ private void updateVolumeEncryptionConversionInDb(boolean targetEncrypted, List< if (targetEncrypted && !convertedSnapshotInstallPaths.isEmpty()) { volumeEncryptedResourceKeyBackend.copyVolumeKeyRefToSnapshots(self.getUuid(), convertedSnapshotInstallPaths.keySet()); } + + volumeBackupEncryptionConversionExtensionHelper.commit(backupConversionContexts); } private void updateConvertedSnapshotPaths(Map snapshotInstallPaths, boolean targetEncrypted) {