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..5db20fa71dab 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,52 @@ 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); + + /** 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; + + /** + * A BackingUp row older than the take-backup command timeout plus this margin is no longer awaited by any + * management server (for example its management server stopped mid-backup), so it does not hold a slot. + */ + private static final int STALE_RESERVATION_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 +196,12 @@ 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); if (VirtualMachine.State.Stopped.equals(vm.getState())) { List vmVolumes = volumeDao.findByInstance(vm.getId()); @@ -167,26 +210,188 @@ 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; 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) { + throw new CloudRuntimeException("Operation to initiate backup timed out, please try again"); + } + 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) { + // Releases the slot: a failed or interrupted backup must never stay BackingUp. + 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. Rows older than the take-backup command + * timeout are left out: no management server is waiting for them any more. + */ + 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 staleBefore = System.currentTimeMillis() - (takeBackupCommandWaitSeconds() + (long) STALE_RESERVATION_MARGIN_SECONDS) * 1000L; + int inFlight = 0; + for (final BackupVO backup : backingUp) { + if (backup.getDate() != null && backup.getDate().getTime() < staleBefore) { + LOG.debug("Not counting stale 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; + } + + /** The longest a management server waits for a TakeBackupCommand answer: commands.timeout if set for it, else wait. */ + protected int takeBackupCommandWaitSeconds() { + int 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) { @@ -450,6 +655,9 @@ public boolean isValidProviderOffering(Long zoneId, String uuid) { @Override public ConfigKey[] getConfigKeys() { return new ConfigKey[]{ + NASBackupParallelExecution, + NASBackupParallelMaxPerHost, + NASBackupParallelQueueTimeout }; } 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..06d46a59d663 --- /dev/null +++ b/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java @@ -0,0 +1,384 @@ +// 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 com.cloud.agent.AgentManager; +import com.cloud.agent.api.Command; +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(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() - (provider.takeBackupCommandWaitSeconds() + 3600L) * 1000L)); + Assert.assertEquals("a BackingUp row nobody waits for any more 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()); + } + + @Test + public void failedBackupReleasesItsSlot() throws Exception { + final SharedDb db = new SharedDb(); + 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))) + .thenThrow(new OperationTimedoutException(null, 11L, 1L, 1, false)); + setField(provider, "hostDao", hostDao); + setField(provider, "backupRepositoryDao", repositoryDao); + setField(provider, "agentManager", agentManager); + overrideDefaultConfigValue(NASBackupProvider.NASBackupParallelExecution, "true"); + + final VMInstanceVO vm = db.vm(1L, 11L, 11L); + Mockito.when(vm.getState()).thenReturn(VirtualMachine.State.Running); + try { + provider.takeBackup(vm); + Assert.fail("a timed out backup must fail"); + } catch (CloudRuntimeException expected) { + Assert.assertTrue(expected.getMessage().contains("timed out")); + } + 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("the failed backup must not hold a slot", 0, provider.countInFlightBackups(11L, 1L)); + } + + @Test + public void commandsTimeoutForTakeBackupBoundsTheStaleWindow() throws Exception { + final NASBackupProvider provider = new NASBackupProvider(); + overrideDefaultConfigValue(AgentManager.GranularWaitTimeForCommands, "CopyCommand=60, TakeBackupCommand=14400"); + try { + Assert.assertEquals(14400, provider.takeBackupCommandWaitSeconds()); + } finally { + overrideDefaultConfigValue(AgentManager.GranularWaitTimeForCommands, ""); + } + Assert.assertEquals(Integer.parseInt(AgentManager.Wait.defaultValue()), provider.takeBackupCommandWaitSeconds()); + } + + @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)); + } +}