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

engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDao.java

@@ -35,5 +35,6 @@ public interface BackupDao extends GenericDao<BackupVO, Long> {
     List<Backup> syncBackups(Long zoneId, Long vmId, List<Backup> externalBackups);
     BackupVO getBackupVO(Backup backup);
     List<Backup> listByOfferingId(Long backupOfferingId);
+    List<BackupVO> listByZoneAndStatus(long zoneId, Backup.Status status);
     BackupResponse newBackupResponse(Backup backup);
 }

engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupDaoImpl.java

@@ -60,6 +60,7 @@ public class BackupDaoImpl extends GenericDaoBase<BackupVO, Long> implements Bac
     BackupOfferingDao backupOfferingDao;
 
     private SearchBuilder<BackupVO> backupSearch;
+    private SearchBuilder<BackupVO> 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<BackupVO> listByZoneAndStatus(long zoneId, Backup.Status status) {
+        SearchCriteria<BackupVO> sc = zoneStatusSearch.create();
+        sc.setParameters("zone_id", zoneId);
+        sc.setParameters("status", status);
+        return listBy(sc);
     }
 
     @Override

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<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 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<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);
+
+    /** 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<VolumeVO> 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> T runUnderHostLock(final long hostId, final Supplier<T> 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<BackupVO> 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
         };
     }