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/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..b26c61927e6d 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 @@ -49,6 +49,7 @@ import org.apache.logging.log4j.LogManager; import javax.inject.Inject; import java.text.SimpleDateFormat; +import java.util.concurrent.ConcurrentHashMap; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -74,6 +75,35 @@ public class NASBackupProvider extends AdapterBase implements BackupProvider, Co private BackupOfferingDao backupOfferingDao; @Inject + 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. Keep it below the agent's worker thread count (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); + + /** In-flight take-backup commands per host, so backups never occupy every agent worker thread. */ + private final ConcurrentHashMap hostBackupSlots = new ConcurrentHashMap<>(); + + private static final class HostBackupSlots { + private int inFlight; + } + private HostDao hostDao; @Inject @@ -154,11 +184,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,13 +198,26 @@ public boolean takeBackup(final VirtualMachine vm) { command.setVolumePaths(volumePaths); } + // Concurrent backups are bounded per host: wait here, on the management server, for one of the + // host's slots so the agent's worker threads are never all taken by long-running backups. + boolean slotHeld = false; + if (parallel) { + acquireHostBackupSlot(host.getId(), NASBackupParallelMaxPerHost.valueIn(zoneId), NASBackupParallelQueueTimeout.valueIn(zoneId)); + slotHeld = true; + } + BackupVO backupVO = null; BackupAnswer answer = null; try { + backupVO = createBackupObject(vm, backupPath); 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"); + } finally { + if (slotHeld) { + releaseHostBackupSlot(host.getId()); + } } if (answer != null && answer.getResult()) { @@ -189,6 +233,70 @@ public boolean takeBackup(final VirtualMachine vm) { return Objects.nonNull(answer) && answer.getResult(); } + /** + * 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; + } + + /** + * Waits for one of the host's backup slots. The caller holds it for the whole agent round-trip and + * releases it in a finally block. The wait happens on the management server (the async job thread), + * never on the agent's worker threads. Fails with CloudRuntimeException once the timeout is reached. + */ + protected void acquireHostBackupSlot(final long hostId, 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 HostBackupSlots slots = hostBackupSlots.computeIfAbsent(hostId, id -> new HostBackupSlots()); + final long deadline = System.currentTimeMillis() + timeout * 1000L; + synchronized (slots) { + while (slots.inFlight >= max) { + 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, slots.inFlight, max, NASBackupParallelMaxPerHost.key(), NASBackupParallelQueueTimeout.key())); + } + try { + slots.wait(remaining); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new CloudRuntimeException("Interrupted while waiting for a NAS backup slot on host " + hostId); + } + } + slots.inFlight++; + LOG.debug("Host {} now has {} of {} NAS backups in flight", hostId, slots.inFlight, max); + } + } + + protected void releaseHostBackupSlot(final long hostId) { + final HostBackupSlots slots = hostBackupSlots.get(hostId); + if (slots == null) { + return; + } + synchronized (slots) { + if (slots.inFlight > 0) { + slots.inFlight--; + } + slots.notifyAll(); + } + } + + protected int getInFlightBackups(final long hostId) { + final HostBackupSlots slots = hostBackupSlots.get(hostId); + if (slots == null) { + return 0; + } + synchronized (slots) { + return slots.inFlight; + } + } + private BackupVO createBackupObject(VirtualMachine vm, String backupPath) { BackupVO backup = new BackupVO(); backup.setVmId(vm.getId()); @@ -450,6 +558,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..f70e5ee642b9 --- /dev/null +++ b/plugins/backup/nas/src/test/java/org/apache/cloudstack/backup/NASBackupProviderTest.java @@ -0,0 +1,114 @@ +// 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.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.cloudstack.framework.config.ConfigKey; +import org.junit.Assert; +import org.junit.Test; + +import com.cloud.utils.exception.CloudRuntimeException; + +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); + } + + @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()); + } + + @Test + public void perHostCapHoldsAndReleases() { + final NASBackupProvider provider = new NASBackupProvider(); + provider.acquireHostBackupSlot(7L, 2, 1); + provider.acquireHostBackupSlot(7L, 2, 1); + Assert.assertEquals(2, provider.getInFlightBackups(7L)); + try { + provider.acquireHostBackupSlot(7L, 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(2, provider.getInFlightBackups(7L)); + provider.releaseHostBackupSlot(7L); + provider.acquireHostBackupSlot(7L, 2, 1); + Assert.assertEquals(2, provider.getInFlightBackups(7L)); + provider.releaseHostBackupSlot(7L); + provider.releaseHostBackupSlot(7L); + Assert.assertEquals(0, provider.getInFlightBackups(7L)); + } + + @Test + public void hostsAreIndependent() { + final NASBackupProvider provider = new NASBackupProvider(); + provider.acquireHostBackupSlot(1L, 1, 1); + provider.acquireHostBackupSlot(2L, 1, 1); + Assert.assertEquals(1, provider.getInFlightBackups(1L)); + Assert.assertEquals(1, provider.getInFlightBackups(2L)); + provider.releaseHostBackupSlot(1L); + provider.releaseHostBackupSlot(2L); + } + + @Test + public void waiterProceedsWhenASlotFrees() throws Exception { + final NASBackupProvider provider = new NASBackupProvider(); + provider.acquireHostBackupSlot(9L, 1, 10); + final CountDownLatch acquired = new CountDownLatch(1); + final AtomicBoolean failed = new AtomicBoolean(false); + final Thread waiter = new Thread(() -> { + try { + provider.acquireHostBackupSlot(9L, 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)); + provider.releaseHostBackupSlot(9L); + Assert.assertTrue("waiter must get the slot once it is released", acquired.await(5, TimeUnit.SECONDS)); + Assert.assertFalse(failed.get()); + Assert.assertEquals(1, provider.getInFlightBackups(9L)); + provider.releaseHostBackupSlot(9L); + waiter.join(1000); + } + + @Test + public void releaseOnUnknownHostIsHarmless() { + final NASBackupProvider provider = new NASBackupProvider(); + provider.releaseHostBackupSlot(404L); + Assert.assertEquals(0, provider.getInFlightBackups(404L)); + } +}