| // 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 org.apache.cloudstack.command; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Date; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CompletionService; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorCompletionService; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ScheduledThreadPoolExecutor; |
| import java.util.concurrent.TimeUnit; |
| import java.util.stream.Collectors; |
| import javax.inject.Inject; |
| import javax.naming.ConfigurationException; |
| |
| import com.cloud.agent.AgentManager; |
| import com.cloud.agent.api.Answer; |
| import com.cloud.agent.api.Command; |
| import com.cloud.agent.api.Command.State; |
| import com.cloud.agent.api.MigrateCommand; |
| import com.cloud.agent.api.PingAnswer; |
| import com.cloud.agent.api.PingCommand; |
| import com.cloud.agent.api.storage.MigrateVolumeCommand; |
| import com.cloud.agent.api.to.DataObjectType; |
| import com.cloud.agent.api.to.DataTO; |
| import com.cloud.cluster.ManagementServerHostVO; |
| import com.cloud.cluster.dao.ManagementServerHostDao; |
| import com.cloud.host.HostVO; |
| import com.cloud.host.Status; |
| import com.cloud.host.dao.HostDao; |
| import com.cloud.storage.Volume; |
| import com.cloud.storage.VolumeApiService; |
| import com.cloud.storage.VolumeVO; |
| import com.cloud.storage.dao.VolumeDao; |
| import com.cloud.user.Account; |
| import com.cloud.user.AccountManager; |
| import com.cloud.utils.Pair; |
| import com.cloud.utils.component.ManagerBase; |
| import com.cloud.utils.concurrency.NamedThreadFactory; |
| import com.cloud.utils.db.Filter; |
| import com.cloud.utils.db.GlobalLock; |
| import com.cloud.utils.exception.CloudRuntimeException; |
| import com.cloud.vm.VMInstanceVO; |
| import com.cloud.vm.VirtualMachine; |
| import com.cloud.vm.dao.VMInstanceDao; |
| |
| import org.apache.cloudstack.api.ApiCommandResourceType; |
| import org.apache.cloudstack.command.dao.ReconcileCommandDao; |
| import org.apache.cloudstack.engine.orchestration.service.VolumeOrchestrationService; |
| import org.apache.cloudstack.engine.subsystem.api.storage.DataStore; |
| import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager; |
| import org.apache.cloudstack.engine.subsystem.api.storage.EndPoint; |
| import org.apache.cloudstack.engine.subsystem.api.storage.EndPointSelector; |
| import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine; |
| import org.apache.cloudstack.framework.config.ConfigKey; |
| import org.apache.cloudstack.framework.config.Configurable; |
| import org.apache.cloudstack.managed.context.ManagedContextRunnable; |
| import org.apache.cloudstack.storage.command.CommandResult; |
| import org.apache.cloudstack.storage.command.CopyCommand; |
| import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao; |
| import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO; |
| import org.apache.cloudstack.storage.to.PrimaryDataStoreTO; |
| import org.apache.cloudstack.storage.volume.VolumeOnStorageTO; |
| import org.apache.cloudstack.utils.identity.ManagementServerNode; |
| import org.apache.commons.collections.CollectionUtils; |
| import org.apache.commons.collections.MapUtils; |
| import org.apache.commons.lang3.StringUtils; |
| import org.springframework.stereotype.Component; |
| |
| @Component |
| public class ReconcileCommandServiceImpl extends ManagerBase implements ReconcileCommandService, Configurable { |
| |
| final static long ManagementServerId = ManagementServerNode.getManagementServerId(); |
| final static int GracePeriod = 10 * 60; // 10 minutes |
| private boolean _reconcileCommandsEnabled = false; |
| |
| private ScheduledExecutorService reconcileCommandsExecutor; |
| private ExecutorService reconcileCommandTaskExecutor; |
| CompletionService<ReconcileCommandResult> completionService; |
| |
| @Inject |
| ReconcileCommandDao reconcileCommandDao; |
| @Inject |
| ManagementServerHostDao managementServerHostDao; |
| @Inject |
| HostDao hostDao; |
| @Inject |
| AgentManager agentManager; |
| @Inject |
| VMInstanceDao vmInstanceDao; |
| @Inject |
| VolumeDao volumeDao; |
| @Inject |
| EndPointSelector endPointSelector; |
| @Inject |
| DataStoreManager dataStoreManager; |
| @Inject |
| VolumeDataStoreDao volumeDataStoreDao; |
| @Inject |
| VolumeOrchestrationService volumeManager; |
| @Inject |
| private VolumeApiService volumeApiService; |
| @Inject |
| private AccountManager accountManager; |
| |
| @Override |
| public boolean configure(final String name, final Map<String, Object> params) throws ConfigurationException { |
| _reconcileCommandsEnabled = ReconcileCommandsEnabled.value(); |
| if (_reconcileCommandsEnabled) { |
| // create thread pool and blocking queue |
| final int workersCount = ReconcileCommandsWorkers.value(); |
| reconcileCommandTaskExecutor = Executors.newFixedThreadPool(workersCount, new NamedThreadFactory("Reconcile-Command-Task-Executor")); |
| final BlockingQueue<Future<ReconcileCommandResult>> queue = new LinkedBlockingQueue<>(workersCount); |
| completionService = new ExecutorCompletionService<>(reconcileCommandTaskExecutor, queue); |
| |
| reconcileCommandsExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("Reconcile-Commands-Worker")); |
| reconcileCommandsExecutor.scheduleWithFixedDelay(new ReconcileCommandsWorker(), |
| ReconcileCommandsInterval.value(), ReconcileCommandsInterval.value(), TimeUnit.SECONDS); |
| } |
| |
| return true; |
| } |
| |
| @Override |
| public boolean stop() { |
| if (reconcileCommandsExecutor != null) { |
| reconcileCommandsExecutor.shutdownNow(); |
| } |
| if (reconcileCommandTaskExecutor != null) { |
| reconcileCommandTaskExecutor.shutdownNow(); |
| } |
| if (_reconcileCommandsEnabled) { |
| reconcileCommandDao.updateCommandsToInterruptedByManagementServerId(ManagementServerId); |
| } |
| |
| return true; |
| } |
| |
| @Override |
| public String getConfigComponentName() { |
| return ReconcileCommandService.class.getName(); |
| } |
| |
| @Override |
| public ConfigKey<?>[] getConfigKeys() { |
| return new ConfigKey<?>[]{ ReconcileCommandsEnabled, ReconcileCommandsInterval, ReconcileCommandsWorkers, ReconcileCommandsMaxAttempts }; |
| } |
| |
| @Override |
| public void persistReconcileCommands(Long hostId, Long requestSequence, Command[] commands) { |
| if (!_reconcileCommandsEnabled) { |
| return; |
| } |
| HostVO host = hostDao.findById(hostId); |
| if (host == null || !SupportedHypervisorTypes.contains(host.getHypervisorType())) { |
| return; |
| } |
| for (Command cmd : commands) { |
| if (cmd.isReconcile()) { |
| persistReconcileCommand(hostId, requestSequence, cmd); |
| } |
| } |
| } |
| |
| private void persistReconcileCommand(Long hostId, Long requestSequence, Command cmd) { |
| ReconcileCommandVO reconcileCommandVO = new ReconcileCommandVO(); |
| reconcileCommandVO.setManagementServerId(ManagementServerId); |
| reconcileCommandVO.setCommandInfo(CommandInfo.GSON.toJson(cmd)); |
| reconcileCommandVO.setCommandName(cmd.toString()); |
| reconcileCommandVO.setCreated(new Date()); |
| reconcileCommandVO.setUpdated(new Date()); |
| 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); |
| } |
| |
| @Override |
| public boolean updateReconcileCommand(long requestSeq, Command command, Answer answer, State newStateByManagement, State newStateByAgent) { |
| String commandKey = getCommandKey(requestSeq, command); |
| logger.debug(String.format("Updating reconcile command %s with answer %s and new states %s-%s", commandKey, answer, newStateByManagement, newStateByAgent)); |
| ReconcileCommandVO reconcileCommandVO = reconcileCommandDao.findCommand(requestSeq, command.toString()); |
| if (reconcileCommandVO == null) { |
| logger.debug(String.format("Skipped updating reconcile command %s due to no record is found in DB", commandKey)); |
| return false; |
| } |
| boolean updated = false; |
| if (newStateByManagement != null) { |
| if (State.RECONCILE_RETRY.equals(newStateByManagement)) { |
| if (State.RECONCILING.equals(reconcileCommandVO.getStateByManagement())) { |
| reconcileCommandVO.setStateByManagement(newStateByManagement); |
| updated = true; |
| } else { |
| logger.debug(String.format("Skipping the update of state by management of command %s from %s to %s", commandKey, reconcileCommandVO.getStateByManagement(), newStateByManagement)); |
| } |
| } else if (!newStateByManagement.equals(reconcileCommandVO.getStateByManagement())) { |
| reconcileCommandVO.setStateByManagement(newStateByManagement); |
| updated = true; |
| } |
| if (State.RECONCILE_FAILED.equals(newStateByManagement)) { |
| reconcileCommandVO.setRetryCount(reconcileCommandVO.getRetryCount() + 1); |
| updated = true; |
| } |
| if (ManagementServerId != ManagementServerNode.getManagementServerId()) { |
| reconcileCommandVO.setManagementServerId(ManagementServerId); |
| updated = true; |
| } |
| } |
| if (newStateByAgent != null) { |
| if (!newStateByAgent.equals(reconcileCommandVO.getStateByAgent())) { |
| reconcileCommandVO.setStateByAgent(newStateByAgent); |
| updated = true; |
| } |
| } |
| String commandInfo = CommandInfo.GSON.toJson(command); |
| if (!commandInfo.equals(reconcileCommandVO.getCommandInfo())) { |
| reconcileCommandVO.setCommandInfo(commandInfo); |
| updated = true; |
| } |
| if (answer != null && (reconcileCommandVO.getAnswerName() == null || answer instanceof ReconcileAnswer |
| || reconcileCommandVO.getAnswerName().equals(answer.toString()))) { |
| reconcileCommandVO.setAnswerName(answer.toString()); |
| reconcileCommandVO.setAnswerInfo(CommandInfo.GSON.toJson(answer)); |
| updated = true; |
| } |
| if (updated) { |
| reconcileCommandVO.setUpdated(new Date()); |
| reconcileCommandDao.update(reconcileCommandVO.getId(), reconcileCommandVO); |
| } |
| return true; |
| } |
| |
| private String getCommandKey(long requestSeq, Command command) { |
| return requestSeq + "-" + command; |
| } |
| |
| private class ReconcileCommandsWorker extends ManagedContextRunnable { |
| @Override |
| protected void runInContext() { |
| GlobalLock gcLock = GlobalLock.getInternLock("Reconcile.Commands.Lock"); |
| try { |
| if (gcLock.lock(3)) { |
| try { |
| reallyRun(); |
| } finally { |
| gcLock.unlock(); |
| } |
| } |
| } finally { |
| gcLock.releaseRef(); |
| } |
| } |
| |
| private List<ReconcileCommandVO> getReconcileCommands() { |
| ManagementServerHostVO msHost = managementServerHostDao.findOneInUpState(new Filter(ManagementServerHostVO.class, "id", true, 0L, 1L)); |
| if (msHost == null || msHost.getMsid() != ManagementServerId) { |
| return new ArrayList<>(); |
| } |
| return reconcileCommandDao.listByState(State.INTERRUPTED, State.TIMED_OUT, State.RECONCILE_RETRY, State.RECONCILING, State.RECONCILE_FAILED, State.CREATED); |
| } |
| |
| public void reallyRun() { |
| List<ReconcileCommandVO> reconcileCommands = getReconcileCommands(); |
| logger.debug(String.format("Reconciling %s command(s) ...", reconcileCommands.size())); |
| for (ReconcileCommandVO reconcileCommand : reconcileCommands) { |
| ReconcileCommandTask task = new ReconcileCommandTask(reconcileCommand); |
| completionService.submit(task); |
| } |
| for (int i = 0; i < reconcileCommands.size(); i++) { |
| try { |
| Future<ReconcileCommandResult> future = completionService.take(); |
| ReconcileCommandResult result = future.get(); |
| long requestSequence = result.getRequestSequence(); |
| Command command = result.getCommand(); |
| ReconcileAnswer answer = result.getAnswer(); |
| String commandKey = getCommandKey(requestSequence, command); |
| if (result.isFailed()) { |
| throw new CloudRuntimeException(String.format("Failed to reconcile command %s due to: %s", commandKey, result.getResult())); |
| } |
| if (answer != null && answer.getResult()) { |
| logger.debug(String.format("Command %s has been reconciled with answer %s", commandKey, answer)); |
| if (result.isReconciled()) { |
| if (processReconcileAnswer(requestSequence, command, answer)) { |
| updateReconcileCommand(requestSequence, result.getCommand(), answer, State.RECONCILED, null); |
| reconcileCommandDao.removeCommand(requestSequence, result.getCommand().toString(), null); |
| } else { |
| updateReconcileCommand(requestSequence, result.getCommand(), answer, State.RECONCILE_RETRY, null); |
| } |
| } else { |
| updateReconcileCommand(requestSequence, result.getCommand(), answer, State.RECONCILE_FAILED, null); |
| } |
| } else if (result.isReconciled()) { |
| logger.info(String.format("Command %s is reconciled but answer is null, skipping the reconciliation", commandKey)); |
| updateReconcileCommand(requestSequence, result.getCommand(), answer, State.RECONCILE_SKIPPED, null); |
| reconcileCommandDao.removeCommand(requestSequence, result.getCommand().toString(), null); |
| } else { |
| logger.info(String.format("Command %s is not reconciled, will retry", commandKey)); |
| updateReconcileCommand(requestSequence, result.getCommand(), answer, State.RECONCILE_RETRY, null); |
| } |
| } catch (InterruptedException | ExecutionException e) { |
| logger.error(String.format("Failed to reconcile command due to: %s", e.getMessage()), e); |
| throw new CloudRuntimeException("Failed to reconcile command"); |
| } |
| } |
| } |
| } |
| |
| |
| public static class ReconcileCommandResult extends CommandResult { |
| long requestSequence; |
| Command command; |
| ReconcileAnswer answer; |
| boolean isReconciled; |
| |
| public ReconcileCommandResult(long requestSequence, Command command, ReconcileAnswer answer, boolean isReconciled) { |
| super(); |
| this.requestSequence = requestSequence; |
| this.command = command; |
| this.answer = answer; |
| this.isReconciled = isReconciled; |
| } |
| |
| public long getRequestSequence() { |
| return requestSequence; |
| } |
| |
| public Command getCommand() { |
| return command; |
| } |
| |
| public ReconcileAnswer getAnswer() { |
| return answer; |
| } |
| |
| public boolean isReconciled() { |
| return isReconciled; |
| } |
| } |
| |
| protected class ReconcileCommandTask implements Callable<ReconcileCommandResult> { |
| long requestSequence; |
| Command command; |
| State stateByManagement; |
| State stateByAgent; |
| Long hostId; |
| Long retryCount; |
| ReconcileCommandVO reconcileCommand; |
| |
| public ReconcileCommandTask(ReconcileCommandVO reconcileCommand) { |
| this.requestSequence = reconcileCommand.getRequestSequence(); |
| this.stateByManagement = reconcileCommand.getStateByManagement(); |
| this.stateByAgent = reconcileCommand.getStateByAgent(); |
| this.hostId = reconcileCommand.getHostId(); |
| this.retryCount = reconcileCommand.getRetryCount(); |
| this.reconcileCommand = reconcileCommand; |
| this.command = ReconcileCommandUtils.parseCommandInfo(reconcileCommand.getCommandName(), reconcileCommand.getCommandInfo()); |
| } |
| |
| @Override |
| public ReconcileCommandResult call() { |
| String commandKey = getCommandKey(requestSequence, command); |
| HostVO host = hostDao.findByIdIncludingRemoved(hostId); |
| assert host != null; |
| if (!SupportedHypervisorTypes.contains(host.getHypervisorType())) { |
| return new ReconcileCommandResult(requestSequence, command, null, false); |
| } |
| |
| logger.debug(String.format("Reconciling command %s with state %s-%s", commandKey, stateByManagement, stateByAgent)); |
| |
| if (State.TIMED_OUT.equals(stateByManagement)) { |
| logger.debug(String.format("The command %s timed out on management server. Reconciling ...", commandKey)); |
| return reconcile(reconcileCommand); |
| } else if (Arrays.asList(State.INTERRUPTED, State.RECONCILE_RETRY).contains(stateByManagement)) { |
| logger.debug(String.format("The command %s is %s on management server. Reconciling ...", commandKey, stateByManagement)); |
| return reconcile(reconcileCommand); |
| } else if (State.RECONCILING.equals(stateByManagement)) { |
| Date now = new Date(); |
| if (reconcileCommand.getUpdated() != null && reconcileCommand.getUpdated().getTime() > now.getTime() - GracePeriod * 1000) { |
| logger.debug(String.format("The command %s is being reconciled, skipping and wait for next run", commandKey)); |
| } else { |
| logger.debug(String.format("The command %s is %s, the state seems out of date, updating to RECONCILE_READY", commandKey, stateByManagement)); |
| reconcileCommand = reconcileCommandDao.findById(reconcileCommand.getId()); |
| reconcileCommand.setStateByManagement(State.RECONCILE_RETRY); |
| reconcileCommandDao.update(reconcileCommand.getId(), reconcileCommand); |
| } |
| } else if (State.RECONCILE_FAILED.equals(stateByManagement)) { |
| if (retryCount != null && retryCount <= ReconcileCommandsMaxAttempts.value()) { |
| logger.debug(String.format("The command %s has been reconciled %s times, retrying", commandKey, retryCount)); |
| return reconcile(reconcileCommand); |
| } else { |
| logger.debug(String.format("The command %s has been reconciled %s times, skipping", commandKey, retryCount)); |
| return new ReconcileCommandResult(requestSequence, command, null, true); |
| } |
| } else if (State.RECONCILED.equals(stateByManagement)) { |
| logger.debug(String.format("The command %s has been reconciled, skipping", commandKey)); |
| } else if (stateByAgent == null) { // state by management is CREATED |
| logger.debug(String.format("Skipping the reconciliation of command %s, because the state by agent is null", commandKey)); |
| } else if (Arrays.asList(State.STARTED, State.PROCESSING, State.PROCESSING_IN_BACKEND).contains(stateByAgent)) { |
| if (host.getRemoved() == null && Status.Up.equals(host.getStatus())) { |
| logger.debug(String.format("Skipping the reconciliation of command %s, because the host %s is Up, the command may be still in processing", commandKey, host)); |
| } else if (host.getRemoved() != null) { |
| logger.debug(String.format("The host %s has been removed on %s, Reconciling command %s ...", host, host.getRemoved(), commandKey)); |
| return reconcile(reconcileCommand); |
| } else { |
| logger.debug(String.format("The host %s is in %s state, Reconciling command %s ...", host, host.getStatus(), commandKey)); |
| return reconcile(reconcileCommand); |
| } |
| } else if (Arrays.asList(State.COMPLETED, State.FAILED).contains(stateByAgent)) { |
| Date now = new Date(); |
| if (reconcileCommand.getUpdated() != null && reconcileCommand.getUpdated().getTime() > now.getTime() - GracePeriod * 1000) { |
| logger.debug(String.format("The command %s is %s on host %s, it seems the answer is not processed by any management server. Skipping ...", commandKey, stateByAgent, host)); |
| } else { |
| logger.debug(String.format("The command %s is %s on host %s, it seems the answer is not processed by any management server. Reconciling ...", commandKey, stateByAgent, host)); |
| return reconcile(reconcileCommand); |
| } |
| } else if (Arrays.asList(State.INTERRUPTED, State.DANGLED_IN_BACKEND).contains(stateByAgent)) { |
| logger.debug(String.format("The command %s is %s on host %s, the cloudstack agent might has been restarted. Reconciling ...", commandKey, stateByAgent, host)); |
| return reconcile(reconcileCommand); |
| } |
| |
| return new ReconcileCommandResult(requestSequence, command, null, false); |
| } |
| } |
| |
| protected ReconcileCommandResult reconcile(ReconcileCommandVO reconcileCommandVO) { |
| Command command = ReconcileCommandUtils.parseCommandInfo(reconcileCommandVO.getCommandName(), reconcileCommandVO.getCommandInfo()); |
| |
| if (!preReconcileCheck(command)) { |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| |
| if (command instanceof MigrateCommand) { |
| updateReconcileCommand(reconcileCommandVO.getRequestSequence(), command, null, State.RECONCILING, null); |
| return reconcile(reconcileCommandVO, (MigrateCommand) command); |
| } else if (command instanceof CopyCommand) { |
| updateReconcileCommand(reconcileCommandVO.getRequestSequence(), command, null, State.RECONCILING, null); |
| return reconcile(reconcileCommandVO, (CopyCommand) command); |
| } else if (command instanceof MigrateVolumeCommand) { |
| updateReconcileCommand(reconcileCommandVO.getRequestSequence(), command, null, State.RECONCILING, null); |
| return reconcile(reconcileCommandVO, (MigrateVolumeCommand) command); |
| } else { |
| logger.error(String.format("Unsupported reconcile command %s ", command)); |
| } |
| |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| |
| boolean preReconcileCheck(Command command) { |
| if (command instanceof MigrateCommand) { |
| MigrateCommand migrateCommand = (MigrateCommand) command; |
| Long vmId = migrateCommand.getVirtualMachine().getId(); |
| VMInstanceVO vm = vmInstanceDao.findById(vmId); |
| if (vm == null) { |
| logger.debug(String.format("Skipping reconciliation of command %s as vm %s has been removed", command, vm)); |
| return false; |
| } |
| if (MapUtils.isEmpty(migrateCommand.getMigrateStorage())) { |
| logger.debug(String.format("Skipping reconciliation of command %s as the migration does not migrate volumes of vm %s", command, vm)); |
| return false; |
| } |
| List<VolumeVO> volumes = volumeDao.findByInstance(vmId); |
| for (VolumeVO volume : volumes) { |
| if (Volume.State.Migrating.equals(volume.getState())) { |
| return true; |
| } |
| } |
| logger.debug(String.format("Skipping reconciliation of command %s as the vm %s does not have volume in Migrating state", command, vm)); |
| return false; |
| } else if (command instanceof CopyCommand) { |
| DataTO srcData = ((CopyCommand) command).getSrcTO(); |
| DataTO destData = ((CopyCommand) command).getDestTO(); |
| if (srcData != null && srcData.getDataStore() instanceof PrimaryDataStoreTO) { |
| VolumeVO volumeVO = volumeDao.findById(srcData.getId()); |
| if (volumeVO == null || !Arrays.asList(Volume.State.Migrating, Volume.State.Ready).contains(volumeVO.getState())) { |
| logger.debug(String.format("Skipping reconciliation of command %s as source volume %s is removed or not Migrating or Ready", command, volumeVO)); |
| return false; |
| } |
| } |
| if (destData != null && destData.getDataStore() instanceof PrimaryDataStoreTO) { |
| VolumeVO volumeVO = volumeDao.findById(destData.getId()); |
| if (volumeVO == null || !Arrays.asList(Volume.State.Migrating, Volume.State.Ready, Volume.State.Creating).contains(volumeVO.getState())) { |
| logger.debug(String.format("Skipping reconciliation of command %s as destination volume %s is removed or not Migrating or Ready or Creating", command, volumeVO)); |
| return false; |
| } |
| } |
| } else if (command instanceof MigrateVolumeCommand) { |
| DataTO srcData = ((MigrateVolumeCommand) command).getSrcData(); |
| DataTO destData = ((MigrateVolumeCommand) command).getDestData(); |
| if (srcData == null || destData == null) { |
| logger.debug(String.format("Skipping reconciliation of command %s as the source volume (%s) or destination volume (%s) is NULL", command, srcData, destData)); |
| return false; |
| } |
| if (srcData.getId() != destData.getId()) { |
| logger.debug(String.format("Skipping reconciliation of command %s as the source volume (id: %s) and destination volume (id: %s) have different ID", command, srcData.getId(), destData.getId())); |
| return false; |
| } |
| VolumeVO volumeVO = volumeDao.findById(srcData.getId()); |
| if (volumeVO == null) { |
| logger.debug(String.format("Skipping reconciliation of command %s as the volume (id: %s) has been removed", command, srcData.getId())); |
| return false; |
| } else if (!Volume.State.Migrating.equals(volumeVO.getState())) { |
| logger.debug(String.format("Skipping reconciliation of command %s as the volume %s (state: %s) is not Migrating state", command, volumeVO, volumeVO.getState())); |
| return false; |
| } |
| if (!volumeVO.getPath().equals(destData.getPath())) { |
| logger.debug(String.format("Skipping reconciliation of command %s as the volume path (%s) is not same as destination volume path (%s)", command, volumeVO.getPath(), destData.getPath())); |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| private ReconcileCommandResult reconcile(ReconcileCommandVO reconcileCommandVO, MigrateCommand command) { |
| Long vmId = command.getVirtualMachine().getId(); |
| VMInstanceVO vm = vmInstanceDao.findById(vmId); |
| if (vm == null) { |
| logger.debug(String.format("VM (id: %s) has been removed", vmId)); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| if (!VirtualMachine.State.Running.equals(vm.getState())) { |
| logger.debug(String.format("VM %s (state: %s) is not Running state, wait for next run", vm, vm.getState())); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, false); |
| } |
| |
| ReconcileMigrateAnswer reconcileMigrateAnswer = new ReconcileMigrateAnswer(); |
| reconcileMigrateAnswer.setResourceType(ApiCommandResourceType.VirtualMachine); |
| reconcileMigrateAnswer.setResourceId(vmId); |
| |
| Long hostId = vm.getHostId(); |
| HostVO sourceHost = hostDao.findById(hostId); |
| if (sourceHost != null && sourceHost.getStatus() == Status.Up) { |
| ReconcileMigrateCommand reconcileMigrateCommand = new ReconcileMigrateCommand(command.getVmName()); |
| Answer reconcileAnswer = agentManager.easySend(sourceHost.getId(), reconcileMigrateCommand); |
| reconcileMigrateAnswer.setHostId(sourceHost.getId()); |
| if (reconcileAnswer instanceof ReconcileMigrateAnswer) { |
| reconcileMigrateAnswer.setVmState(((ReconcileMigrateAnswer) reconcileAnswer).getVmState()); |
| reconcileMigrateAnswer.setVmDisks(((ReconcileMigrateAnswer) reconcileAnswer).getVmDisks()); |
| } |
| } |
| |
| boolean isReconciled = (reconcileMigrateAnswer.getVmState() != null); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, reconcileMigrateAnswer, isReconciled); |
| } |
| |
| private EndPoint getEndpoint(DataTO srcData, DataTO destData) { |
| EndPoint endPoint = null; |
| if (srcData.getDataStore() instanceof PrimaryDataStoreTO) { |
| PrimaryDataStoreTO srcDataStore = (PrimaryDataStoreTO) srcData.getDataStore(); |
| if (srcDataStore.isManaged()) { |
| return null; |
| } |
| DataStore store = dataStoreManager.getPrimaryDataStore(srcDataStore.getId()); |
| endPoint = endPointSelector.select(store); |
| } else if (destData != null && destData.getDataStore() instanceof PrimaryDataStoreTO) { |
| PrimaryDataStoreTO destDataStore = (PrimaryDataStoreTO) destData.getDataStore(); |
| if (destDataStore.isManaged()) { |
| return null; |
| } |
| DataStore store = dataStoreManager.getPrimaryDataStore(destDataStore.getId()); |
| endPoint = endPointSelector.select(store); |
| } |
| return endPoint; |
| } |
| |
| private ReconcileCommandResult reconcile(ReconcileCommandVO reconcileCommandVO, CopyCommand command) { |
| DataTO srcData = command.getSrcTO(); |
| DataTO destData = command.getDestTO(); |
| if (srcData == null || destData == null) { |
| logger.debug(String.format("Unable to reconcile command %s with srcData %s and destData %s", command, srcData, destData)); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, false); |
| } |
| Long hostId = reconcileCommandVO.getHostId(); |
| HostVO host = hostDao.findById(hostId); |
| if (host == null || !Status.Up.equals(host.getStatus())) { |
| EndPoint endPoint = getEndpoint(srcData, destData); |
| if (endPoint == null) { |
| logger.debug(String.format("Unable to reconcile command %s with srcData %s and destData %s as endpoint is null", command, srcData, destData)); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, false); |
| } |
| host = hostDao.findById(endPoint.getId()); |
| } |
| |
| // Send reconcileCommand to the host |
| logger.info(String.format("Reconciling command %s via host %s", command, host)); |
| ReconcileCopyCommand reconcileCommand = new ReconcileCopyCommand(srcData, destData, command.getOptions(), command.getOptions2()); |
| Answer reconcileAnswer = agentManager.easySend(host.getId(), reconcileCommand); |
| if (!(reconcileAnswer instanceof ReconcileAnswer)) { |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, (ReconcileAnswer) reconcileAnswer, true); |
| } |
| |
| private ReconcileCommandResult reconcile(ReconcileCommandVO reconcileCommandVO, MigrateVolumeCommand command) { |
| DataTO srcData = command.getSrcData(); |
| DataTO destData = command.getDestData(); |
| if (srcData == null || destData == null) { |
| logger.debug(String.format("Unable to reconcile command %s with srcData %s and destData %s", command, srcData, destData)); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| VolumeVO volume = volumeDao.findById(srcData.getId()); |
| if (volume == null) { |
| logger.debug(String.format("Unable to reconcile command %s with removed volume (id: %s)", command, srcData.getId())); |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| |
| Long hostId = reconcileCommandVO.getHostId(); |
| HostVO host = hostDao.findById(hostId); |
| if (host == null || !Status.Up.equals(host.getStatus())) { |
| EndPoint endPoint = getEndpoint(srcData, destData); |
| if (endPoint == null) { |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, false); |
| } |
| host = hostDao.findById(endPoint.getId()); |
| } |
| |
| // Send reconcileCommand to the host |
| logger.info(String.format("Reconciling command %s via host %s", command, host)); |
| ReconcileMigrateVolumeCommand reconcileCommand = new ReconcileMigrateVolumeCommand(srcData, destData); |
| if (volume.getInstanceId() != null) { |
| VMInstanceVO vmInstance = vmInstanceDao.findById(volume.getInstanceId()); |
| if (vmInstance != null) { |
| reconcileCommand.setVmName(vmInstance.getInstanceName()); |
| } |
| } |
| Answer reconcileAnswer = agentManager.easySend(host.getId(), reconcileCommand); |
| if (!(reconcileAnswer instanceof ReconcileAnswer)) { |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, null, true); |
| } |
| return new ReconcileCommandResult(reconcileCommandVO.getRequestSequence(), command, (ReconcileAnswer) reconcileAnswer, true); |
| } |
| |
| @Override |
| public void processCommand(Command pingCommand, Answer pingAnswer) { |
| if (pingCommand instanceof PingCommand && pingAnswer instanceof PingAnswer) { |
| CommandInfo[] commandInfos = ((PingCommand) pingCommand).getCommandInfos(); |
| for (CommandInfo commandInfo : commandInfos) { |
| processCommandInfo(commandInfo, (PingAnswer) pingAnswer); |
| } |
| } |
| } |
| |
| private void processCommandInfo(CommandInfo commandInfo, PingAnswer pingAnswer) { |
| Command parsedCommand = ReconcileCommandUtils.parseCommandInfo(commandInfo); |
| Answer parsedAnswer = ReconcileCommandUtils.parseAnswerFromCommandInfo(commandInfo); |
| if (parsedCommand != null && parsedCommand.isReconcile()) { |
| if (updateReconcileCommand(commandInfo.getRequestSeq(), parsedCommand, parsedAnswer, null, commandInfo.getState())) { |
| pingAnswer.addReconcileCommand(getCommandKey(commandInfo.getRequestSeq(), parsedCommand)); |
| } |
| } |
| } |
| |
| @Override |
| public void processAnswers(long requestSeq, Command[] commands, Answer[] answers) { |
| if (commands.length != answers.length) { |
| logger.error(String.format("Incorrect number of commands (%s) and answers (%s)", commands.length, answers.length)); |
| } |
| for (int i = 0; i < answers.length; i++) { |
| Command command = commands[i]; |
| Answer answer = answers[i]; |
| if (command.isReconcile() && answer.getResult()) { |
| reconcileCommandDao.removeCommand(requestSeq, command.toString(), State.COMPLETED); |
| } |
| } |
| } |
| |
| @Override |
| public void updateReconcileCommandToInterruptedByManagementServerId(long managementServerId) { |
| logger.debug("Updating reconcile command to interrupted by management server id " + managementServerId); |
| reconcileCommandDao.updateCommandsToInterruptedByManagementServerId(managementServerId); |
| } |
| |
| @Override |
| public void updateReconcileCommandToInterruptedByHostId(long hostId) { |
| logger.debug("Updating reconcile command to interrupted by host id " + hostId); |
| reconcileCommandDao.updateCommandsToInterruptedByHostId(hostId); |
| } |
| |
| private boolean processReconcileAnswer(long requestSequence, Command cmd, ReconcileAnswer reconcileAnswer) { |
| if (cmd instanceof MigrateCommand && reconcileAnswer instanceof ReconcileMigrateAnswer) { |
| MigrateCommand command = (MigrateCommand) cmd; |
| ReconcileMigrateAnswer answer = (ReconcileMigrateAnswer) reconcileAnswer; |
| return processReconcileMigrateAnswer(command, answer); |
| } else if (cmd instanceof CopyCommand && reconcileAnswer instanceof ReconcileCopyAnswer) { |
| CopyCommand command = (CopyCommand) cmd; |
| ReconcileCopyAnswer answer = (ReconcileCopyAnswer) reconcileAnswer; |
| return processReconcileCopyAnswer(requestSequence, command, answer); |
| } else if (cmd instanceof MigrateVolumeCommand && reconcileAnswer instanceof ReconcileMigrateVolumeAnswer) { |
| MigrateVolumeCommand command = (MigrateVolumeCommand) cmd; |
| ReconcileMigrateVolumeAnswer answer = (ReconcileMigrateVolumeAnswer) reconcileAnswer; |
| return processReconcileMigrateVolumeAnswer(requestSequence, command, answer); |
| } |
| return true; |
| } |
| |
| private boolean processReconcileMigrateAnswer(MigrateCommand command, ReconcileMigrateAnswer reconcileAnswer) { |
| Long vmId = command.getVirtualMachine().getId(); |
| VMInstanceVO vm = vmInstanceDao.findById(vmId); |
| if (vm == null) { |
| logger.debug(String.format("VM (id: %s) has been removed", vmId)); |
| return true; |
| } |
| if (!VirtualMachine.State.Running.equals(vm.getState())) { |
| logger.debug(String.format("VM %s (state: %s) is not Running state", vm, vm.getState())); |
| return true; |
| } |
| Map<String, MigrateCommand.MigrateDiskInfo> migrateDiskInfoMap = command.getMigrateStorage(); |
| if (MapUtils.isEmpty(migrateDiskInfoMap)) { |
| return true; |
| } |
| if (!VirtualMachine.State.Running.equals(reconcileAnswer.getVmState())) { |
| logger.debug(String.format("VM %s is %s state on the host, will retry", vm, vm.getState())); |
| return false; |
| } |
| List<String> diskPaths = reconcileAnswer.getVmDisks(); |
| logger.debug(String.format("The disks attached to the VM %s after live vm migration are: %s", vm, diskPaths)); |
| |
| List<VolumeVO> volumes = volumeDao.findByInstance(vmId); |
| for (VolumeVO volumeVO : volumes) { |
| if (Volume.State.Migrating.equals(volumeVO.getState())) { |
| logger.debug(String.format("Reconciling vm %s with volume %s in Migrating state", vm, volumeVO)); |
| 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)); |
| continue; |
| } |
| if (isVolumeMigrated) { |
| logger.debug(String.format("Adding destination volume %s to vm %s as part of reconciliation of command %s and answer %s", destVolume, vm, command, reconcileAnswer)); |
| destVolume.setState(Volume.State.Ready); |
| destVolume.setInstanceId(vmId); |
| volumeDao.update(destVolume.getId(), destVolume); |
| |
| logger.debug(String.format("Removing volume %s from vm %s as part of reconciliation of command %s and answer %s", volumeVO, vm, command, reconcileAnswer)); |
| volumeVO.setState(Volume.State.Destroy); |
| volumeVO.setVolumeType(Volume.Type.DATADISK); |
| volumeVO.setInstanceId(null); |
| volumeDao.update(volumeVO.getId(), volumeVO); |
| } else { |
| logger.debug(String.format("Removing destination volume %s from vm %s as part of reconciliation of command %s and answer %s", destVolume, vm, command, reconcileAnswer)); |
| destVolume.setState(Volume.State.Destroy); |
| destVolume.setVolumeType(Volume.Type.DATADISK); |
| destVolume.setInstanceId(null); |
| volumeDao.update(destVolume.getId(), destVolume); |
| |
| logger.debug(String.format("Adding volume %s to vm %s as part of reconciliation of command %s and answer %s", volumeVO, vm, command, reconcileAnswer)); |
| volumeVO.setState(Volume.State.Ready); |
| volumeVO.setInstanceId(vmId); |
| volumeDao.update(volumeVO.getId(), volumeVO); |
| } |
| } |
| } |
| } |
| volumes = volumeDao.findByInstance(vmId); |
| logger.debug(String.format("The disks of volumes attached to the VM %s after reconciliation of successful vm migration are: %s", vm, volumes.stream().map(VolumeVO::getPath).collect(Collectors.toList()))); |
| |
| return true; |
| } |
| |
| private Boolean isVolumeMigrated(List<String> diskPaths, String sourceVolumePath, String destVolumePath) { |
| if (StringUtils.isAnyBlank(sourceVolumePath, destVolumePath)) { |
| return null; |
| } |
| for (String diskPath : diskPaths) { |
| if (diskPath.contains(sourceVolumePath) && !diskPath.contains(destVolumePath)) { |
| return false; // Not migrated |
| } else if (!diskPath.contains(sourceVolumePath) && diskPath.contains(destVolumePath)) { |
| return true; // Migrated |
| } |
| } |
| return null; // Unable to determine |
| } |
| |
| private boolean processReconcileCopyAnswer(long requestSequence, CopyCommand command, ReconcileCopyAnswer reconcileAnswer) { |
| DataTO srcData = command.getSrcTO(); |
| DataTO destData = command.getDestTO(); |
| |
| final Long srcStoreId = srcData.getDataStore() instanceof PrimaryDataStoreTO ? ((PrimaryDataStoreTO) srcData.getDataStore()).getId() : null; |
| final Long destStoreId = destData.getDataStore() instanceof PrimaryDataStoreTO ? ((PrimaryDataStoreTO) destData.getDataStore()).getId() : null; |
| |
| VolumeVO sourceVolume = srcData.getObjectType().equals(DataObjectType.VOLUME) && srcData.getDataStore() instanceof PrimaryDataStoreTO ? volumeDao.findByIdIncludingRemoved(srcData.getId()) : null; |
| VolumeVO destVolume = destData.getObjectType().equals(DataObjectType.VOLUME) && destData.getDataStore() instanceof PrimaryDataStoreTO ? volumeDao.findByIdIncludingRemoved(destData.getId()) : null; |
| |
| if (reconcileAnswer.isSkipped()) { |
| logger.debug(String.format("The reconcile command for source volume (id: %s) to destination volume (id: %s) is ignored because it is skipped, due to reason: %s", srcData.getId(), destData.getId(), reconcileAnswer.getReason())); |
| processVolumesIfReconcileCopyCommandIsSkipped(srcStoreId, sourceVolume, destStoreId, destVolume, command, reconcileAnswer); |
| return true; |
| } |
| if (!reconcileAnswer.getResult()) { |
| logger.debug(String.format("The reconcile command for source volume (id: %s) to destination volume (id: %s) is ignored because the result is false, due to %s", srcData.getId(), destData.getId(), reconcileAnswer.getDetails())); |
| return false; |
| } |
| |
| ReconcileCommandVO reconcileCommandVO = reconcileCommandDao.findCommand(requestSequence, command.toString()); |
| if (reconcileCommandVO == null) { |
| logger.debug(String.format("The reconcile command for source volume (id: %s) to destination volume (id: %s) is not found in database, ignoring", srcData.getId(), destData.getId())); |
| return true; |
| } |
| |
| ReconcileCopyAnswer previousReconcileAnswer = null; |
| if (destVolume != null) { |
| if (reconcileCommandVO.getAnswerName() == null) { |
| logger.debug(String.format("The reconcile command for source volume (id: %s) to destination volume (id: %s) does not have previous answer in database, ignoring this time", srcData.getId(), destData.getId())); |
| return false; |
| } |
| Answer previousAnswer = ReconcileCommandUtils.parseAnswerFromAnswerInfo(reconcileCommandVO.getAnswerName(), reconcileCommandVO.getAnswerInfo()); |
| if (!(previousAnswer instanceof ReconcileCopyAnswer)) { |
| logger.debug(String.format("The reconcile command for source volume (id: %s) to destination volume (id: %s) does not have previous reconcileAnswer in database, ignoring this time", srcData.getId(), destData.getId())); |
| return false; |
| } |
| previousReconcileAnswer = (ReconcileCopyAnswer) previousAnswer; |
| } |
| |
| VolumeOnStorageTO volumeOnSource = reconcileAnswer.getVolumeOnSource(); |
| VolumeOnStorageTO volumeOnDestination = reconcileAnswer.getVolumeOnDestination(); |
| Pair<Volume.State, Volume.State> statePair = getVolumeStateOnSourceAndDestination(srcData, destData, volumeOnSource, volumeOnDestination, previousReconcileAnswer); |
| Volume.State sourceVolumeState = statePair.first(); |
| Volume.State destVolumeState = statePair.second(); |
| logger.debug(String.format("Processing volume (id: %s, state: %s) on source store and volume (id: %s, state: %s) on destination store", srcData.getId(), sourceVolumeState, destData.getId(), destVolumeState)); |
| |
| if (sourceVolume != null && destVolume != null) { |
| // copy from primary to primary (offline volume migration) |
| return processReconcileCopyAnswerFromPrimaryToPrimary(srcStoreId, sourceVolume, sourceVolumeState, destStoreId, destVolume, destVolumeState, volumeOnDestination); |
| } else if (sourceVolume == null && destVolume != null) { |
| // copy from secondary to primary |
| return processReconcileCopyAnswerFromImageCacheToPrimary(destStoreId, destVolume, destVolumeState, volumeOnDestination, command, reconcileAnswer); |
| } else if (sourceVolume != null && sourceVolumeState != null) { |
| // copy from primary to secondary |
| return processReconcileCopyAnswerFromPrimaryToSecondary(srcStoreId, sourceVolume, sourceVolumeState, command, reconcileAnswer); |
| } |
| return false; |
| } |
| |
| private boolean processVolumesIfReconcileCopyCommandIsSkipped(Long srcStoreId, VolumeVO sourceVolume, Long destStoreId, VolumeVO destVolume, |
| CopyCommand command, ReconcileCopyAnswer reconcileAnswer) { |
| if (sourceVolume == null && destVolume != null) { |
| // copy from secondary to primary |
| return processReconcileCopyAnswerFromImageCacheToPrimary(destStoreId, destVolume, Volume.State.Destroy, null, command, reconcileAnswer); |
| } else if (sourceVolume != null && destVolume == null) { |
| // copy from primary to secondary |
| return processReconcileCopyAnswerFromPrimaryToSecondary(srcStoreId, sourceVolume, Volume.State.Ready, command, reconcileAnswer); |
| } |
| return false; |
| } |
| |
| private boolean processReconcileCopyAnswerFromPrimaryToPrimary(Long srcStoreId, VolumeVO sourceVolume, Volume.State sourceVolumeState, |
| Long destStoreId, VolumeVO destVolume, Volume.State destVolumeState, VolumeOnStorageTO volumeOnDestination) { |
| boolean isSourceMigrating = sourceVolume != null && sourceVolume.getRemoved() == null && sourceVolume.getState().equals(Volume.State.Migrating); |
| if (Volume.State.Ready.equals(sourceVolumeState)) { |
| if (isSourceMigrating && srcStoreId != null && srcStoreId.equals(sourceVolume.getPoolId()) && destVolumeState != null) { |
| logger.debug(String.format("Updating source volume %s to %s state", sourceVolume, Volume.State.Ready)); |
| sourceVolume.setState(Volume.State.Ready); |
| sourceVolume.setUpdated(new Date()); |
| volumeDao.update(sourceVolume.getId(), sourceVolume); |
| } |
| if (Volume.State.Creating.equals(destVolume.getState()) && destStoreId != null && destStoreId.equals(destVolume.getPoolId()) && destVolumeState != null) { |
| logger.debug(String.format("Updating destination volume %s to %s state", destVolume, destVolumeState)); |
| destVolume.setState(destVolumeState); |
| destVolume.setUpdated(new Date()); |
| volumeDao.update(destVolume.getId(), destVolume); |
| } |
| return true; |
| } else if (Volume.State.Ready.equals(destVolumeState)) { |
| if (Volume.State.Creating.equals(destVolume.getState()) && destStoreId != null && destStoreId.equals(destVolume.getPoolId())) { |
| logger.debug(String.format("Updating destination volume %s to %s state", destVolume, Volume.State.Ready)); |
| destVolume.setState(Volume.State.Ready); |
| destVolume.setUpdated(new Date()); |
| destVolume.setInstanceId(sourceVolume.getInstanceId()); |
| destVolume.setPath(volumeOnDestination.getPath()); // Update path of destination volume |
| destVolume.set_iScsiName(volumeOnDestination.getPath()); |
| volumeDao.update(destVolume.getId(), destVolume); |
| } |
| if (isSourceMigrating && srcStoreId != null && srcStoreId.equals(sourceVolume.getPoolId()) && sourceVolumeState != null) { |
| logger.debug(String.format("Updating source volume %s to %s state", sourceVolume, sourceVolumeState)); |
| sourceVolume.setState(sourceVolumeState); |
| sourceVolume.setUpdated(new Date()); |
| sourceVolume.setInstanceId(null); |
| volumeDao.update(sourceVolume.getId(), sourceVolume); |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| private boolean processReconcileCopyAnswerFromImageCacheToPrimary(Long destStoreId, VolumeVO destVolume, Volume.State destVolumeState, VolumeOnStorageTO volumeOnDestination, |
| CopyCommand command, ReconcileCopyAnswer reconcileAnswer) { |
| if (destVolume.getRemoved() == null && Volume.State.Creating.equals(destVolume.getState()) && destStoreId != null && destStoreId.equals(destVolume.getPoolId())) { |
| Long lastVolumeId = destVolume.getLastId(); |
| logger.debug(String.format("Searching for last volume with id = %s", destVolume.getLastId())); |
| VolumeVO lastVolume = volumeDao.findById(lastVolumeId); |
| if (lastVolume != null && Arrays.asList(Volume.State.Migrating, Volume.State.Ready).contains(lastVolume.getState()) |
| && Volume.State.Destroy.equals(destVolumeState)) { |
| destVolume.setState(destVolumeState); |
| if (volumeOnDestination != null) { |
| destVolume.setPath(volumeOnDestination.getPath()); // Update path of destination volume |
| } |
| volumeDao.update(destVolume.getId(), destVolume); |
| if (Volume.State.Migrating.equals(lastVolume.getState())) { |
| lastVolume.setState(Volume.State.Ready); // Update last volume to Ready |
| volumeDao.update(lastVolume.getId(), lastVolume); |
| } |
| // remove record from volume_store_ref with Copying state |
| VolumeDataStoreVO volumeDataStoreVO = volumeDataStoreDao.findByVolume(lastVolume.getId()); |
| if (volumeDataStoreVO != null && volumeDataStoreVO.getState().equals(ObjectInDataStoreStateMachine.State.Copying)) { |
| logger.debug(String.format("Removing record (id: %s) for volume %s from volume_store_ref as part of reconciliation of command %s and answer %s", volumeDataStoreVO.getId(), lastVolume, command, reconcileAnswer)); |
| volumeDataStoreDao.remove(volumeDataStoreVO.getId()); |
| } |
| } |
| return true; |
| } |
| return false; |
| } |
| |
| private boolean processReconcileCopyAnswerFromPrimaryToSecondary(Long srcStoreId, VolumeVO sourceVolume, Volume.State sourceVolumeState, |
| CopyCommand command, ReconcileCopyAnswer reconcileAnswer) { |
| boolean isSourceMigrating = sourceVolume != null && sourceVolume.getRemoved() == null && sourceVolume.getState().equals(Volume.State.Migrating); |
| if (isSourceMigrating && srcStoreId != null && srcStoreId.equals(sourceVolume.getPoolId())) { |
| logger.debug(String.format("Updating source volume %s to %s state", sourceVolume, sourceVolumeState)); |
| sourceVolume.setState(sourceVolumeState); // Update source volume state |
| volumeDao.update(sourceVolume.getId(), sourceVolume); |
| |
| // remove record from volume_store_ref with Creating state |
| VolumeDataStoreVO volumeDataStoreVO = volumeDataStoreDao.findByVolume(sourceVolume.getId()); |
| if (volumeDataStoreVO != null && volumeDataStoreVO.getState().equals(ObjectInDataStoreStateMachine.State.Creating)) { |
| logger.debug(String.format("Removing record (id: %s) for volume %s from volume_store_ref as part of reconciliation of command %s and answer %s", volumeDataStoreVO.getId(), sourceVolume, command, reconcileAnswer)); |
| volumeDataStoreDao.remove(volumeDataStoreVO.getId()); |
| } |
| } |
| logger.debug(String.format("Searching for volumes with last_id = %s", sourceVolume.getId())); |
| VolumeVO newVolume = volumeDao.findByLastIdAndState(sourceVolume.getId(), Volume.State.Creating); |
| if (newVolume != null) { |
| logger.debug(String.format("Removing volume %s as part of reconciliation of command %s and answer %s", newVolume, command, reconcileAnswer)); |
| newVolume.setState(Volume.State.Destroy); |
| newVolume.setVolumeType(Volume.Type.DATADISK); |
| newVolume.setInstanceId(null); |
| newVolume.setRemoved(new Date()); |
| volumeDao.update(newVolume.getId(), newVolume); |
| } |
| return true; |
| } |
| |
| private boolean processReconcileMigrateVolumeAnswer(long requestSequence, MigrateVolumeCommand command, ReconcileMigrateVolumeAnswer reconcileAnswer) { |
| DataTO srcData = command.getSrcData(); |
| DataTO destData = command.getDestData(); |
| if (srcData == null || destData == null) { |
| logger.debug(String.format("The source (%s) and destination (%s) of MigrateCommand must be non-empty", srcData, destData)); |
| return true; |
| } |
| if (srcData.getId() != destData.getId()) { |
| logger.debug(String.format("The source volume (id: %s) and destination volume (id: %s) of MigrateCommand must be same ID", srcData.getId(), destData.getId())); |
| return true; |
| } |
| VolumeVO sourceVolume = volumeDao.findByIdIncludingRemoved(srcData.getId()); |
| if (sourceVolume == null || sourceVolume.getRemoved() != null) { |
| logger.debug(String.format("Volume (id: %s) has been removed in CloudStack", srcData.getId())); |
| return true; |
| } |
| if (!sourceVolume.getState().equals(Volume.State.Migrating)) { |
| logger.debug(String.format("Volume %s (state: %s) is not in Migrating state", sourceVolume, sourceVolume.getState())); |
| return true; |
| } |
| |
| if (!(srcData.getDataStore() instanceof PrimaryDataStoreTO) && (destData.getDataStore() instanceof PrimaryDataStoreTO)) { |
| logger.debug(String.format("The source (role: %s) and destination (role: %s) of MigrateCommand must be Primary", srcData.getDataStore().getRole(), destData.getDataStore().getRole())); |
| return true; |
| } |
| |
| ReconcileCommandVO reconcileCommandVO = reconcileCommandDao.findCommand(requestSequence, command.toString()); |
| if (reconcileCommandVO == null) { |
| logger.debug(String.format("The reconcile command for migrating volume %s is not found in database, ignoring", sourceVolume)); |
| return true; |
| } |
| if (reconcileCommandVO.getAnswerName() == null) { |
| logger.debug(String.format("The reconcile command for migrating volume %s does not have previous answer in database, ignoring this time", sourceVolume)); |
| return false; |
| } |
| Answer previousAnswer = ReconcileCommandUtils.parseAnswerFromAnswerInfo(reconcileCommandVO.getAnswerName(), reconcileCommandVO.getAnswerInfo()); |
| if (!(previousAnswer instanceof ReconcileMigrateVolumeAnswer)) { |
| logger.debug(String.format("The reconcile command for for migrating volume %s does not have previous reconcileAnswer in database, ignoring this time", sourceVolume)); |
| return false; |
| } |
| List<String> diskPaths = reconcileAnswer.getVmDiskPaths(); |
| logger.debug(String.format("The disks attached to the VM %s after live volume migration are: %s", reconcileAnswer.getVmName(), diskPaths)); |
| |
| PrimaryDataStoreTO srcDataStore = (PrimaryDataStoreTO) srcData.getDataStore(); |
| PrimaryDataStoreTO destDataStore = (PrimaryDataStoreTO) destData.getDataStore(); |
| |
| VolumeOnStorageTO volumeOnSource = reconcileAnswer.getVolumeOnSource(); |
| VolumeOnStorageTO volumeOnDestination = reconcileAnswer.getVolumeOnDestination(); |
| ReconcileMigrateVolumeAnswer previousReconcileAnswer = (ReconcileMigrateVolumeAnswer) previousAnswer; |
| Pair<Volume.State, Volume.State> statePair = getVolumeStateOnSourceAndDestination(srcData, destData, volumeOnSource, volumeOnDestination, previousReconcileAnswer); |
| Volume.State sourceVolumeState = statePair.first(); |
| Volume.State destVolumeState = statePair.second(); |
| logger.debug(String.format("Processing volume (id: %s, state: %s) on source pool and volume (id: %s, state: %s) on destination pool", srcData.getId(), sourceVolumeState, destData.getId(), destVolumeState)); |
| if (Volume.State.Ready.equals(sourceVolumeState)) { |
| updateVolumeAndDestroyOldVolume(sourceVolume, srcData, srcDataStore, destData, destDataStore, volumeOnDestination); |
| return true; |
| } else if (Volume.State.Ready.equals(destVolumeState)) { |
| updateVolumeAndDestroyOldVolume(sourceVolume, destData, destDataStore, srcData, srcDataStore, volumeOnSource); |
| return true; |
| } else if (CollectionUtils.isNotEmpty(diskPaths)) { |
| // VM is Running |
| if (!Volume.State.Migrating.equals(destVolumeState) && volumeOnDestination != null) { |
| for (String diskPath : diskPaths) { |
| if (diskPath.equals(volumeOnDestination.getFullPath())) { |
| logger.debug(String.format("The VM %s is running with the volume %s on destination pool %s, the volume has been migrated", reconcileAnswer.getVmName(), volumeOnDestination.getFullPath(), destData.getDataStore())); |
| updateVolumeAndDestroyOldVolume(sourceVolume, destData, destDataStore, srcData, srcDataStore, volumeOnSource); |
| return true; |
| } else if (diskPath.equals(volumeOnSource.getFullPath())) { |
| logger.debug(String.format("The VM %s is running with the volume %s on source pool %s, the volume has not been migrated", reconcileAnswer.getVmName(), volumeOnSource.getFullPath(), srcData.getDataStore())); |
| updateVolumeAndDestroyOldVolume(sourceVolume, srcData, srcDataStore, destData, destDataStore, volumeOnDestination); |
| return true; |
| } |
| } |
| } |
| } |
| return false; |
| } |
| |
| private void updateVolumeAndDestroyOldVolume(VolumeVO sourceVolume, DataTO srcData, PrimaryDataStoreTO srcDataStore, DataTO destData, PrimaryDataStoreTO destDataStore, VolumeOnStorageTO volumeToBeDeleted) { |
| |
| logger.debug(String.format("Updating volume %s to %s state", sourceVolume, Volume.State.Ready)); |
| sourceVolume.setState(Volume.State.Ready); |
| sourceVolume.setPoolId(srcDataStore.getId()); // restore pool_id and update path |
| sourceVolume.setPoolType(srcDataStore.getPoolType()); |
| sourceVolume.setPath(srcData.getPath()); |
| sourceVolume.set_iScsiName(srcData.getPath()); |
| sourceVolume.setUpdated(new Date()); |
| volumeDao.update(sourceVolume.getId(), sourceVolume); |
| |
| if (volumeToBeDeleted != null) { |
| logger.debug(String.format("Creating a dummy volume from %s on pool %s", sourceVolume, destDataStore.getId())); |
| Volume newVol = volumeManager.allocateDuplicateVolume(sourceVolume, null, null); |
| VolumeVO newVolume = (VolumeVO) newVol; |
| newVolume.setInstanceId(null); |
| newVolume.setPoolId(destDataStore.getId()); |
| newVolume.setPoolType(destDataStore.getPoolType()); |
| newVolume.setState(Volume.State.Creating); |
| newVolume.setPath(destData.getPath()); |
| newVolume.set_iScsiName(destData.getPath()); |
| volumeDao.update(newVolume.getId(), newVolume); |
| |
| logger.debug(String.format("Deleting the dummy volume %s on pool %s", newVolume, destDataStore.getId())); |
| volumeApiService.destroyVolume(newVolume.getId(), accountManager.getAccount(Account.ACCOUNT_ID_SYSTEM), true, true); |
| } |
| } |
| |
| private Pair<Volume.State, Volume.State> getVolumeStateOnSourceAndDestination(DataTO srcData, DataTO destData, VolumeOnStorageTO volumeOnSource, VolumeOnStorageTO volumeOnDestination, ReconcileVolumeAnswer previousReconcileAnswer) { |
| final Long srcStoreId = srcData.getDataStore() instanceof PrimaryDataStoreTO ? ((PrimaryDataStoreTO) srcData.getDataStore()).getId() : null; |
| final Long destStoreId = destData.getDataStore() instanceof PrimaryDataStoreTO ? ((PrimaryDataStoreTO) destData.getDataStore()).getId() : null; |
| |
| VolumeOnStorageTO previousVolumeOnDestination = previousReconcileAnswer != null ? previousReconcileAnswer.getVolumeOnDestination() : null; |
| |
| if (volumeOnSource != null) { |
| if (volumeOnDestination == null) { |
| if (volumeOnSource.getPath() != null) { |
| logger.debug(String.format("Volume (id :%s) exist on source (id: %s) and volume (id: %s) does not exist on destination (id: %s), updating state to Ready on source pool", srcData.getId(), srcStoreId, destData.getId(), destStoreId)); |
| return new Pair<>(Volume.State.Ready, null); |
| } else { |
| logger.debug(String.format("Volume (id :%s) cannot be found on source (id: %s) and volume (id: %s) does not exist on destination (id: %s), updating state to Ready on source pool", srcData.getId(), srcStoreId, destData.getId(), destStoreId)); |
| return new Pair<>(Volume.State.Destroy, null); |
| } |
| } |
| if (volumeOnDestination.getPath() == null) { |
| logger.debug(String.format("Volume (id :%s) exist on source (id: %s) and volume (id: %s) cannot be found on destination (id: %s), updating state to Ready on source pool", srcData.getId(), srcStoreId, destData.getId(), destStoreId)); |
| return new Pair<>(Volume.State.Ready, Volume.State.Destroy); |
| } |
| boolean isDestinationVolumeChanged = (previousVolumeOnDestination != null && volumeOnDestination.getSize() != previousVolumeOnDestination.getSize()); |
| if (isDestinationVolumeChanged) { |
| logger.debug(String.format("Volume (id :%s) on destination (id: %s) is still being updated, skipping", destData.getId(), destStoreId)); |
| return new Pair<>(Volume.State.Migrating, Volume.State.Migrating); |
| } else if (destData.getId() == srcData.getId()) { |
| logger.debug(String.format("Volume (id :%s) on destination (id: %s) is not updated, cannot determine the state on source and destination pool", destData.getId(), destStoreId)); |
| return new Pair<>(null, null); |
| } else { |
| logger.debug(String.format("Volume (id :%s) on destination (id: %s) is not updated, updating state to Ready on source pool and Destroy on destination pool", destData.getId(), destStoreId)); |
| return new Pair<>(Volume.State.Ready, Volume.State.Destroy); |
| } |
| } else if (volumeOnDestination != null) { |
| boolean isDestinationVolumeChanged = (previousVolumeOnDestination != null && volumeOnDestination.getSize() != previousVolumeOnDestination.getSize()); |
| if (isDestinationVolumeChanged) { |
| logger.debug(String.format("Volume (id :%s) on destination (id: %s) is still being updated, skipping", destData.getId(), destStoreId)); |
| return new Pair<>(srcStoreId != null ? Volume.State.Migrating : null, Volume.State.Migrating); |
| } else if (srcStoreId != null) { |
| // from primary to primary |
| logger.debug(String.format("Volume (id: %s) does not exist on source (id: %s) but volume (id: %s) exist on destination (id: %s), updating state to Ready on destination pool", srcData.getId(), srcStoreId, destData.getId(), destStoreId)); |
| return new Pair<>(Volume.State.Destroy, Volume.State.Ready); |
| } else { |
| // from secondary to primary |
| logger.debug(String.format("Volume (id: %s) exist on destination (id: %s), however it is copied from secondary, updating state to Destroy on destination pool", destData.getId(), destStoreId)); |
| return new Pair<>(null, Volume.State.Destroy); |
| } |
| } |
| |
| 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(); |
| } |
| } |
