Skip to content

Fix block commit wait when using libvirt events #13914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JoaoJandre wants to merge 2 commits into
base: main
Choose a base branch
from
+50 −19
Open

Fix block commit wait when using libvirt events #13914

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0f97674
Fix waiting for block commit
Aug 6, 2026
1a893ea
Fix concurrency issue
Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 21 additions & 2 deletions .../hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/BlockCommitListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)) {
Expand All @@ -55,17 +71,20 @@ public void onEvent(Domain domain, String diskPath, BlockJobType type, BlockJobS
switch (status) {
case COMPLETED:
result = null;
semaphore.release();
return;
case READY:
try {
logger.debug("Pivoting disk [{}] of VM [{}].", diskPath, vmName);
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();
}
}
}
19 changes: 11 additions & 8 deletions ...rvisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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" +
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
}

/**
Expand Down
27 changes: 18 additions & 9 deletions ...ors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/LibvirtComputingResourceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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";
Expand All @@ -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";
Expand All @@ -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";
Expand All @@ -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";
Expand All @@ -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";
Expand Down
Loading