Reconcile: add columns for resource id and type

weizhouapache · weizhouapache · commit e24876cc9815 · 2025-03-26T11:01:21.000+01:00
- add column resource_id and resource_type
- set resource_id and resource_type based on the srcData
- skip Migrating VMs with reconcile commands
- clean up volumes attached to Migrating vms without reconcile commands
- skip Migrating volumes with reconcile commands
- skip Migrating volumes attached to vms with reconcile commands
- cleanup volumes in Migrating/Creating with last_id
- add vmName to ReconcileMigrateAnswer
diff --git a/api/src/main/java/org/apache/cloudstack/command/ReconcileCommandService.java b/api/src/main/java/org/apache/cloudstack/command/ReconcileCommandService.java
@@ -20,6 +20,7 @@
 import com.cloud.agent.api.Answer;
 import com.cloud.agent.api.Command;
 import com.cloud.hypervisor.Hypervisor;
+import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.framework.config.ConfigKey;
 
 import java.util.Arrays;
@@ -59,4 +60,6 @@ public interface ReconcileCommandService {
     void updateReconcileCommandToInterruptedByManagementServerId(long managementServerId);
 
     void updateReconcileCommandToInterruptedByHostId(long hostId);
+
+    boolean isReconcileResourceNeeded(long resourceId, ApiCommandResourceType resourceType);
 }
