NAS backup: run take-backup commands concurrently on the KVM host, bounded per host

core/src/main/java/org/apache/cloudstack/backup/TakeBackupCommand.java

@@ -32,6 +32,7 @@ public class TakeBackupCommand extends Command {
     private List<String> 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<String> 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;
     }
 }

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());
+    }
+}

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<Boolean> 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<Integer> 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<Integer> 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<Long, HostBackupSlots> 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<VolumeVO> 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
         };
     }
 

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));
+    }
+}