Fix KVM incremental volume snapshot migration between secondary storages

core/src/main/java/com/cloud/agent/api/MigrateBetweenSecondaryStoragesCommandAnswer.java

@@ -0,0 +1,41 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.cloud.agent.api;
+
+import com.cloud.utils.Pair;
+
+import java.util.List;
+
+public class MigrateBetweenSecondaryStoragesCommandAnswer extends Answer {
+
+    List<Pair<Long, String>> migratedResourcesIdAndCheckpointPath;
+
+    public MigrateBetweenSecondaryStoragesCommandAnswer() {
+    }
+
+    public MigrateBetweenSecondaryStoragesCommandAnswer(MigrateSnapshotsBetweenSecondaryStoragesCommand cmd, boolean success, String result, List<Pair<Long, String>> migratedResourcesIdAndCheckpointPath) {
+        super(cmd, success, result);
+        this.migratedResourcesIdAndCheckpointPath = migratedResourcesIdAndCheckpointPath;
+    }
+
+    public List<Pair<Long, String>> getMigratedResources() {
+        return migratedResourcesIdAndCheckpointPath;
+    }
+}

core/src/main/java/com/cloud/agent/api/MigrateSnapshotsBetweenSecondaryStoragesCommand.java

@@ -0,0 +1,64 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.cloud.agent.api;
+
+import com.cloud.agent.api.to.DataStoreTO;
+import com.cloud.agent.api.to.DataTO;
+
+import java.util.List;
+import java.util.Set;
+
+public class MigrateSnapshotsBetweenSecondaryStoragesCommand extends Command {
+
+    DataStoreTO srcDataStore;
+    DataStoreTO destDataStore;
+    List<DataTO> snapshotChain;
+    Set<Long> snapshotsIdToMigrate;
+
+    public MigrateSnapshotsBetweenSecondaryStoragesCommand() {
+    }
+
+    public MigrateSnapshotsBetweenSecondaryStoragesCommand(List<DataTO> snapshotChain, DataStoreTO srcDataStore, DataStoreTO destDataStore, Set<Long> snapshotsIdToMigrate) {
+        this.srcDataStore = srcDataStore;
+        this.destDataStore = destDataStore;
+        this.snapshotChain = snapshotChain;
+        this.snapshotsIdToMigrate = snapshotsIdToMigrate;
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return false;
+    }
+
+    public List<DataTO> getSnapshotChain() {
+        return snapshotChain;
+    }
+
+    public Set<Long> getSnapshotsIdToMigrate() {
+        return snapshotsIdToMigrate;
+    }
+
+    public DataStoreTO getSrcDataStore() {
+        return srcDataStore;
+    }
+
+    public DataStoreTO getDestDataStore() {
+        return destDataStore;
+    }
+}

engine/api/src/main/java/org/apache/cloudstack/engine/subsystem/api/storage/SnapshotInfo.java

