diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/BlockCommitListener.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/BlockCommitListener.java index ab4513642efa..217feff0a4b6 100644 --- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/BlockCommitListener.java +++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/BlockCommitListener.java @@ -27,23 +27,39 @@ import org.libvirt.event.BlockJobStatus; import org.libvirt.event.BlockJobType; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + public class BlockCommitListener implements BlockJobListener { private String result; private String vmName; private Logger logger; private String logid; + private Semaphore semaphore; protected BlockCommitListener(String vmName, String logid) { this.vmName = vmName; this.logid = logid; - logger = LogManager.getLogger(getClass()); + this.logger = LogManager.getLogger(getClass()); + this.semaphore = new Semaphore(0); + this.result = String.format("Failed to block commit disk of VM [%s]. Libvirt did not launch an event for it.", vmName); } - protected String getResult() { + protected String getResult(int timeout) { + this.waitBlockCommit(timeout); return result; } + protected void waitBlockCommit(int timeout) { + try { + logger.debug("Trying to acquire result semaphore. If the correct event was not launched, will wait for [{}] seconds before giving up.", timeout); + this.semaphore.tryAcquire(timeout, TimeUnit.SECONDS); + } catch (InterruptedException ex) { + logger.error("Thread that was tracking the progress for the block commit job of vm {} was interrupted.", vmName, ex); + } + } + @Override public void onEvent(Domain domain, String diskPath, BlockJobType type, BlockJobStatus status) { if (!BlockJobType.COMMIT.equals(type) && !BlockJobType.ACTIVE_COMMIT.equals(type)) { @@ -55,6 +71,7 @@ public void onEvent(Domain domain, String diskPath, BlockJobType type, BlockJobS switch (status) { case COMPLETED: result = null; + semaphore.release(); return; case READY: try { @@ -62,10 +79,12 @@ public void onEvent(Domain domain, String diskPath, BlockJobType type, BlockJobS domain.blockJobAbort(diskPath, Domain.BlockJobAbortFlags.PIVOT); } catch (LibvirtException ex) { result = String.format("Failed to pivot disk due to [%s].", ex.getMessage()); + semaphore.release(); } return; default: result = String.format("Failed to block commit disk with status [%s].", status); + semaphore.release(); } } } diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java index 4281036d9456..a3e460944821 100644 --- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java +++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java @@ -6652,21 +6652,22 @@ protected void mergeSnapshotIntoBaseFileWithEventsAndConfigurableTimeout(Domain } BlockCommitListener blockCommitListener = getBlockCommitListener(vmName); + int remainingTimeout = 0; + String mergeResult = "Got an exception during block commit wait. Check earlier logs for more info."; try { vm.addBlockJobListener(blockCommitListener); logger.info("Starting block commit of QCOW2 delta [{}] of VM [{}]. Using parameters: diskLabel [{}]; baseFilePath [{}]; topFilePath [{}]; commitFlags [{}]", - snapshotName, - vmName, diskLabel, baseFilePath, topFilePath, commitFlags); + snapshotName, vmName, diskLabel, baseFilePath, topFilePath, commitFlags); vm.blockCommit(diskLabel, baseFilePath, topFilePath, 0, commitFlags); - checkBlockCommitProgress(vm, diskLabel, vmName, snapshotName, topFilePath, baseFilePath); + remainingTimeout = checkBlockCommitProgress(vm, diskLabel, vmName, snapshotName, topFilePath, baseFilePath); + mergeResult = blockCommitListener.getResult(remainingTimeout); } finally { vm.removeBlockJobListener(blockCommitListener); } - String mergeResult = blockCommitListener.getResult(); if (mergeResult != null) { String commitError = String.format("Failed the block commit of top file [%s] into base file [%s] for snapshot [%s] of VM [%s]. The job will be left running to avoid" + " data corruption, but ACS will return an error and volume [%s] will need to be normalized manually. If the commit involved the active image, the pivot will" + @@ -6731,7 +6732,7 @@ protected BlockCommitListener getBlockCommitListener(String vmName) { return new BlockCommitListener(vmName, ThreadContext.get("logcontextid")); } - protected void checkBlockCommitProgress(Domain vm, String diskLabel, String vmName, String snapshotName, String topFilePath, String baseFilePath) { + protected int checkBlockCommitProgress(Domain vm, String diskLabel, String vmName, String snapshotName, String topFilePath, String baseFilePath) { int timeout = qcow2DeltaMergeTimeout; DomainBlockJobInfo result; long lastCommittedBytes = 0; @@ -6752,12 +6753,12 @@ protected void checkBlockCommitProgress(Domain vm, String diskLabel, String vmNa result = vm.getBlockJobInfo(diskLabel, 0); } catch (LibvirtException ex) { logger.warn("Exception while getting block job info {}: [{}].", partialLog, ex.getMessage(), ex); - return; + return timeout; } if (result == null || result.type == 0 && result.end == 0 && result.cur == 0) { logger.debug("Block commit job {} has already finished.", partialLog); - return; + return timeout; } long currentCommittedBytes = result.cur; @@ -6767,7 +6768,9 @@ protected void checkBlockCommitProgress(Domain vm, String diskLabel, String vmNa lastCommittedBytes = currentCommittedBytes; endBytes = result.end; } - logger.warn("Block commit {} has timed out after waiting at least {} seconds. The progress of the operation was [{}] of [{}].", partialLog, qcow2DeltaMergeTimeout, lastCommittedBytes, endBytes); + logger.warn("Block commit {} has timed out after waiting at least {} seconds. The progress of the operation was [{}] of [{}].", partialLog, + qcow2DeltaMergeTimeout, lastCommittedBytes, endBytes); + return 0; } /** diff --git a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResourceTest.java b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResourceTest.java index 639f1dc2894c..216f3e0b193b 100644 --- a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResourceTest.java +++ b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResourceTest.java @@ -27,6 +27,7 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.nullable; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doReturn; @@ -6696,9 +6697,11 @@ public void mergeSnapshotIntoBaseFileTestActiveAndDeleteFlags() throws Exception threadContextMockedStatic.when(() -> ThreadContext.get(Mockito.anyString())).thenReturn("logid"); - Mockito.doNothing().when(domainMock).addBlockJobListener(Mockito.any()); + Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any()); + Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt()); + Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock); Mockito.doReturn(null).when(domainMock).getBlockJobInfo(Mockito.anyString(), Mockito.anyInt()); - Mockito.doNothing().when(domainMock).removeBlockJobListener(Mockito.any()); + Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock); String diskLabel = "vda"; String baseFilePath = "/file"; @@ -6723,8 +6726,10 @@ public void mergeSnapshotIntoBaseFileTestActiveFlag() throws Exception { threadContextMockedStatic.when(() -> ThreadContext.get(Mockito.anyString())).thenReturn("logid"); - Mockito.doNothing().when(domainMock).addBlockJobListener(Mockito.any()); - Mockito.doNothing().when(domainMock).removeBlockJobListener(Mockito.any()); + Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any()); + Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt()); + Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock); + Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock); Mockito.doNothing().when(libvirtComputingResourceSpy).manuallyDeleteUnusedSnapshotFile(Mockito.anyBoolean(), Mockito.anyString()); String diskLabel = "vda"; @@ -6748,9 +6753,11 @@ public void mergeSnapshotIntoBaseFileTestDeleteFlag() throws Exception { libvirtComputingResourceSpy.qcow2DeltaMergeTimeout = 10; libvirtUtilitiesHelperMockedStatic.when(() -> LibvirtUtilitiesHelper.isLibvirtSupportingFlagDeleteOnCommandVirshBlockcommit(Mockito.any())).thenReturn(true); threadContextMockedStatic.when(() -> ThreadContext.get(Mockito.anyString())).thenReturn("logid"); - Mockito.doNothing().when(domainMock).addBlockJobListener(Mockito.any()); + Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any()); + Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt()); + Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock); Mockito.doReturn(null).when(domainMock).getBlockJobInfo(Mockito.anyString(), Mockito.anyInt()); - Mockito.doNothing().when(domainMock).removeBlockJobListener(Mockito.any()); + Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock); Mockito.doNothing().when(libvirtComputingResourceSpy).manuallyDeleteUnusedSnapshotFile(Mockito.anyBoolean(), Mockito.anyString()); String diskLabel = "vda"; @@ -6774,9 +6781,11 @@ public void mergeSnapshotIntoBaseFileTestNoFlags() throws Exception { libvirtComputingResourceSpy.qcow2DeltaMergeTimeout = 10; libvirtUtilitiesHelperMockedStatic.when(() -> LibvirtUtilitiesHelper.isLibvirtSupportingFlagDeleteOnCommandVirshBlockcommit(Mockito.any())).thenReturn(false); threadContextMockedStatic.when(() -> ThreadContext.get(Mockito.anyString())).thenReturn("logid"); - Mockito.doNothing().when(domainMock).addBlockJobListener(Mockito.any()); + Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any()); + Mockito.doReturn(null).when(blockCommitListenerMock).getResult(anyInt()); + Mockito.doNothing().when(domainMock).addBlockJobListener(blockCommitListenerMock); Mockito.doReturn(null).when(domainMock).getBlockJobInfo(Mockito.anyString(), Mockito.anyInt()); - Mockito.doNothing().when(domainMock).removeBlockJobListener(Mockito.any()); + Mockito.doNothing().when(domainMock).removeBlockJobListener(blockCommitListenerMock); Mockito.doNothing().when(libvirtComputingResourceSpy).manuallyDeleteUnusedSnapshotFile(Mockito.anyBoolean(), Mockito.anyString()); String diskLabel = "vda"; @@ -6803,7 +6812,7 @@ public void mergeSnapshotIntoBaseFileTestMergeFailsThrowException() throws Excep Mockito.doNothing().when(domainMock).removeBlockJobListener(Mockito.any()); Mockito.doReturn(blockCommitListenerMock).when(libvirtComputingResourceSpy).getBlockCommitListener(Mockito.any()); - Mockito.doReturn("Failed").when(blockCommitListenerMock).getResult(); + Mockito.doReturn("Failed").when(blockCommitListenerMock).getResult(anyInt()); String diskLabel = "vda"; String baseFilePath = "/file";