diff --git a/core/src/main/java/org/apache/cloudstack/command/ReconcileMigrateAnswer.java b/core/src/main/java/org/apache/cloudstack/command/ReconcileMigrateAnswer.java
@@ -26,13 +26,15 @@
 public class ReconcileMigrateAnswer extends ReconcileAnswer {
 
     Long hostId;
+    String vmName;
     VirtualMachine.State vmState;
     List<String> vmDisks;
 
     public ReconcileMigrateAnswer() {
     }
 
-    public ReconcileMigrateAnswer(VirtualMachine.State vmState) {
+    public ReconcileMigrateAnswer(String vmName, VirtualMachine.State vmState) {
+        this.vmName = vmName;
         this.vmState = vmState;
     }
 
@@ -44,6 +46,10 @@ public void setHostId(Long hostId) {
         this.hostId = hostId;
     }
 
+    public String getVmName() {
+        return vmName;
+    }
+
     public VirtualMachine.State getVmState() {
         return vmState;
     }
diff --git a/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java b/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java
@@ -163,7 +163,5 @@ public interface VolumeDao extends GenericDao<VolumeVO, Long>, StateDao<Volume.S
 
     VolumeVO findOneByIScsiName(String iScsiName);
 
-    VolumeVO findByInstanceIdAndPath(Long instanceId, String path);
-
-    VolumeVO findByLastIdAndState(long lastVolumeId, Volume.State state);
+    VolumeVO findByLastIdAndState(long lastVolumeId, Volume.State...states);
 }
diff --git a/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java b/engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java
@@ -49,6 +49,7 @@
 import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.GenericSearchBuilder;
+import com.cloud.utils.db.QueryBuilder;
 import com.cloud.utils.db.SearchBuilder;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.SearchCriteria.Func;
@@ -398,7 +399,6 @@ public VolumeDaoImpl() {
         AllFieldsSearch.and("passphraseId", AllFieldsSearch.entity().getPassphraseId(), Op.EQ);
         AllFieldsSearch.and("iScsiName", AllFieldsSearch.entity().get_iScsiName(), Op.EQ);
         AllFieldsSearch.and("path", AllFieldsSearch.entity().getPath(), Op.EQ);
-        AllFieldsSearch.and("lastId", AllFieldsSearch.entity().getLastId(), Op.EQ);
         AllFieldsSearch.done();
 
         RootDiskStateSearch = createSearchBuilder();
@@ -914,18 +914,10 @@ public VolumeVO findOneByIScsiName(String iScsiName) {
     }
 
     @Override
-    public VolumeVO findByInstanceIdAndPath(Long instanceId, String path) {
-        SearchCriteria<VolumeVO> sc = AllFieldsSearch.create();
-        sc.setParameters("instanceId", instanceId);
-        sc.setParameters("path", path);
-        return findOneBy(sc);
-    }
-
-    @Override
-    public VolumeVO findByLastIdAndState(long lastVolumeId, State state) {
-        SearchCriteria<VolumeVO> sc = AllFieldsSearch.create();
-        sc.setParameters("lastId", lastVolumeId);
-        sc.setParameters("state", state);
-        return findOneBy(sc);
+    public VolumeVO findByLastIdAndState(long lastVolumeId, State ...states) {
+        QueryBuilder<VolumeVO> sc = QueryBuilder.create(VolumeVO.class);
+        sc.and(sc.entity().getLastId(), SearchCriteria.Op.EQ,  lastVolumeId);
+        sc.and(sc.entity().getState(), SearchCriteria.Op.IN,  (Object[]) states);
+        return sc.find();
     }
 }
diff --git a/engine/schema/src/main/java/org/apache/cloudstack/command/ReconcileCommandVO.java b/engine/schema/src/main/java/org/apache/cloudstack/command/ReconcileCommandVO.java
@@ -32,6 +32,7 @@
 import com.cloud.utils.db.GenericDao;
 
 import org.apache.cloudstack.acl.InfrastructureEntity;
+import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.api.InternalIdentity;
 
 @Entity
@@ -51,6 +52,12 @@ public class ReconcileCommandVO implements InfrastructureEntity, InternalIdentit
     @Column(name = "request_sequence")
     private long requestSequence;
 
+    @Column(name = "resource_id")
+    private long resourceId;
+
+    @Column(name = "resource_type")
+    private ApiCommandResourceType resourceType;
+
     @Column(name = "state_by_management")
     private Command.State stateByManagement;
 
@@ -111,6 +118,22 @@ public void setRequestSequence(long requestSequence) {
         this.requestSequence = requestSequence;
     }
 
+    public long getResourceId() {
+        return resourceId;
+    }
+
+    public void setResourceId(long resourceId) {
+        this.resourceId = resourceId;
+    }
+
+    public ApiCommandResourceType getResourceType() {
+        return resourceType;
+    }
+
+    public void setResourceType(ApiCommandResourceType resourceType) {
+        this.resourceType = resourceType;
+    }
+
     public Command.State getStateByManagement() {
         return stateByManagement;
     }
diff --git a/engine/schema/src/main/java/org/apache/cloudstack/command/dao/ReconcileCommandDao.java b/engine/schema/src/main/java/org/apache/cloudstack/command/dao/ReconcileCommandDao.java
@@ -20,6 +20,7 @@
 import com.cloud.agent.api.Command;
 import com.cloud.utils.db.GenericDao;
 
+import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.command.ReconcileCommandVO;
 
 import java.util.List;
@@ -39,4 +40,6 @@ public interface ReconcileCommandDao extends GenericDao<ReconcileCommandVO, Long
     void updateCommandsToInterruptedByManagementServerId(long managementServerId);
 
     void updateCommandsToInterruptedByHostId(long hostId);
+
+    List<ReconcileCommandVO> listByResourceIdAndTypeAndStates(long resourceId, ApiCommandResourceType resourceType, Command.State... states);
 }
diff --git a/engine/schema/src/main/java/org/apache/cloudstack/command/dao/ReconcileCommandDaoImpl.java b/engine/schema/src/main/java/org/apache/cloudstack/command/dao/ReconcileCommandDaoImpl.java
@@ -20,6 +20,7 @@
 import java.util.Date;
 import java.util.List;
 
+import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.command.ReconcileCommandVO;
 import org.springframework.stereotype.Component;
 
@@ -35,6 +36,7 @@
 public class ReconcileCommandDaoImpl extends GenericDaoBase<ReconcileCommandVO, Long> implements ReconcileCommandDao {
 
     final SearchBuilder<ReconcileCommandVO> updateCommandSearch;
+    final SearchBuilder<ReconcileCommandVO> resourceSearch;
 
     public ReconcileCommandDaoImpl() {
 
@@ -46,6 +48,12 @@ public ReconcileCommandDaoImpl() {
         updateCommandSearch.and("reqSequence", updateCommandSearch.entity().getRequestSequence(), SearchCriteria.Op.EQ);
         updateCommandSearch.and("commandName", updateCommandSearch.entity().getCommandName(), SearchCriteria.Op.EQ);
         updateCommandSearch.done();
+
+        resourceSearch = createSearchBuilder();
+        resourceSearch.and("resourceId", resourceSearch.entity().getResourceId(), SearchCriteria.Op.EQ);
+        resourceSearch.and("resourceType", resourceSearch.entity().getResourceType(), SearchCriteria.Op.EQ);
+        resourceSearch.and("stateByManagement", resourceSearch.entity().getStateByManagement(), SearchCriteria.Op.IN);
+        resourceSearch.done();
     }
 
     @Override
@@ -114,4 +122,13 @@ public void updateCommandsToInterruptedByHostId(long hostId) {
 
         update(vo, sc);
     }
+
+    @Override
+    public List<ReconcileCommandVO> listByResourceIdAndTypeAndStates(long resourceId, ApiCommandResourceType resourceType, Command.State... states) {
+        QueryBuilder<ReconcileCommandVO> sc = QueryBuilder.create(ReconcileCommandVO.class);
+        sc.and(sc.entity().getResourceId(), SearchCriteria.Op.EQ,  resourceId);
+        sc.and(sc.entity().getResourceType(), SearchCriteria.Op.EQ,  resourceType);
+        sc.and(sc.entity().getStateByManagement(), SearchCriteria.Op.IN,  (Object[]) states);
+        return sc.list();
+    }
 }
diff --git a/engine/schema/src/main/resources/META-INF/db/schema-42010to42100.sql b/engine/schema/src/main/resources/META-INF/db/schema-42010to42100.sql
@@ -44,6 +44,8 @@ CREATE TABLE IF NOT EXISTS `cloud`.`reconcile_commands` (
     `management_server_id` bigint unsigned NOT NULL COMMENT 'node id of the management server',
     `host_id` bigint unsigned NOT NULL COMMENT 'id of the host',
     `request_sequence` bigint unsigned NOT NULL COMMENT 'sequence of the request',
+    `resource_id` bigint unsigned DEFAULT NULL COMMENT 'id of the resource',
+    `resource_type` varchar(255) NOT NULL COMMENT 'type if the resource',
     `state_by_management` varchar(255) COMMENT 'state of the command updated by management server',
     `state_by_agent` varchar(255) COMMENT 'state of the command updated by cloudstack agent',
     `command_name` varchar(255) COMMENT 'name of the command',
diff --git a/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java b/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
@@ -71,8 +71,8 @@
 import com.cloud.network.dao.NetworkDao;
 import com.cloud.network.dao.NetworkVO;
 import com.cloud.storage.Snapshot;
-import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.Volume;
+import com.cloud.storage.VolumeVO;
 import com.cloud.storage.VolumeDetailVO;
 import com.cloud.storage.dao.SnapshotDao;
 import com.cloud.storage.dao.SnapshotDetailsDao;
@@ -169,6 +169,8 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
     private NetworkDao networkDao;
     @Inject
     private NetworkOrchestrationService networkOrchestrationService;
+    @Inject
+    private ReconcileCommandService reconcileCommandService;
 
     private volatile long _executionRunNumber = 1;
 
@@ -1199,10 +1201,22 @@ private boolean cleanupVolume(final long volumeId) {
             return true;
         }
         if (vol.getState().isTransitional()) {
-            if (Volume.State.Migrating.equals(vol.getState()) && Arrays.asList(ImageFormat.RAW, ImageFormat.QCOW2).contains(vol.getFormat())
-                    && ReconcileCommandService.ReconcileCommandsEnabled.value()) {
-                logger.debug("Skipping cleaning up Migrating volume: " + vol);
-                return true;
+            if (Volume.State.Migrating.equals(vol.getState())) {
+                if (ReconcileCommandService.ReconcileCommandsEnabled.value()) {
+                    if (reconcileCommandService.isReconcileResourceNeeded(volumeId, ApiCommandResourceType.Volume)) {
+                        logger.debug(String.format("Skipping cleaning up Migrating volume: %s, it will be reconciled", vol));
+                        return true;
+                    }
+                    if (vol.getInstanceId() != null && reconcileCommandService.isReconcileResourceNeeded(vol.getInstanceId(), ApiCommandResourceType.VirtualMachine)) {
+                        logger.debug(String.format("Skipping cleaning up Migrating volume: %s, the vm %s will be reconciled", vol, _vmInstanceDao.findById(vol.getInstanceId())));
+                        return true;
+                    }
+                }
+                VolumeVO destVolume = _volsDao.findByLastIdAndState(vol.getId(), Volume.State.Migrating, Volume.State.Creating);
+                if (destVolume != null) {
+                    logger.debug(String.format("Found destination volume of Migrating volume %s: %s", vol, destVolume));
+                    cleanupVolume(destVolume.getId());
+                }
             }
             logger.debug("Cleaning up volume with Id: " + volumeId);
             boolean status = vol.stateTransit(Volume.Event.OperationFailed);
@@ -1220,6 +1234,18 @@ private boolean cleanupVirtualMachine(final long vmId) throws Exception {
             return true;
         }
         if (vmInstanceVO.getState().isTransitional()) {
+            if (VirtualMachine.State.Migrating.equals(vmInstanceVO.getState())) {
+                if (ReconcileCommandService.ReconcileCommandsEnabled.value()
+                        && reconcileCommandService.isReconcileResourceNeeded(vmId, ApiCommandResourceType.VirtualMachine)) {
+                    logger.debug(String.format("Skipping cleaning up Instance %s, it will be reconciled", vmInstanceVO));
+                    return true;
+                }
+                logger.debug("Cleaning up volumes with instance Id: " + vmId);
+                List<VolumeVO> volumes = _volsDao.findByInstance(vmInstanceVO.getId());
+                for (VolumeVO volume : volumes) {
+                    cleanupVolume(volume.getId());
+                }
+            }
             logger.debug("Cleaning up Instance with Id: " + vmId);
             return virtualMachineManager.stateTransitTo(vmInstanceVO, VirtualMachine.Event.OperationFailed, vmInstanceVO.getHostId());
         }
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtReconcileCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtReconcileCommandWrapper.java
@@ -84,12 +84,12 @@ private ReconcileAnswer handle(final ReconcileMigrateCommand reconcileCommand, f
             if (VirtualMachine.State.Running.equals(state)) {
                 disks = getVmDiskPaths(libvirtComputingResource.getDisks(conn, vmName));
             }
-            answer = new ReconcileMigrateAnswer(state);
+            answer = new ReconcileMigrateAnswer(vmName, state);
             answer.setVmDisks(disks);
         } catch (LibvirtException e) {
             logger.debug(String.format("Failed to get state of VM %s, assume it is Stopped", vmName));
             VirtualMachine.State state = VirtualMachine.State.Stopped;
-            answer = new ReconcileMigrateAnswer(state);
+            answer = new ReconcileMigrateAnswer(vmName, state);
         }
         return answer;
     }
diff --git a/server/src/main/java/org/apache/cloudstack/command/ReconcileCommandServiceImpl.java b/server/src/main/java/org/apache/cloudstack/command/ReconcileCommandServiceImpl.java
@@ -196,6 +196,22 @@ private void persistReconcileCommand(Long hostId, Long requestSequence, Command
         reconcileCommandVO.setStateByManagement(State.CREATED);
         reconcileCommandVO.setHostId(hostId);
         reconcileCommandVO.setRequestSequence(requestSequence);
+        if (cmd instanceof CopyCommand) {
+            CopyCommand copyCmd = (CopyCommand)cmd;
+            DataTO srcData = copyCmd.getSrcTO();
+            if (srcData != null && srcData.getDataStore() instanceof PrimaryDataStoreTO) {
+                reconcileCommandVO.setResourceType(ApiCommandResourceType.Volume);
+                reconcileCommandVO.setResourceId(srcData.getId());
+            }
+        } else if (cmd instanceof MigrateCommand) {
+            MigrateCommand migrateCommand = (MigrateCommand)cmd;
+            reconcileCommandVO.setResourceType(ApiCommandResourceType.VirtualMachine);
+            reconcileCommandVO.setResourceId(migrateCommand.getVirtualMachine().getId());
+        } else if (cmd instanceof MigrateVolumeCommand) {
+            MigrateVolumeCommand migrateVolumeCommand = (MigrateVolumeCommand)cmd;
+            reconcileCommandVO.setResourceType(ApiCommandResourceType.Volume);
+            reconcileCommandVO.setResourceId(migrateVolumeCommand.getVolumeId());
+        }
         reconcileCommandDao.persist(reconcileCommandVO);
     }
 
@@ -745,6 +761,7 @@ private boolean processReconcileMigrateAnswer(MigrateCommand command, ReconcileM
                 logger.debug(String.format("Searching for volumes with last_id = %s", volumeVO.getId()));
                 VolumeVO destVolume = volumeDao.findByLastIdAndState(volumeVO.getId(), Volume.State.Migrating);
                 if (destVolume != null) {
+                    logger.debug(String.format("Found destination volume with last_id = %s: %s", volumeVO.getId(), destVolume));
                     Boolean isVolumeMigrated = isVolumeMigrated(diskPaths, volumeVO.getPath(), destVolume.getPath());
                     if (isVolumeMigrated == null) {
                         logger.debug(String.format("Unable to determine if source volume %s has been migrated to destination volume %s", volumeVO, destVolume));
@@ -1118,4 +1135,10 @@ private Pair<Volume.State, Volume.State> getVolumeStateOnSourceAndDestination(Da
         return new Pair<>(null, null);
     }
 
+    @Override
+    public boolean isReconcileResourceNeeded(long resourceId, ApiCommandResourceType resourceType) {
+        return !reconcileCommandDao.listByResourceIdAndTypeAndStates(resourceId, resourceType,
+                State.INTERRUPTED, State.TIMED_OUT, State.RECONCILE_RETRY, State.RECONCILING, State.RECONCILE_FAILED, State.CREATED)
+                .isEmpty();
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -26,13 +26,15 @@`
`26`	`26`	`public class ReconcileMigrateAnswer extends ReconcileAnswer {`
`27`	`27`
`28`	`28`	`Long hostId;`
	`29`	`+ String vmName;`
`29`	`30`	`VirtualMachine.State vmState;`
`30`	`31`	`List<String> vmDisks;`
`31`	`32`
`32`	`33`	`public ReconcileMigrateAnswer() {`
`33`	`34`	`}`
`34`	`35`
`35`		`- public ReconcileMigrateAnswer(VirtualMachine.State vmState) {`
	`36`	`+ public ReconcileMigrateAnswer(String vmName, VirtualMachine.State vmState) {`
	`37`	`+ this.vmName = vmName;`
`36`	`38`	`this.vmState = vmState;`
`37`	`39`	`}`
`38`	`40`
`@@ -44,6 +46,10 @@ public void setHostId(Long hostId) {`
`44`	`46`	`this.hostId = hostId;`
`45`	`47`	`}`
`46`	`48`
	`49`	`+ public String getVmName() {`
	`50`	`+ return vmName;`
	`51`	`+ }`
	`52`	`+`
`47`	`53`	`public VirtualMachine.State getVmState() {`
`48`	`54`	`return vmState;`
`49`	`55`	`}`
Original file line number	Diff line number	Diff line change
`@@ -163,7 +163,5 @@ public interface VolumeDao extends GenericDao<VolumeVO, Long>, StateDao<Volume.S`
`163`	`163`
`164`	`164`	`VolumeVO findOneByIScsiName(String iScsiName);`
`165`	`165`
`166`		`- VolumeVO findByInstanceIdAndPath(Long instanceId, String path);`
`167`		`-`
`168`		`- VolumeVO findByLastIdAndState(long lastVolumeId, Volume.State state);`
	`166`	`+ VolumeVO findByLastIdAndState(long lastVolumeId, Volume.State...states);`
`169`	`167`	`}`
Original file line number	Diff line number	Diff line change
`@@ -84,12 +84,12 @@ private ReconcileAnswer handle(final ReconcileMigrateCommand reconcileCommand, f`
`84`	`84`	`if (VirtualMachine.State.Running.equals(state)) {`
`85`	`85`	`disks = getVmDiskPaths(libvirtComputingResource.getDisks(conn, vmName));`
`86`	`86`	`}`
`87`		`- answer = new ReconcileMigrateAnswer(state);`
	`87`	`+ answer = new ReconcileMigrateAnswer(vmName, state);`
`88`	`88`	`answer.setVmDisks(disks);`
`89`	`89`	`} catch (LibvirtException e) {`
`90`	`90`	`logger.debug(String.format("Failed to get state of VM %s, assume it is Stopped", vmName));`
`91`	`91`	`VirtualMachine.State state = VirtualMachine.State.Stopped;`
`92`		`- answer = new ReconcileMigrateAnswer(state);`
	`92`	`+ answer = new ReconcileMigrateAnswer(vmName, state);`
`93`	`93`	`}`
`94`	`94`	`return answer;`
`95`	`95`	`}`