@@ -30,6 +30,8 @@ public interface SnapshotInfo extends DataObject, Snapshot {
 
     SnapshotInfo getParent();
 
+    List<SnapshotInfo> getParents();
+
     String getPath();
 
     DataStore getImageStore();
@@ -40,6 +42,8 @@ public interface SnapshotInfo extends DataObject, Snapshot {
 
     List<SnapshotInfo> getChildren();
 
+    List<SnapshotInfo> getChildAndGrandchildren();
+
     VolumeInfo getBaseVolume();
 
     void addPayload(Object data);

engine/components-api/src/main/java/com/cloud/storage/StorageManager.java

@@ -228,6 +228,9 @@ public interface StorageManager extends StorageService {
             ConfigKey.Scope.Global,
             null);
 
+    ConfigKey<Integer> AgentMaxDataMigrationWaitTime = new ConfigKey<>("Advanced", Integer.class, "agent.max.data.migration.wait.time", "3600",
+            "The maximum time (in seconds) that the secondary storage data migration command sent to the KVM Agent will be executed before a timeout occurs.", true, ConfigKey.Scope.Cluster);
+
     /**
      * should we execute in sequence not involving any storages?
      * @return true if commands should execute in sequence

engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/DataMigrationUtility.java

@@ -22,10 +22,13 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import javax.inject.Inject;
 
@@ -92,7 +95,7 @@ public class DataMigrationUtility {
      *  "Ready" "Allocated", "Destroying", "Destroyed", "Failed". If this is the case, and if the migration policy is complete,
      *  the migration is terminated.
      */
-    public boolean filesReadyToMigrate(Long srcDataStoreId, List<TemplateDataStoreVO> templates, List<SnapshotDataStoreVO> snapshots, List<VolumeDataStoreVO> volumes) {
+    public boolean filesReadyToMigrate(List<TemplateDataStoreVO> templates, List<SnapshotDataStoreVO> snapshots, List<VolumeDataStoreVO> volumes) {
         State[] validStates = {State.Ready, State.Allocated, State.Destroying, State.Destroyed, State.Failed};
         boolean isReady = true;
         for (TemplateDataStoreVO template : templates) {
@@ -114,7 +117,8 @@ private boolean filesReadyToMigrate(Long srcDataStoreId) {
         List<TemplateDataStoreVO> templates = templateDataStoreDao.listByStoreId(srcDataStoreId);
         List<SnapshotDataStoreVO> snapshots = snapshotDataStoreDao.listByStoreId(srcDataStoreId, DataStoreRole.Image);
         List<VolumeDataStoreVO> volumes = volumeDataStoreDao.listByStoreId(srcDataStoreId);
-        return filesReadyToMigrate(srcDataStoreId, templates, snapshots, volumes);
+
+        return filesReadyToMigrate(templates, snapshots, volumes);
     }
 
     protected void checkIfCompleteMigrationPossible(ImageStoreService.MigrationPolicy policy, Long srcDataStoreId) {
@@ -163,29 +167,40 @@ public int compare(Map.Entry<Long, Pair<Long, Long>> e1, Map.Entry<Long, Pair<Lo
     }
 
     protected List<DataObject> getSortedValidSourcesList(DataStore srcDataStore, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains,
-            Map<DataObject, Pair<List<TemplateInfo>, Long>> childTemplates, List<TemplateDataStoreVO> templates, List<SnapshotDataStoreVO> snapshots) {
+            Map<DataObject, Pair<List<TemplateInfo>, Long>> childTemplates, List<TemplateDataStoreVO> templates, List<SnapshotDataStoreVO> snapshots, Set<Long> snapshotIdsToMigrate) {
         List<DataObject> files = new ArrayList<>();
 
         files.addAll(getAllReadyTemplates(srcDataStore, childTemplates, templates));
-        files.addAll(getAllReadySnapshotsAndChains(srcDataStore, snapshotChains, snapshots));
+        files.addAll(getAllReadySnapshotsAndChains(srcDataStore, snapshotChains, snapshots, snapshotIdsToMigrate));
 
         files = sortFilesOnSize(files, snapshotChains);
 
         return files;
     }
 
     protected List<DataObject> getSortedValidSourcesList(DataStore srcDataStore, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains,
-            Map<DataObject, Pair<List<TemplateInfo>, Long>> childTemplates) {
+            Map<DataObject, Pair<List<TemplateInfo>, Long>> childTemplates, Set<Long> snapshotIdsToMigrate) {
         List<DataObject> files = new ArrayList<>();
         files.addAll(getAllReadyTemplates(srcDataStore, childTemplates));
-        files.addAll(getAllReadySnapshotsAndChains(srcDataStore, snapshotChains));
+        files.addAll(getAllReadySnapshotsAndChains(srcDataStore, snapshotChains, snapshotIdsToMigrate));
         files.addAll(getAllReadyVolumes(srcDataStore));
 
         files = sortFilesOnSize(files, snapshotChains);
 
         return files;
     }
 
+    private List<SnapshotInfo> createKvmIncrementalSnapshotChain(SnapshotDataStoreVO snapshot) {
+        List<SnapshotInfo> chain = new LinkedList<>();
+        SnapshotInfo snapshotInfo = snapshotFactory.getSnapshot(snapshot.getSnapshotId(), snapshot.getDataStoreId(), snapshot.getRole());
+
+        chain.addAll(snapshotInfo.getParents());
+        chain.add(snapshotInfo);
+        chain.addAll(snapshotInfo.getChildAndGrandchildren());
+
+        return chain;
+    }
+
     protected List<DataObject> sortFilesOnSize(List<DataObject> files, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains) {
         Collections.sort(files, new Comparator<DataObject>() {
             @Override
@@ -261,16 +276,24 @@ protected boolean shouldMigrateTemplate(TemplateDataStoreVO template, VMTemplate
      * for each parent snapshot and the cumulative size of the chain - this is done to ensure that all the snapshots in a chain
      * are migrated to the same datastore
      */
-    protected List<DataObject> getAllReadySnapshotsAndChains(DataStore srcDataStore, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains, List<SnapshotDataStoreVO> snapshots) {
+    protected List<DataObject> getAllReadySnapshotsAndChains(DataStore srcDataStore, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains, List<SnapshotDataStoreVO> snapshots, Set<Long> snapshotIdsToMigrate) {
         List<SnapshotInfo> files = new LinkedList<>();
+        Set<Long> snapshotIdsAlreadyInChain = new HashSet<>();
         for (SnapshotDataStoreVO snapshot : snapshots) {
             SnapshotVO snapshotVO = snapshotDao.findById(snapshot.getSnapshotId());
             if (snapshot.getState() == ObjectInDataStoreStateMachine.State.Ready &&
-                    snapshotVO != null && snapshotVO.getHypervisorType() != Hypervisor.HypervisorType.Simulator
-                    && snapshot.getParentSnapshotId() == 0 ) {
-                SnapshotInfo snap = snapshotFactory.getSnapshot(snapshotVO.getSnapshotId(), snapshot.getDataStoreId(), snapshot.getRole());
-                if (snap != null) {
-                    files.add(snap);
+                    snapshotVO != null && snapshotVO.getHypervisorType() != Hypervisor.HypervisorType.Simulator) {
+                if (snapshotVO.getHypervisorType() == Hypervisor.HypervisorType.KVM && snapshot.getKvmCheckpointPath() != null && !snapshotIdsAlreadyInChain.contains(snapshotVO.getId())) {
+                    List<SnapshotInfo> kvmIncrementalSnapshotChain = createKvmIncrementalSnapshotChain(snapshot);
+                    SnapshotInfo parent = kvmIncrementalSnapshotChain.get(0);
+                    snapshotIdsAlreadyInChain.addAll(kvmIncrementalSnapshotChain.stream().map(DataObject::getId).collect(Collectors.toSet()));
+                    snapshotChains.put(parent, new Pair<>(kvmIncrementalSnapshotChain, getTotalChainSize(kvmIncrementalSnapshotChain.stream().filter(snap -> snapshotIdsToMigrate.contains(snap.getId())).collect(Collectors.toList()))));
+                    files.add(parent);
+                } else if (snapshot.getParentSnapshotId() == 0) {
+                    SnapshotInfo snap = snapshotFactory.getSnapshot(snapshotVO.getSnapshotId(), snapshot.getDataStoreId(), snapshot.getRole());
+                    if (snap != null) {
+                        files.add(snap);
+                    }
                 }
             }
         }
@@ -285,15 +308,16 @@ protected List<DataObject> getAllReadySnapshotsAndChains(DataStore srcDataStore,
                     chain.addAll(children);
                 }
             }
-            snapshotChains.put(parent, new Pair<List<SnapshotInfo>, Long>(chain, getTotalChainSize(chain)));
+            snapshotChains.putIfAbsent(parent, new Pair<>(chain, getTotalChainSize(chain)));
         }
 
         return (List<DataObject>) (List<?>) files;
     }
 
-    protected List<DataObject> getAllReadySnapshotsAndChains(DataStore srcDataStore, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains) {
+    protected List<DataObject> getAllReadySnapshotsAndChains(DataStore srcDataStore, Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChains, Set<Long> snapshotIdsToMigrate) {
         List<SnapshotDataStoreVO> snapshots = snapshotDataStoreDao.listByStoreId(srcDataStore.getId(), DataStoreRole.Image);
-        return getAllReadySnapshotsAndChains(srcDataStore, snapshotChains, snapshots);
+        snapshotIdsToMigrate.addAll(snapshots.stream().map(SnapshotDataStoreVO::getSnapshotId).collect(Collectors.toSet()));
+        return getAllReadySnapshotsAndChains(srcDataStore, snapshotChains, snapshots, snapshotIdsToMigrate);
     }
 
     protected Long getTotalChainSize(List<? extends DataObject> chain) {