Correção de _deadlock_ na migração de recursos entre armazenamentos secundários

Author: Fabrício Duarte

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

@@ -273,12 +273,22 @@ public int compare(DataObject o1, DataObject o2) {
 
     protected List<DataObject> getAllReadyTemplates(DataStore srcDataStore, Map<DataObject, Pair<List<TemplateInfo>, Long>> childTemplates, List<TemplateDataStoreVO> templates) {
         List<TemplateInfo> files = new LinkedList<>();
+        Set<Long> idsForMigration = new HashSet<>();
+
         for (TemplateDataStoreVO template : templates) {
-            VMTemplateVO templateVO = templateDao.findById(template.getTemplateId());
-            if (shouldMigrateTemplate(template, templateVO)) {
-                files.add(templateFactory.getTemplate(template.getTemplateId(), srcDataStore));
+            long templateId = template.getTemplateId();
+            if (idsForMigration.contains(templateId)) {
+                logger.warn("Template store reference [{}] is duplicated; not considering it for migration.", template);
+                continue;
+            }
+            VMTemplateVO templateVO = templateDao.findById(templateId);
+            if (!shouldMigrateTemplate(template, templateVO)) {
+                continue;
             }
+            files.add(templateFactory.getTemplate(template.getTemplateId(), srcDataStore));
+            idsForMigration.add(templateId);
         }
+
         for (TemplateInfo template: files) {
             List<VMTemplateVO> children = templateDao.listByParentTemplatetId(template.getId());
             List<TemplateInfo> temps = new ArrayList<>();
@@ -288,6 +298,7 @@ protected List<DataObject> getAllReadyTemplates(DataStore srcDataStore, Map<Data
             }
             childTemplates.put(template, new Pair<>(temps, getTotalChainSize(temps)));
         }
+
         return (List<DataObject>) (List<?>) files;
     }
 
@@ -382,13 +393,29 @@ protected Long getTotalChainSize(List<? extends DataObject> chain) {
 
     protected List<DataObject> getAllReadyVolumes(DataStore srcDataStore, List<VolumeDataStoreVO> volumes) {
         List<DataObject> files = new LinkedList<>();
+        Set<Long> idsForMigration = new HashSet<>();
+
         for (VolumeDataStoreVO volume : volumes) {
-            if (volume.getState() == ObjectInDataStoreStateMachine.State.Ready) {
-                VolumeInfo volumeInfo = volumeFactory.getVolume(volume.getVolumeId(), srcDataStore);
-                if (volumeInfo != null && volumeInfo.getHypervisorType() != Hypervisor.HypervisorType.Simulator) {
-                    files.add(volumeInfo);
-                }
+            long volumeId = volume.getVolumeId();
+            if (idsForMigration.contains(volumeId)) {
+                logger.warn("Volume store reference [{}] is duplicated; not considering it for migration.", volume);
+                continue;
+            }
+            if (volume.getState() != ObjectInDataStoreStateMachine.State.Ready) {
+                logger.debug("Not migrating volume [{}] because its state is not ready.", volume);
+                continue;
+            }
+            VolumeInfo volumeInfo = volumeFactory.getVolume(volume.getVolumeId(), srcDataStore);
+            if (volumeInfo == null) {
+                logger.debug("Not migrating volume [{}] because we could not get its information.", volume);
+                continue;
+            }
+            if (volumeInfo.getHypervisorType() == Hypervisor.HypervisorType.Simulator) {
+                logger.debug("Not migrating volume [{}] because its hypervisor type is simulator.", volume);
+                continue;
             }
+            files.add(volumeInfo);
+            idsForMigration.add(volumeId);
         }
         return files;
     }

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

@@ -93,9 +93,12 @@
 import com.cloud.utils.Pair;
 import com.cloud.utils.component.ManagerBase;
 import com.cloud.utils.exception.CloudRuntimeException;
+import org.apache.logging.log4j.ThreadContext;
 
 public class StorageOrchestrator extends ManagerBase implements StorageOrchestrationService, Configurable {
 
+    private static final String LOGCONTEXTID = "logcontextid";
+
     @Inject
     SnapshotDataStoreDao snapshotDataStoreDao;
     @Inject
@@ -137,6 +140,8 @@ public class StorageOrchestrator extends ManagerBase implements StorageOrchestra
     Integer numConcurrentCopyTasksPerSSVM = 2;
 
     private final Map<Long, ThreadPoolExecutor> zoneExecutorMap = new ConcurrentHashMap<>();
+    private final Map<Long, Integer> zonePendingWorkCountMap = new ConcurrentHashMap<>();
+
     private final Map<Long, ExecutorService> zoneKvmIncrementalExecutorMap = new ConcurrentHashMap<>();
 
     @Override
@@ -513,12 +518,18 @@ private void createMigrateDataTask(DataObject chosenFileForMigration, Map<DataOb
     }
 
     protected <T> Future<T> submit(Long zoneId, Callable<T> task) {
-        if (!zoneExecutorMap.containsKey(zoneId)) {
-            zoneExecutorMap.put(zoneId, new ThreadPoolExecutor(numConcurrentCopyTasksPerSSVM, numConcurrentCopyTasksPerSSVM,
-                    30, TimeUnit.MINUTES, new MigrateBlockingQueue<>(numConcurrentCopyTasksPerSSVM)));
+        ThreadPoolExecutor executor;
+        synchronized (this) {
+            if (!zoneExecutorMap.containsKey(zoneId)) {
+                zoneExecutorMap.put(zoneId, new ThreadPoolExecutor(numConcurrentCopyTasksPerSSVM, numConcurrentCopyTasksPerSSVM,
+                        30, TimeUnit.MINUTES, new MigrateBlockingQueue<>(numConcurrentCopyTasksPerSSVM)));
+                zonePendingWorkCountMap.put(zoneId, 0);
+            }
+            zonePendingWorkCountMap.merge(zoneId, 1, Integer::sum);
+            scaleExecutorIfNecessary(zoneId);
+            executor = zoneExecutorMap.get(zoneId);
         }
-        scaleExecutorIfNecessary(zoneId);
-        return zoneExecutorMap.get(zoneId).submit(task);
+        return executor.submit(task);
     }
 
     protected <T> Future<T> submitKvmIncrementalMigration(Long zoneId, Callable<T> task) {
@@ -540,24 +551,23 @@ protected void scaleExecutorIfNecessary(Long zoneId) {
         }
     }
 
-    protected void tryCleaningUpExecutor(Long zoneId) {
+    protected synchronized void tryCleaningUpExecutor(Long zoneId) {
         if (!zoneExecutorMap.containsKey(zoneId)) {
             logger.debug("No executor exists for zone [{}].", zoneId);
             return;
         }
 
-        ThreadPoolExecutor executor = zoneExecutorMap.get(zoneId);
-        synchronized (executor) {
-            int activeTasks = executor.getActiveCount();
-            if (activeTasks > 1) {
-                logger.debug("Not cleaning executor of zone [{}] yet, as there are [{}] active tasks.", zoneId, activeTasks);
-                return;
-            }
-
-            logger.debug("Cleaning executor of zone [{}].", zoneId);
-            zoneExecutorMap.remove(zoneId);
-            executor.shutdown();
+        zonePendingWorkCountMap.merge(zoneId, -1, Integer::sum);
+        Integer pendingWorkCount = zonePendingWorkCountMap.get(zoneId);
+        if (pendingWorkCount > 0) {
+            logger.debug("Not cleaning executor of zone [{}] yet, as there is [{}] pending work.", zoneId, pendingWorkCount);
+            return;
         }
+
+        logger.debug("Cleaning executor of zone [{}].", zoneId);
+        ThreadPoolExecutor executor = zoneExecutorMap.get(zoneId);
+        zoneExecutorMap.remove(zoneId);
+        executor.shutdown();
     }
 
     private MigrationResponse handleResponse(List<Future<DataObjectResult>> futures, MigrationPolicy migrationPolicy, String message, boolean success) {
@@ -733,10 +743,13 @@ private class MigrateDataTask implements Callable<DataObjectResult> {
         private DataStore destDataStore;
         private Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChain;
         private Map<DataObject, Pair<List<TemplateInfo>, Long>> templateChain;
+        private String logid;
+
         public MigrateDataTask(DataObject file, DataStore srcDataStore, DataStore destDataStore) {
             this.file = file;
             this.srcDataStore = srcDataStore;
             this.destDataStore = destDataStore;
+            this.logid = ThreadContext.get(LOGCONTEXTID);
         }
 
         public void setSnapshotChains(Map<DataObject, Pair<List<SnapshotInfo>, Long>> snapshotChain) {
@@ -758,6 +771,8 @@ public DataObject getFile() {
 
         @Override
         public DataObjectResult call() {
+            ThreadContext.put(LOGCONTEXTID, logid);
+
             DataObjectResult result;
             AsyncCallFuture<DataObjectResult> future = secStgSrv.migrateData(file, srcDataStore, destDataStore, snapshotChain, templateChain);
             try {
@@ -767,22 +782,29 @@ public DataObjectResult call() {
                 result = new DataObjectResult(file);
                 result.setResult(e.toString());
             }
+
             tryCleaningUpExecutor(srcDataStore.getScope().getScopeId());
+            ThreadContext.clearAll();
+
             return result;
         }
     }
 
     private class CopyTemplateTask implements Callable<TemplateApiResult> {
         private TemplateInfo sourceTmpl;
         private DataStore destStore;
+        private String logid;
 
         public CopyTemplateTask(TemplateInfo sourceTmpl, DataStore destStore) {
             this.sourceTmpl = sourceTmpl;
             this.destStore = destStore;
+            this.logid = ThreadContext.get(LOGCONTEXTID);
         }
 
         @Override
         public TemplateApiResult call() {
+            ThreadContext.put(LOGCONTEXTID, logid);
+
             TemplateApiResult result;
             AsyncCallFuture<TemplateApiResult> future = templateService.copyTemplateToImageStore(sourceTmpl, destStore);
             try {
@@ -793,7 +815,10 @@ public TemplateApiResult call() {
                 result = new TemplateApiResult(sourceTmpl);
                 result.setResult(e.getMessage());
             }
+
             tryCleaningUpExecutor(destStore.getScope().getScopeId());
+            ThreadContext.clearAll();
+
             return result;
         }
     }

engine/storage/image/src/main/java/org/apache/cloudstack/storage/image/TemplateServiceImpl.java

@@ -58,6 +58,7 @@
 import org.apache.cloudstack.framework.messagebus.MessageBus;
 import org.apache.cloudstack.framework.messagebus.PublishScope;
 import org.apache.cloudstack.storage.command.CommandResult;
+import org.apache.cloudstack.storage.command.CopyCmdAnswer;
 import org.apache.cloudstack.storage.command.DeleteCommand;
 import org.apache.cloudstack.storage.datastore.DataObjectManager;
 import org.apache.cloudstack.storage.datastore.ObjectInDataStoreManager;
@@ -677,6 +678,15 @@ public AsyncCallFuture<TemplateApiResult> copyTemplateToImageStore(DataObject so
         TemplateOpContext<TemplateApiResult> context = new TemplateOpContext<>(null, destTmpl, future);
         AsyncCallbackDispatcher<TemplateServiceImpl, CopyCommandResult> caller = AsyncCallbackDispatcher.create(this);
         caller.setCallback(caller.getTarget().copyTemplateToImageStoreCallback(null, null)).setContext(context);
+
+        if (source.getDataStore().getId() == destStore.getId()) {
+            logger.debug("Destination image store [{}] is the same as the origin; returning success.");
+            CopyCmdAnswer answer = new CopyCmdAnswer(source.getTO());
+            CopyCommandResult result = new CopyCommandResult("", answer);
+            caller.complete(result);
+            return future;
+        }
+
         _motionSrv.copyAsync(sourceTmpl, destTmpl, caller);
         return future;
     }