diff --git a/core/src/main/java/org/apache/cloudstack/backup/TakeBackupCommand.java b/core/src/main/java/org/apache/cloudstack/backup/TakeBackupCommand.java index 93855ea17211..f225c55bb46e 100644 --- a/core/src/main/java/org/apache/cloudstack/backup/TakeBackupCommand.java +++ b/core/src/main/java/org/apache/cloudstack/backup/TakeBackupCommand.java @@ -32,6 +32,7 @@ public class TakeBackupCommand extends Command { private List volumePaths; @LogLevel(LogLevel.Log4jLevel.Off) private String mountOptions; + private boolean executeInSequence = false; public TakeBackupCommand(String vmName, String backupPath) { super(); @@ -89,6 +90,13 @@ public void setVolumePaths(List volumePaths) { @Override public boolean executeInSequence() { - return true; + // Parallel by default: each backup uses its own per-VM on-NAS path, so concurrent + // runs do not contend. Operators can force sequential execution per zone via the + // backup.nas.parallel.execution.enabled setting. + return executeInSequence; + } + + public void setExecuteInSequence(boolean executeInSequence) { + this.executeInSequence = executeInSequence; } } diff --git a/core/src/test/java/org/apache/cloudstack/backup/TakeBackupCommandTest.java b/core/src/test/java/org/apache/cloudstack/backup/TakeBackupCommandTest.java new file mode 100644 index 000000000000..0f86c2142bdb --- /dev/null +++ b/core/src/test/java/org/apache/cloudstack/backup/TakeBackupCommandTest.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 org.apache.cloudstack.backup; + +import org.junit.Assert; +import org.junit.Test; + +public class TakeBackupCommandTest { + + @Test + public void testExecuteInSequenceDefaultsToParallel() { + TakeBackupCommand command = new TakeBackupCommand("vm-1", "/backups/vm-1"); + // Default: run in parallel with other backup/delete commands. + Assert.assertFalse(command.executeInSequence()); + } + + @Test + public void testExecuteInSequenceIsSettable() { + TakeBackupCommand command = new TakeBackupCommand("vm-1", "/backups/vm-1"); + + command.setExecuteInSequence(true); + Assert.assertTrue(command.executeInSequence()); + + command.setExecuteInSequence(false); + Assert.assertFalse(command.executeInSequence()); + } +} diff --git a/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDao.java b/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDao.java index 89a13245b0a0..5dbfb8d0a096 100644 --- a/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDao.java +++ b/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDao.java @@ -35,5 +35,6 @@ public interface BackupDao extends GenericDao { List syncBackups(Long zoneId, Long vmId, List externalBackups); BackupVO getBackupVO(Backup backup); List listByOfferingId(Long backupOfferingId); + List listByZoneAndStatus(long zoneId, Backup.Status status); BackupResponse newBackupResponse(Backup backup); } diff --git a/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDaoImpl.java b/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDaoImpl.java index 5a9cd0620374..9ae70bec280b 100644 --- a/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDaoImpl.java +++ b/engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDaoImpl.java @@ -60,6 +60,7 @@ public class BackupDaoImpl extends GenericDaoBase implements Bac BackupOfferingDao backupOfferingDao; private SearchBuilder backupSearch; + private SearchBuilder zoneStatusSearch; public BackupDaoImpl() { } @@ -72,6 +73,19 @@ protected void init() { backupSearch.and("backup_offering_id", backupSearch.entity().getBackupOfferingId(), SearchCriteria.Op.EQ); backupSearch.and("zone_id", backupSearch.entity().getZoneId(), SearchCriteria.Op.EQ); backupSearch.done(); + + zoneStatusSearch = createSearchBuilder(); + zoneStatusSearch.and("zone_id", zoneStatusSearch.entity().getZoneId(), SearchCriteria.Op.EQ); + zoneStatusSearch.and("status", zoneStatusSearch.entity().getStatus(), SearchCriteria.Op.EQ); + zoneStatusSearch.done(); + } + + @Override + public List listByZoneAndStatus(long zoneId, Backup.Status status) { + SearchCriteria sc = zoneStatusSearch.create(); + sc.setParameters("zone_id", zoneId); + sc.setParameters("status", status); + return listBy(sc); } @Override diff --git a/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java b/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java index 565ea29acf8b..f341ff9a00d5 100644 --- a/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java +++ b/plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java @@ -34,6 +34,7 @@ import com.cloud.storage.dao.VolumeDao; import com.cloud.utils.Pair; import com.cloud.utils.component.AdapterBase; +import com.cloud.utils.db.GlobalLock; import com.cloud.utils.exception.CloudRuntimeException; import com.cloud.vm.VirtualMachine; import com.cloud.vm.dao.VMInstanceDao; @@ -59,11 +60,62 @@ import java.util.HashMap; import java.util.Objects; import java.util.UUID; +import java.util.function.Supplier; import java.util.stream.Collectors; public class NASBackupProvider extends AdapterBase implements BackupProvider, Configurable { private static final Logger LOG = LogManager.getLogger(NASBackupProvider.class); + public static final ConfigKey NASBackupParallelExecution = new ConfigKey<>("Advanced", Boolean.class, + "backup.nas.parallel.execution.enabled", + "true", + "Let NAS take-backup commands run concurrently on a KVM host instead of queueing behind every earlier command on that host " + + "and holding up every later one. Concurrency is bounded per host by backup.nas.parallel.max.per.host. " + + "Disable to restore strictly sequential execution.", + true, ConfigKey.Scope.Zone); + + public static final ConfigKey NASBackupParallelMaxPerHost = new ConfigKey<>("Advanced", Integer.class, + "backup.nas.parallel.max.per.host", + "2", + "Maximum number of NAS take-backup commands in flight on one KVM host when parallel execution is enabled; further " + + "backups for that host wait on the management server that received them. The limit holds across all management " + + "servers: in-flight backups are counted from the database under a cluster-wide lock per host. Keep it below the " + + "agent's worker thread count (workers in agent.properties, default 5) so start, stop, reboot and migrate commands " + + "are never queued behind backups.", + true, ConfigKey.Scope.Zone); + + public static final ConfigKey NASBackupParallelQueueTimeout = new ConfigKey<>("Advanced", Integer.class, + "backup.nas.parallel.queue.timeout", + "7200", + "Seconds a NAS take-backup may wait for a free per-host slot before it fails.", + true, ConfigKey.Scope.Zone); + + public static final ConfigKey NASBackupTimeout = new ConfigKey<>("Advanced", Integer.class, + "backup.nas.backup.timeout", + "7200", + "Seconds the management server waits for a NAS take-backup to finish on the host (commands.timeout for " + + "TakeBackupCommand takes precedence when set). The default matches the KVM agent's cmds.timeout (7200), after " + + "which the agent stops the backup script; keep this at or above cmds.timeout. A backup that has not answered " + + "in time may still be running on the host, so it stays BackingUp and keeps its host slot until it is twice " + + "this old, when it is marked Error (it stays in the backup list and can be deleted).", + true, ConfigKey.Scope.Zone); + + /** Seconds to wait for a host's cluster-wide lock; it is only held for the count-and-reserve step. */ + private static final int HOST_SLOT_LOCK_WAIT_SECONDS = 30; + + /** + * The least time a BackingUp row is held beyond the take-backup wait before it is treated as abandoned. The hold + * is the wait again (so twice the wait in all), but never less than this. + */ + private static final int MIN_ABANDONED_MARGIN_SECONDS = 300; + + /** Bounds of the back-off between two attempts to reserve a slot on a full host. */ + protected long slotPollMinMillis = 1000L; + protected long slotPollMaxMillis = 15000L; + + /** Wakes the waiters on this management server as soon as one of its own backups finishes. */ + private final Object localSlotRelease = new Object(); + @Inject private BackupDao backupDao; @@ -154,11 +206,14 @@ public boolean takeBackup(final VirtualMachine vm) { final String backupPath = String.format("%s/%s", vm.getInstanceName(), new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss").format(creationDate)); - BackupVO backupVO = createBackupObject(vm, backupPath); TakeBackupCommand command = new TakeBackupCommand(vm.getInstanceName(), backupPath); command.setBackupRepoType(backupRepository.getType()); command.setBackupRepoAddress(backupRepository.getAddress()); command.setMountOptions(backupRepository.getMountOptions()); + final long zoneId = vm.getDataCenterId(); + final boolean parallel = applyExecutionPolicy(command, zoneId); + final int waitSeconds = takeBackupCommandWaitSeconds(zoneId); + command.setWait(waitSeconds); if (VirtualMachine.State.Stopped.equals(vm.getState())) { List vmVolumes = volumeDao.findByInstance(vm.getId()); @@ -167,26 +222,245 @@ public boolean takeBackup(final VirtualMachine vm) { command.setVolumePaths(volumePaths); } - BackupAnswer answer = null; + // Concurrent backups are bounded per host across all management servers: the BackingUp row of this + // backup is the reservation of one of the host's slots, created only while the host has a free slot. + // The wait happens here, on the management server, never on the agent's worker threads. + final BackupVO backupVO = parallel + ? reserveHostBackupSlot(vm, host.getId(), backupPath, NASBackupParallelMaxPerHost.valueIn(zoneId), NASBackupParallelQueueTimeout.valueIn(zoneId)) + : createBackupObject(vm, backupPath); + boolean succeeded = false; + boolean outcomeUnknown = false; try { - answer = (BackupAnswer) agentManager.send(host.getId(), command); - } catch (AgentUnavailableException e) { - throw new CloudRuntimeException("Unable to contact backend control plane to initiate backup"); - } catch (OperationTimedoutException e) { - throw new CloudRuntimeException("Operation to initiate backup timed out, please try again"); - } - - if (answer != null && answer.getResult()) { + final BackupAnswer answer; + try { + answer = (BackupAnswer) agentManager.send(host.getId(), command); + } catch (AgentUnavailableException e) { + throw new CloudRuntimeException("Unable to contact backend control plane to initiate backup"); + } catch (OperationTimedoutException e) { + // The host did not answer in time, but its backup may still be running and writing to the NAS. + // Keep the BackingUp row, and with it the host slot, so the backup stays visible and the host is + // not given another backup early. markAbandonedBackups turns it into Error once it is too old. + outcomeUnknown = true; + LOG.warn("NAS backup {} of VM {} did not answer within {} seconds; it may still be running on host {}. " + + "It stays BackingUp and keeps its host slot until it is {} seconds old, then it is marked Error", + backupVO.getUuid(), vm.getInstanceName(), waitSeconds, host.getId(), abandonedAfterSeconds(waitSeconds)); + throw new CloudRuntimeException(String.format("Backup did not finish within %d seconds and may still be " + + "running on the host; it stays in BackingUp state until the host is done or it is marked Error", waitSeconds)); + } + if (answer == null || !answer.getResult()) { + return false; + } + succeeded = true; backupVO.setDate(new Date()); backupVO.setSize(answer.getSize()); backupVO.setStatus(Backup.Status.BackedUp); backupVO.setBackedUpVolumes(BackupManagerImpl.createVolumeInfoFromVolumes(volumeDao.findByInstance(vm.getId()))); return backupDao.update(backupVO.getId(), backupVO); - } else { - backupVO.setStatus(Backup.Status.Failed); - backupDao.remove(backupVO.getId()); + } finally { + if (!succeeded && !outcomeUnknown) { + // Releases the slot: the host reported a failure or never got the command, so nothing is running. + backupVO.setStatus(Backup.Status.Failed); + backupDao.update(backupVO.getId(), backupVO); + backupDao.remove(backupVO.getId()); + } + if (parallel) { + wakeLocalSlotWaiters(); + } + } + } + + /** + * Applies the zone's execution policy to the command and returns true when the command will run + * concurrently with other commands on the host (and so must be gated per host). + */ + protected boolean applyExecutionPolicy(final TakeBackupCommand command, final long zoneId) { + final boolean parallel = Boolean.TRUE.equals(NASBackupParallelExecution.valueIn(zoneId)); + command.setExecuteInSequence(!parallel); + return parallel; + } + + static String hostBackupLockName(final long hostId) { + return "nas-backup-host-" + hostId; + } + + /** + * Reserves one of the host's backup slots and returns the BackingUp backup row that holds it. The row is + * the reservation: it is created only while fewer than maxPerHost NAS backups are in flight on the host, + * counted from the database under the host's cluster-wide lock, so the limit holds across every management + * server. The lock is held only for the count-and-insert step, never for the backup itself. The slot is + * released when the row leaves BackingUp (BackedUp, or Failed and removed). When the host is full this + * waits with back-off, up to timeoutSeconds, then fails with CloudRuntimeException without leaving a row. + */ + protected BackupVO reserveHostBackupSlot(final VirtualMachine vm, final long hostId, final String backupPath, + final Integer maxPerHost, final Integer timeoutSeconds) { + final int max = Math.max(1, maxPerHost == null ? 1 : maxPerHost); + final int timeout = Math.max(0, timeoutSeconds == null ? 0 : timeoutSeconds); + final long deadline = System.currentTimeMillis() + timeout * 1000L; + final int[] lastCount = {-1}; + long pause = slotPollMinMillis; + while (true) { + final BackupVO reserved = runUnderHostLock(hostId, () -> { + final int inFlight = countInFlightBackups(hostId, vm.getDataCenterId()); + lastCount[0] = inFlight; + if (inFlight >= max) { + return null; + } + final BackupVO backup = createBackupObject(vm, backupPath); + LOG.debug("Host {} now has {} of {} NAS backups in flight", hostId, inFlight + 1, max); + return backup; + }); + if (reserved != null) { + return reserved; + } + final long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + throw new CloudRuntimeException(String.format( + "Timed out after %d seconds waiting for a NAS backup slot on host %d (%d of %d in flight); " + + "raise %s or %s, or retry when the host's backups finish", + timeout, hostId, Math.max(lastCount[0], 0), max, NASBackupParallelMaxPerHost.key(), NASBackupParallelQueueTimeout.key())); + } + synchronized (localSlotRelease) { + try { + localSlotRelease.wait(Math.max(1L, Math.min(pause, remaining))); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new CloudRuntimeException("Interrupted while waiting for a NAS backup slot on host " + hostId); + } + } + pause = Math.min(pause * 2, slotPollMaxMillis); + } + } + + /** + * Runs the action while holding the host's cluster-wide database lock and returns its result, or null when + * the lock could not be taken in time (the caller then retries like it does for a full host). + */ + protected T runUnderHostLock(final long hostId, final Supplier action) { + final GlobalLock lock = GlobalLock.getInternLock(hostBackupLockName(hostId)); + try { + if (!lock.lock(HOST_SLOT_LOCK_WAIT_SECONDS)) { + LOG.debug("Could not take lock {} within {} seconds, will retry", lock.getName(), HOST_SLOT_LOCK_WAIT_SECONDS); + return null; + } + try { + return action.get(); + } finally { + lock.unlock(); + } + } finally { + lock.releaseRef(); + } + } + + /** + * Counts the NAS backups in flight on the host, from the database, so every management server sees the same + * number. A backup belongs to the host its VM is on, or was last on when stopped, which is how + * getVMHypervisorHost picks the host the command is sent to. A running VM cannot migrate while libvirt runs + * its backup job, so the attribution does not drift for running VMs. A row keeps its slot after the take-backup + * wait runs out, because the host may still be running it, and is only left out once it is twice the wait old + * (abandoned: for example its management server stopped mid-backup), so a slot can never leak forever. + */ + protected int countInFlightBackups(final long hostId, final long zoneId) { + final List backingUp = backupDao.listByZoneAndStatus(zoneId, Backup.Status.BackingUp); + if (CollectionUtils.isEmpty(backingUp)) { + return 0; + } + final long abandonedBefore = abandonedBefore(zoneId); + int inFlight = 0; + for (final BackupVO backup : backingUp) { + if (backup.getDate() != null && backup.getDate().getTime() < abandonedBefore) { + LOG.debug("Not counting abandoned BackingUp backup {} (started {}) towards host {}", backup.getUuid(), backup.getDate(), hostId); + continue; + } + final BackupOffering offering = backupOfferingDao.findByIdIncludingRemoved(backup.getBackupOfferingId()); + if (offering != null && !getName().equalsIgnoreCase(offering.getProvider())) { + continue; + } + final VirtualMachine backupVm = vmInstanceDao.findByIdIncludingRemoved(backup.getVmId()); + if (backupVm == null) { + continue; + } + final Long vmHostId = backupVm.getHostId() != null ? backupVm.getHostId() : backupVm.getLastHostId(); + if (vmHostId != null && vmHostId == hostId) { + inFlight++; + } + } + return inFlight; + } + + /** Seconds after which a BackingUp row that has not answered is treated as abandoned: twice the wait, at least wait + 5 minutes. */ + static long abandonedAfterSeconds(final int waitSeconds) { + return (long) waitSeconds + Math.max(waitSeconds, MIN_ABANDONED_MARGIN_SECONDS); + } + + /** BackingUp rows of the zone that started before this time are abandoned. */ + protected long abandonedBefore(final long zoneId) { + return System.currentTimeMillis() - abandonedAfterSeconds(takeBackupCommandWaitSeconds(zoneId)) * 1000L; + } + + /** + * Marks the VM's abandoned NAS backups Error: rows still BackingUp twice the take-backup wait after they started. + * By then the agent has stopped the backup script (cmds.timeout) and no management server is waiting for an + * answer, but the outcome on the NAS is unknown, so the row is kept, visible and deletable, instead of removed. + * Runs from the periodic backup sync task on every management server; marking the same row twice is harmless. + */ + protected int markAbandonedBackups(final VirtualMachine vm) { + final List backups = backupDao.listByVmId(vm.getDataCenterId(), vm.getId()); + if (CollectionUtils.isEmpty(backups)) { + return 0; + } + final long abandonedBefore = abandonedBefore(vm.getDataCenterId()); + int marked = 0; + for (final Backup backup : backups) { + if (!Backup.Status.BackingUp.equals(backup.getStatus()) || backup.getDate() == null + || backup.getDate().getTime() >= abandonedBefore) { + continue; + } + final BackupVO row = backupDao.findById(backup.getId()); + if (row == null || !Backup.Status.BackingUp.equals(row.getStatus())) { + continue; + } + LOG.warn("NAS backup {} of VM {} started at {} and never reported back; marking it Error. Its files on the " + + "backup repository may be incomplete, delete the backup to remove them", row.getUuid(), vm.getInstanceName(), row.getDate()); + row.setStatus(Backup.Status.Error); + backupDao.update(row.getId(), row); + marked++; + } + return marked; + } + + /** + * The longest a management server waits for a TakeBackupCommand answer: commands.timeout if set for it, + * else backup.nas.backup.timeout, else (when that is 0) the global wait. The provider sets it as the command's wait. + */ + protected int takeBackupCommandWaitSeconds(final long zoneId) { + int wait = Math.max(0, Objects.requireNonNullElse(NASBackupTimeout.valueIn(zoneId), 0)); + if (wait == 0) { + wait = Math.max(0, Objects.requireNonNullElse(AgentManager.Wait.value(), 0)); + } + final String granular = AgentManager.GranularWaitTimeForCommands.value(); + if (granular != null) { + for (final String entry : granular.split(",")) { + final String[] pair = entry.trim().split("="); + if (pair.length == 2 && TakeBackupCommand.class.getSimpleName().equals(pair[0].trim())) { + try { + final int value = Integer.parseInt(pair[1].trim()); + if (value > 0) { + wait = value; + } + } catch (NumberFormatException e) { + LOG.debug("Ignoring invalid commands.timeout entry {}", entry); + } + } + } + } + return wait; + } + + protected void wakeLocalSlotWaiters() { + synchronized (localSlotRelease) { + localSlotRelease.notifyAll(); } - return Objects.nonNull(answer) && answer.getResult(); } private BackupVO createBackupObject(VirtualMachine vm, String backupPath) { @@ -430,6 +704,7 @@ public boolean willDeleteBackupsOnOfferingRemoval() { @Override public void syncBackups(VirtualMachine vm, Backup.Metric metric) { // TODO: check and sum/return backups metrics on per VM basis + markAbandonedBackups(vm); } @Override @@ -450,6 +725,10 @@ public boolean isValidProviderOffering(Long zoneId, String uuid) { @Override public ConfigKey[] getConfigKeys() { return new ConfigKey[]{ + NASBackupParallelExecution, + NASBackupParallelMaxPerHost, + NASBackupParallelQueueTimeout, + NASBackupTimeout }; } diff --git a/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java b/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java new file mode 100644 index 000000000000..d507df2b617b --- /dev/null +++ b/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java @@ -0,0 +1,507 @@ +// 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.backup; + +import java.lang.reflect.Field; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import javax.inject.Inject; + +import org.apache.cloudstack.framework.config.ConfigKey; +import org.junit.Assert; +import org.apache.cloudstack.backup.dao.BackupDao; +import org.apache.cloudstack.backup.dao.BackupOfferingDao; +import org.apache.cloudstack.backup.dao.BackupRepositoryDao; +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.stubbing.Answer; + +import com.cloud.agent.AgentManager; +import com.cloud.agent.api.Command; +import com.cloud.exception.AgentUnavailableException; +import com.cloud.exception.OperationTimedoutException; +import com.cloud.host.HostVO; +import com.cloud.host.Status; +import com.cloud.host.dao.HostDao; +import com.cloud.hypervisor.Hypervisor; +import com.cloud.storage.dao.VolumeDao; +import com.cloud.utils.exception.CloudRuntimeException; +import com.cloud.vm.VMInstanceVO; +import com.cloud.vm.VirtualMachine; +import com.cloud.vm.dao.VMInstanceDao; + +public class NASBackupProviderTest { + + private static void overrideDefaultConfigValue(final ConfigKey key, final String value) throws Exception { + final Field field = ConfigKey.class.getDeclaredField("_defaultValue"); + field.setAccessible(true); + field.set(key, value); + final Field cached = ConfigKey.class.getDeclaredField("_value"); + cached.setAccessible(true); + cached.set(key, null); + } + + @Test + public void executionPolicyReachesTheCommand() throws Exception { + final NASBackupProvider provider = new NASBackupProvider(); + final TakeBackupCommand command = new TakeBackupCommand("vm-1", "/backups/vm-1"); + + overrideDefaultConfigValue(NASBackupProvider.NASBackupParallelExecution, "false"); + Assert.assertFalse(provider.applyExecutionPolicy(command, 1L)); + Assert.assertTrue("disabled setting must make the command sequential", command.executeInSequence()); + + overrideDefaultConfigValue(NASBackupProvider.NASBackupParallelExecution, "true"); + Assert.assertTrue(provider.applyExecutionPolicy(command, 1L)); + Assert.assertFalse("enabled setting must let the command run concurrently", command.executeInSequence()); + } + + /** + * The database as every management server sees it: the backups table, a VM placement table and the + * per-host cluster-wide lock (GlobalLock in production, a JVM lock here). + */ + private static final class SharedDb { + final List rows = new ArrayList<>(); + final AtomicLong nextId = new AtomicLong(1); + final Map vms = new HashMap<>(); + final Map hostLocks = new HashMap<>(); + final BackupDao backupDao = Mockito.mock(BackupDao.class); + final VMInstanceDao vmInstanceDao = Mockito.mock(VMInstanceDao.class); + final BackupOfferingDao backupOfferingDao = Mockito.mock(BackupOfferingDao.class); + final VolumeDao volumeDao = Mockito.mock(VolumeDao.class); + + SharedDb() { + Mockito.when(volumeDao.findByInstance(Mockito.anyLong())).thenReturn(Collections.emptyList()); + Mockito.when(backupOfferingDao.findByIdIncludingRemoved(Mockito.anyLong())) + .thenReturn(new BackupOfferingVO(1L, "nas-repo", "nas", "nas", "nas", true)); + Mockito.when(backupDao.persist(Mockito.any(BackupVO.class))).thenAnswer(inv -> { + final BackupVO backup = inv.getArgument(0); + setField(backup, "id", nextId.getAndIncrement()); + synchronized (rows) { + rows.add(backup); + } + return backup; + }); + Mockito.when(backupDao.listByZoneAndStatus(Mockito.anyLong(), Mockito.any(Backup.Status.class))).thenAnswer(inv -> { + final long zoneId = inv.getArgument(0); + final Backup.Status status = inv.getArgument(1); + synchronized (rows) { + return rows.stream().filter(b -> b.getRemoved() == null && b.getZoneId() == zoneId && b.getStatus() == status) + .collect(Collectors.toList()); + } + }); + Mockito.when(backupDao.remove(Mockito.anyLong())).thenAnswer(inv -> { + final long id = inv.getArgument(0); + synchronized (rows) { + rows.stream().filter(b -> b.getId() == id).forEach(b -> b.setRemoved(new Date())); + } + return true; + }); + Mockito.when(backupDao.update(Mockito.anyLong(), Mockito.any(BackupVO.class))).thenReturn(true); + Mockito.when(backupDao.findById(Mockito.anyLong())).thenAnswer(inv -> { + final long id = inv.getArgument(0); + synchronized (rows) { + return rows.stream().filter(b -> b.getId() == id && b.getRemoved() == null).findFirst().orElse(null); + } + }); + Mockito.when(backupDao.listByVmId(Mockito.any(), Mockito.anyLong())).thenAnswer(inv -> { + final long vmId = inv.getArgument(1); + synchronized (rows) { + return rows.stream().filter(b -> b.getRemoved() == null && b.getVmId() == vmId).collect(Collectors.toList()); + } + }); + Mockito.when(vmInstanceDao.findByIdIncludingRemoved(Mockito.anyLong())).thenAnswer(inv -> vms.get((Long) inv.getArgument(0))); + } + + VMInstanceVO vm(final long vmId, final Long hostId, final Long lastHostId) { + final VMInstanceVO vm = Mockito.mock(VMInstanceVO.class); + Mockito.when(vm.getId()).thenReturn(vmId); + Mockito.when(vm.getHostId()).thenReturn(hostId); + Mockito.when(vm.getLastHostId()).thenReturn(lastHostId); + Mockito.when(vm.getDataCenterId()).thenReturn(1L); + Mockito.when(vm.getBackupOfferingId()).thenReturn(1L); + Mockito.when(vm.getInstanceName()).thenReturn("i-2-" + vmId + "-VM"); + vms.put(vmId, vm); + return vm; + } + + int inFlightOn(final long hostId) { + synchronized (rows) { + return (int) rows.stream().filter(b -> b.getRemoved() == null && b.getStatus() == Backup.Status.BackingUp) + .filter(b -> { + final VMInstanceVO vm = vms.get(b.getVmId()); + return vm != null && Long.valueOf(hostId).equals(vm.getHostId() != null ? vm.getHostId() : vm.getLastHostId()); + }).count(); + } + } + + synchronized ReentrantLock hostLock(final long hostId) { + return hostLocks.computeIfAbsent(hostId, id -> new ReentrantLock()); + } + + /** One management server: its own provider instance, sharing nothing with the others except this database. */ + NASBackupProvider managementServer() { + final NASBackupProvider provider = new NASBackupProvider() { + @Override + protected T runUnderHostLock(final long hostId, final Supplier action) { + final ReentrantLock lock = hostLock(hostId); + lock.lock(); + try { + return action.get(); + } finally { + lock.unlock(); + } + } + }; + provider.slotPollMinMillis = 5L; + provider.slotPollMaxMillis = 20L; + setField(provider, "backupDao", backupDao); + setField(provider, "vmInstanceDao", vmInstanceDao); + setField(provider, "backupOfferingDao", backupOfferingDao); + setField(provider, "volumeDao", volumeDao); + return provider; + } + } + + private static void setField(final Object target, final String name, final Object value) { + Class type = target.getClass(); + while (type != null) { + try { + final Field field = type.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + return; + } catch (NoSuchFieldException e) { + type = type.getSuperclass(); + } catch (IllegalAccessException e) { + throw new IllegalStateException(e); + } + } + throw new IllegalStateException("no field " + name); + } + + @Test + public void perHostCapHoldsAndReleases() { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = db.managementServer(); + final BackupVO first = provider.reserveHostBackupSlot(db.vm(1L, 7L, 7L), 7L, "a", 2, 1); + provider.reserveHostBackupSlot(db.vm(2L, 7L, 7L), 7L, "b", 2, 1); + Assert.assertEquals(2, provider.countInFlightBackups(7L, 1L)); + try { + provider.reserveHostBackupSlot(db.vm(3L, 7L, 7L), 7L, "c", 2, 1); + Assert.fail("third backup on a host capped at 2 must not get a slot"); + } catch (CloudRuntimeException expected) { + Assert.assertTrue(expected.getMessage().contains("host 7")); + } + Assert.assertEquals("a refused reservation must not leave a row behind", 2, db.rows.size()); + first.setStatus(Backup.Status.BackedUp); + provider.reserveHostBackupSlot(db.vm(4L, 7L, 7L), 7L, "d", 2, 1); + Assert.assertEquals(2, provider.countInFlightBackups(7L, 1L)); + } + + @Test + public void hostsAreIndependentAndStoppedVmsCountOnTheirLastHost() { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = db.managementServer(); + provider.reserveHostBackupSlot(db.vm(1L, 1L, 1L), 1L, "a", 1, 1); + provider.reserveHostBackupSlot(db.vm(2L, null, 2L), 2L, "b", 1, 1); + Assert.assertEquals(1, provider.countInFlightBackups(1L, 1L)); + Assert.assertEquals(1, provider.countInFlightBackups(2L, 1L)); + Assert.assertEquals(0, provider.countInFlightBackups(3L, 1L)); + } + + @Test + public void staleAndForeignRowsDoNotHoldASlot() { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = db.managementServer(); + final BackupVO stale = provider.reserveHostBackupSlot(db.vm(1L, 5L, 5L), 5L, "a", 1, 1); + stale.setDate(new Date(System.currentTimeMillis() - (NASBackupProvider.abandonedAfterSeconds(provider.takeBackupCommandWaitSeconds(1L)) + 60L) * 1000L)); + Assert.assertEquals("an abandoned BackingUp row must not hold a slot", 0, provider.countInFlightBackups(5L, 1L)); + + final BackupVO other = provider.reserveHostBackupSlot(db.vm(2L, 5L, 5L), 5L, "b", 1, 1); + Mockito.when(db.backupOfferingDao.findByIdIncludingRemoved(Mockito.anyLong())) + .thenReturn(new BackupOfferingVO(1L, "x", "veeam", "veeam", "veeam", true)); + Assert.assertEquals("backups of another provider must not count", 0, provider.countInFlightBackups(5L, 1L)); + Assert.assertNotNull(other); + } + + @Test + public void waiterProceedsWhenASlotFrees() throws Exception { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = db.managementServer(); + final BackupVO held = provider.reserveHostBackupSlot(db.vm(1L, 9L, 9L), 9L, "a", 1, 10); + final VMInstanceVO second = db.vm(2L, 9L, 9L); + final CountDownLatch acquired = new CountDownLatch(1); + final AtomicBoolean failed = new AtomicBoolean(false); + final Thread waiter = new Thread(() -> { + try { + provider.reserveHostBackupSlot(second, 9L, "b", 1, 10); + acquired.countDown(); + } catch (CloudRuntimeException e) { + failed.set(true); + } + }); + waiter.start(); + Assert.assertFalse("waiter must block while the only slot is held", acquired.await(300, TimeUnit.MILLISECONDS)); + held.setStatus(Backup.Status.BackedUp); + provider.wakeLocalSlotWaiters(); + Assert.assertTrue("waiter must get the slot once it is released", acquired.await(5, TimeUnit.SECONDS)); + Assert.assertFalse(failed.get()); + Assert.assertEquals(1, provider.countInFlightBackups(9L, 1L)); + waiter.join(1000); + } + + /** + * Two management servers, each with its own provider and no shared memory, send backups for the same host + * at the same time. The combined in-flight count read from the shared database must never exceed the cap. + */ + @Test + public void capHoldsAcrossManagementServers() throws Exception { + final SharedDb db = new SharedDb(); + final NASBackupProvider ms1 = db.managementServer(); + final NASBackupProvider ms2 = db.managementServer(); + final int max = 2; + final int backups = 12; + final AtomicInteger peak = new AtomicInteger(); + final AtomicInteger done = new AtomicInteger(); + final List errors = Collections.synchronizedList(new ArrayList<>()); + final ExecutorService pool = Executors.newFixedThreadPool(backups); + final CountDownLatch start = new CountDownLatch(1); + for (int i = 0; i < backups; i++) { + final NASBackupProvider provider = i % 2 == 0 ? ms1 : ms2; + final VMInstanceVO vm = db.vm(100L + i, 42L, 42L); + pool.submit(() -> { + try { + start.await(); + final BackupVO backup = provider.reserveHostBackupSlot(vm, 42L, "p" + vm.getId(), max, 30); + peak.accumulateAndGet(db.inFlightOn(42L), Math::max); + Thread.sleep(20); + peak.accumulateAndGet(db.inFlightOn(42L), Math::max); + backup.setStatus(Backup.Status.BackedUp); + provider.wakeLocalSlotWaiters(); + done.incrementAndGet(); + } catch (Throwable t) { + errors.add(t); + } + return null; + }); + } + start.countDown(); + pool.shutdown(); + Assert.assertTrue(pool.awaitTermination(60, TimeUnit.SECONDS)); + Assert.assertTrue("unexpected errors: " + errors, errors.isEmpty()); + Assert.assertEquals(backups, done.get()); + Assert.assertTrue("peak " + peak.get() + " exceeded the cap of " + max, peak.get() <= max); + Assert.assertEquals(max, peak.get()); + } + + /** A provider wired to one host (id 11) and an agent that answers every command with the given behaviour. */ + private static NASBackupProvider wiredToHost(final SharedDb db, final Answer agentBehaviour) throws Exception { + final NASBackupProvider provider = db.managementServer(); + final HostDao hostDao = Mockito.mock(HostDao.class); + final HostVO host = Mockito.mock(HostVO.class); + Mockito.when(host.getId()).thenReturn(11L); + Mockito.when(host.getStatus()).thenReturn(Status.Up); + Mockito.when(host.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM); + Mockito.when(hostDao.findById(11L)).thenReturn(host); + final BackupRepositoryDao repositoryDao = Mockito.mock(BackupRepositoryDao.class); + Mockito.when(repositoryDao.findByBackupOfferingId(Mockito.anyLong())).thenReturn(Mockito.mock(BackupRepository.class)); + final AgentManager agentManager = Mockito.mock(AgentManager.class); + Mockito.when(agentManager.send(Mockito.anyLong(), Mockito.any(Command.class))).thenAnswer(agentBehaviour); + setField(provider, "hostDao", hostDao); + setField(provider, "backupRepositoryDao", repositoryDao); + setField(provider, "agentManager", agentManager); + overrideDefaultConfigValue(NASBackupProvider.NASBackupParallelExecution, "true"); + return provider; + } + + private static VMInstanceVO runningVm(final SharedDb db, final long vmId) { + final VMInstanceVO vm = db.vm(vmId, 11L, 11L); + Mockito.when(vm.getState()).thenReturn(VirtualMachine.State.Running); + return vm; + } + + /** + * Damans227's case: the management server stops waiting while the host is still writing the backup. The row + * must stay BackingUp and keep its slot, on this and on every other management server. + */ + @Test + public void timedOutBackupKeepsItsRowAndItsSlot() throws Exception { + final SharedDb db = new SharedDb(); + final AtomicInteger sentWait = new AtomicInteger(-1); + final NASBackupProvider ms1 = wiredToHost(db, inv -> { + sentWait.set(((Command) inv.getArgument(1)).getWait()); + throw new OperationTimedoutException(null, 11L, 1L, 1, false); + }); + try { + ms1.takeBackup(runningVm(db, 1L)); + Assert.fail("a backup that did not answer in time must not report success"); + } catch (CloudRuntimeException expected) { + Assert.assertTrue(expected.getMessage().contains("may still be running")); + } + Assert.assertEquals("the command must wait backup.nas.backup.timeout, not the global wait", + Integer.parseInt(NASBackupProvider.NASBackupTimeout.defaultValue()), sentWait.get()); + Assert.assertEquals(1, db.rows.size()); + final BackupVO row = db.rows.get(0); + Assert.assertEquals("the backup must stay visible while the host may still run it", Backup.Status.BackingUp, row.getStatus()); + Assert.assertNull("the backup must not be removed", row.getRemoved()); + Assert.assertEquals("the slot must stay held", 1, ms1.countInFlightBackups(11L, 1L)); + + final NASBackupProvider ms2 = db.managementServer(); + try { + ms2.reserveHostBackupSlot(runningVm(db, 2L), 11L, "b", 1, 0); + Assert.fail("another management server must see the slot as taken"); + } catch (CloudRuntimeException expected) { + Assert.assertTrue(expected.getMessage().contains("1 of 1 in flight")); + } + Assert.assertEquals("the sync task must leave a backup the host may still be running alone", 0, ms2.markAbandonedBackups(db.vms.get(1L))); + Assert.assertEquals(Backup.Status.BackingUp, row.getStatus()); + } + + @Test + public void backupTheHostReportsFailedReleasesItsSlot() throws Exception { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = wiredToHost(db, inv -> new BackupAnswer(inv.getArgument(1), false, "disk full")); + Assert.assertFalse(provider.takeBackup(runningVm(db, 1L))); + Assert.assertEquals(1, db.rows.size()); + Assert.assertEquals(Backup.Status.Failed, db.rows.get(0).getStatus()); + Assert.assertNotNull(db.rows.get(0).getRemoved()); + Assert.assertEquals("a backup the host reported failed must not hold a slot", 0, provider.countInFlightBackups(11L, 1L)); + } + + @Test + public void backupThatNeverReachedTheHostReleasesItsSlot() throws Exception { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = wiredToHost(db, inv -> { + throw new AgentUnavailableException(11L); + }); + try { + provider.takeBackup(runningVm(db, 1L)); + Assert.fail("an unreachable host must fail the backup"); + } catch (CloudRuntimeException expected) { + Assert.assertTrue(expected.getMessage().contains("Unable to contact")); + } + Assert.assertNotNull(db.rows.get(0).getRemoved()); + Assert.assertEquals(0, provider.countInFlightBackups(11L, 1L)); + } + + @Test + public void successfulBackupIsBackedUpAndReleasesItsSlot() throws Exception { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = wiredToHost(db, inv -> { + final BackupAnswer answer = new BackupAnswer(inv.getArgument(1), true, "42"); + answer.setSize(42L); + return answer; + }); + Assert.assertTrue(provider.takeBackup(runningVm(db, 1L))); + Assert.assertEquals(Backup.Status.BackedUp, db.rows.get(0).getStatus()); + Assert.assertNull(db.rows.get(0).getRemoved()); + Assert.assertEquals(0, provider.countInFlightBackups(11L, 1L)); + } + + /** + * A BackingUp row twice the wait old has been given up by its management server (or that management server + * stopped) and the agent has stopped the script: it frees its slot and the sync task marks it Error, keeping it. + */ + @Test + public void abandonedBackupIsMarkedErrorAndKept() throws Exception { + final SharedDb db = new SharedDb(); + final NASBackupProvider provider = db.managementServer(); + final VMInstanceVO vm = runningVm(db, 1L); + final long wait = provider.takeBackupCommandWaitSeconds(1L); + final BackupVO pastWait = provider.reserveHostBackupSlot(vm, 11L, "a", 5, 1); + pastWait.setDate(new Date(System.currentTimeMillis() - (wait + 60L) * 1000L)); + final BackupVO abandoned = provider.reserveHostBackupSlot(vm, 11L, "b", 5, 1); + abandoned.setDate(new Date(System.currentTimeMillis() - (2 * wait + 60L) * 1000L)); + final BackupVO done = provider.reserveHostBackupSlot(vm, 11L, "c", 5, 1); + done.setDate(new Date(System.currentTimeMillis() - (3 * wait) * 1000L)); + done.setStatus(Backup.Status.BackedUp); + + Assert.assertEquals("only the row past the wait but not abandoned holds a slot", 1, provider.countInFlightBackups(11L, 1L)); + provider.syncBackups(vm, null); + Assert.assertEquals(Backup.Status.BackingUp, pastWait.getStatus()); + Assert.assertEquals(Backup.Status.Error, abandoned.getStatus()); + Assert.assertNull("an abandoned backup stays in the list so it can be seen and deleted", abandoned.getRemoved()); + Assert.assertEquals(Backup.Status.BackedUp, done.getStatus()); + Assert.assertEquals("marking is idempotent across management servers", 0, db.managementServer().markAbandonedBackups(vm)); + } + + @Test + public void abandonedWindowIsTwiceTheWaitAndAtLeastFiveMinutesMore() { + Assert.assertEquals(14400L, NASBackupProvider.abandonedAfterSeconds(7200)); + Assert.assertEquals(360L, NASBackupProvider.abandonedAfterSeconds(60)); + } + + @Test + public void commandsTimeoutForTakeBackupBoundsTheStaleWindow() throws Exception { + final NASBackupProvider provider = new NASBackupProvider(); + overrideDefaultConfigValue(AgentManager.GranularWaitTimeForCommands, "CopyCommand=60, TakeBackupCommand=14400"); + try { + Assert.assertEquals(14400, provider.takeBackupCommandWaitSeconds(1L)); + } finally { + overrideDefaultConfigValue(AgentManager.GranularWaitTimeForCommands, ""); + } + Assert.assertEquals(Integer.parseInt(NASBackupProvider.NASBackupTimeout.defaultValue()), provider.takeBackupCommandWaitSeconds(1L)); + overrideDefaultConfigValue(NASBackupProvider.NASBackupTimeout, "0"); + try { + Assert.assertEquals(Integer.parseInt(AgentManager.Wait.defaultValue()), provider.takeBackupCommandWaitSeconds(1L)); + } finally { + overrideDefaultConfigValue(NASBackupProvider.NASBackupTimeout, "7200"); + } + } + + @Test + public void hostLockNameIsPerHost() { + Assert.assertEquals("nas-backup-host-42", NASBackupProvider.hostBackupLockName(42L)); + } + + /** + * Spring ignores @Inject on static fields, so an annotation that drifts onto a ConfigKey constant + * silently leaves the DAO it used to belong to unset (hostDao was null here once). + */ + @Test + public void everyInjectedFieldIsAnInstanceFieldAndHostDaoIsInjected() throws Exception { + for (final Field field : NASBackupProvider.class.getDeclaredFields()) { + if (field.isAnnotationPresent(Inject.class)) { + Assert.assertFalse("@Inject on static field " + field.getName() + " is ignored by Spring", + Modifier.isStatic(field.getModifiers())); + } + if (field.getType().getSimpleName().endsWith("Dao") || field.getType().getSimpleName().endsWith("Manager")) { + Assert.assertTrue("dependency field " + field.getName() + " is not annotated with @Inject", + field.isAnnotationPresent(Inject.class)); + } + } + final Field hostDao = NASBackupProvider.class.getDeclaredField("hostDao"); + Assert.assertEquals(HostDao.class, hostDao.getType()); + Assert.assertTrue(hostDao.isAnnotationPresent(Inject.class)); + } +}