fix(nas-backup): address Copilot review on PR #13074

* nasbackup.sh: emit INCREMENTAL_FALLBACK= and BITMAP_CREATED= on stdout
  (Script.executePipedCommands in LibvirtTakeBackupCommandWrapper reads
  stdout only, not stderr — so these markers were never parsed). Fixes
  stopped-VM fallback recording and bitmap persistence.

* nasbackup.sh: fix `2>&1 > /dev/null` → `> /dev/null 2>&1` (the original
  ordering leaves stderr pointed at the now-redirected stdout, dropping
  virsh error details).

* nasbackup.sh: capture qemu-agent thaw response correctly (was being
  swallowed by `> /dev/null`).

* nasbackup.sh: usage text now reflects the implemented --parent-paths
  (plural, comma-separated) flag instead of misleading --parent-path.

* NASBackupProvider: default nas.backup.incremental.enabled to false so
  existing zones keep legacy full-only behavior on upgrade; opt in
  per-zone when ready. Description updated to make this explicit.

* NASBackupProvider.composeParentBackupPaths: sort current VM volumes by
  deviceId before positional comparison with parent's backed-up volumes,
  and verify per-position UUID alignment. If any disk was detached and a
  different one attached in its place, return null and force a full
  instead of silently rebasing onto the wrong parent file.

* NASBackupChainKeys.TYPE: doc now explicitly notes the lowercase casing
  difference from Backup.Status uppercase enum values.

* test_backup_recovery_nas.py: capture original zone-scoped
  nas.backup.full.every via Configurations.list at the start of each
  incremental test and restore that exact value in finally, instead of
  hardcoding 10 (which leaked into shared environments).

Author: jmsperu

plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupChainKeys.java

@@ -36,7 +36,13 @@ public final class NASBackupChainKeys {
     /** Position within the chain: 0 for the full, 1 for the first incremental, and so on. */
     public static final String CHAIN_POSITION = "nas.chain_position";
 
-    /** Backup type marker: {@value #TYPE_FULL} or {@value #TYPE_INCREMENTAL}. Mirrors {@code backups.type} for fast lookup without a join. */
+    /**
+     * Backup type marker: {@value #TYPE_FULL} or {@value #TYPE_INCREMENTAL}.
+     * Stored in {@code backup_details} as a lowercase string for fast lookup without a join
+     * against {@code backups.type} (which is an upper-case Backup.Status enum value).
+     * NOTE: do NOT compare directly against {@code Backup.Status} string casing; use the
+     * constants below.
+     */
     public static final String TYPE = "nas.type";
 
     public static final String TYPE_FULL = "full";

plugins/backup/nas/src/main/java/org/apache/cloudstack/backup/NASBackupProvider.java

@@ -100,11 +100,13 @@ public class NASBackupProvider extends AdapterBase implements BackupProvider, Co
 
     ConfigKey<Boolean> NASBackupIncrementalEnabled = new ConfigKey<>("Advanced", Boolean.class,
             "nas.backup.incremental.enabled",
-            "true",
-            "Master switch for NAS incremental backups. When false, every NAS backup is taken as a full " +
-                    "regardless of nas.backup.full.every. Toggling this is safe at any time: switching off " +
-                    "forces the next backup to be a fresh full anchor (existing chains stay restorable), " +
-                    "switching back on resumes incrementals on the next full + incremental cycle.",
+            "false",
+            "Master switch for NAS incremental backups. Defaults to false so existing zones keep the " +
+                    "legacy full-only behavior on upgrade; opt in per-zone when ready to use chains. " +
+                    "When false, every NAS backup is taken as a full regardless of nas.backup.full.every. " +
+                    "Toggling this is safe at any time: switching off forces the next backup to be a fresh " +
+                    "full anchor (existing chains stay restorable), switching back on resumes incrementals " +
+                    "on the next full + incremental cycle.",
             true,
             ConfigKey.Scope.Zone,
             BackupFrameworkEnabled.key());
@@ -391,20 +393,38 @@ private List<String> composeParentBackupPaths(Backup parent, long vmId) {
             return null;
         }
 
-        // Sanity: the current VM must have the same number of volumes. If it doesn't (volume
-        // added or removed since the parent), positional alignment is unsafe — fall back to
-        // full at the caller.
+        // Sanity 1: the current VM must have the same number of volumes. If it doesn't (volume
+        // added or removed since the parent), positional alignment is unsafe — caller falls back to full.
         List<VolumeVO> currentVols = volumeDao.findByInstance(vmId);
         if (currentVols == null || currentVols.size() != parentVols.size()) {
             return null;
         }
 
-        List<String> paths = new ArrayList<>(parentVols.size());
+        // Sanity 2: VolumeDao.findByInstance() has no ordering guarantee. Sort the current
+        // volumes by deviceId so positional comparison against parentVols (also deviceId-ordered
+        // at backup time) is meaningful.
+        List<VolumeVO> currentSorted = new ArrayList<>(currentVols);
+        currentSorted.sort(Comparator.comparing(Volume::getDeviceId));
+
+        // Sanity 3: verify each current volume's UUID matches the parent's recorded UUID at the
+        // same position. If any disk was detached + a different one attached in its place, the
+        // chain cannot be safely continued — force a full instead of silently rebasing onto the
+        // wrong parent file.
         for (int i = 0; i < parentVols.size(); i++) {
-            String volUuid = parentVols.get(i).getUuid();
-            if (volUuid == null || volUuid.isEmpty()) {
+            String parentUuid = parentVols.get(i).getUuid();
+            String currentUuid = currentSorted.get(i).getUuid();
+            if (parentUuid == null || parentUuid.isEmpty()
+                    || currentUuid == null || !parentUuid.equals(currentUuid)) {
+                LOG.debug("Volume identity mismatch at position {} for VM {}: parent uuid {} vs current uuid {}. " +
+                        "Forcing a full backup to start a fresh chain.",
+                        i, vmId, parentUuid, currentUuid);
                 return null;
             }
+        }
+
+        List<String> paths = new ArrayList<>(parentVols.size());
+        for (int i = 0; i < parentVols.size(); i++) {
+            String volUuid = parentVols.get(i).getUuid();
             String prefix = (i == 0) ? "root" : "datadisk";
             paths.add(dir + "/" + prefix + "." + volUuid + ".qcow2");
         }

scripts/vm/hypervisor/kvm/nasbackup.sh

@@ -237,17 +237,19 @@ XML
   # Start push backup, atomically registering the new checkpoint when applicable.
   local backup_begin=0
   if [[ $make_checkpoint -eq 1 ]]; then
-    if virsh -c qemu:///system backup-begin --domain $VM --backupxml $dest/backup.xml --checkpointxml $dest/checkpoint.xml 2>&1 > /dev/null; then
+    # Order matters: redirect stdout to /dev/null first, then merge stderr into stdout.
+    # The reversed `2>&1 > /dev/null` form leaves stderr pointing at the original tty.
+    if virsh -c qemu:///system backup-begin --domain $VM --backupxml $dest/backup.xml --checkpointxml $dest/checkpoint.xml > /dev/null 2>&1; then
       backup_begin=1;
     fi
   else
-    if virsh -c qemu:///system backup-begin --domain $VM --backupxml $dest/backup.xml 2>&1 > /dev/null; then
+    if virsh -c qemu:///system backup-begin --domain $VM --backupxml $dest/backup.xml > /dev/null 2>&1; then
       backup_begin=1;
     fi
   fi
 
   if [[ $thaw -eq 1 ]]; then
-    if ! response=$(virsh -c qemu:///system qemu-agent-command "$VM" '{"execute":"guest-fsfreeze-thaw"}' 2>&1 > /dev/null); then
+    if ! response=$(virsh -c qemu:///system qemu-agent-command "$VM" '{"execute":"guest-fsfreeze-thaw"}' 2>&1); then
       echo "Failed to thaw the filesystem for vm $VM: $response"
       cleanup
       exit 1
@@ -358,7 +360,9 @@ backup_stopped_vm() {
   # to full and signal the fallback so the orchestrator can record it as a full
   # in the chain.
   if [[ "$MODE" == "incremental" ]]; then
-    echo "INCREMENTAL_FALLBACK=full (VM stopped — incremental requires running VM)" >&2
+    # Emit on stdout so Script.executePipedCommands in LibvirtTakeBackupCommandWrapper
+    # can parse it and record the backup as FULL.
+    echo "INCREMENTAL_FALLBACK=full (VM stopped — incremental requires running VM)"
   fi
 
   mount_operation
@@ -404,8 +408,10 @@ backup_stopped_vm() {
   # Surface the bitmap name we created so the orchestrator can persist it as
   # the VM's active_checkpoint_id. Empty when sources weren't file-backed or
   # qemu-img bitmap failed — orchestrator handles either case.
+  # Stdout (not stderr) so Script.executePipedCommands in the Java wrapper
+  # can parse it — matches the backup_running_vm path.
   if [[ "$bitmap_seeded" == "1" ]]; then
-    echo "BITMAP_CREATED=$BITMAP_NEW" >&2
+    echo "BITMAP_CREATED=$BITMAP_NEW"
   fi
 
   ls -l --numeric-uid-gid $dest | awk '{print $5}'
@@ -505,12 +511,14 @@ cleanup() {
 function usage {
   echo ""
   echo "Usage: $0 -o <operation> -v|--vm <domain name> -t <storage type> -s <storage address> -m <mount options> -p <backup path> -d <disks path> -q|--quiesce <true|false>"
-  echo "         [-M|--mode <full|incremental>] [--bitmap-new <name>] [--bitmap-parent <name>] [--parent-path <path>]"
+  echo "         [-M|--mode <full|incremental>] [--bitmap-new <name>] [--bitmap-parent <name>] [--parent-paths <p1,p2,...>]"
   echo ""
   echo "Incremental backup options (running VMs only; requires QEMU >= 4.2 and libvirt >= 7.2):"
   echo "  -M|--mode full          Take a full backup AND create a checkpoint (--bitmap-new required) for future incrementals."
   echo "  -M|--mode incremental   Take an incremental backup since --bitmap-parent and create new checkpoint --bitmap-new."
-  echo "                          Requires --bitmap-parent, --bitmap-new, and --parent-path (parent backup file for rebase)."
+  echo "                          Requires --bitmap-parent, --bitmap-new, and --parent-paths (comma-separated list, one"
+  echo "                          parent qcow2 path per disk: root.<uuid>.qcow2, datadisk.<uuid>.qcow2, … same order"
+  echo "                          as -d|--disks)."
   echo "  Without -M, behaves as legacy full-only backup with no checkpoint creation."
   echo ""
   exit 1

test/integration/smoke/test_backup_recovery_nas.py

@@ -274,13 +274,25 @@ def test_vm_backup_create_vm_from_backup_in_another_zone(self):
     # repair, refuse-delete-full-with-children, and stopped-VM fallback.
     #
     # All tests set nas.backup.full.every to a small value (3) so a chain
-    # forms quickly without needing many backup iterations. They restore
-    # the original value at teardown.
+    # forms quickly without needing many backup iterations. The original
+    # zone value (whatever the test environment has configured) is captured
+    # before the test runs and restored verbatim in finally, so we don't
+    # leak config changes across tests on shared environments.
 
     def _set_full_every(self, value):
         Configurations.update(self.apiclient, name='nas.backup.full.every',
                               value=str(value), zoneid=self.zone.id)
 
+    def _get_full_every(self):
+        """Read the current zone-scoped (or global fallback) value of nas.backup.full.every."""
+        configs = Configurations.list(self.apiclient, name='nas.backup.full.every',
+                                      zoneid=self.zone.id)
+        if configs and len(configs) > 0 and configs[0].value is not None:
+            return configs[0].value
+        # Fall back to global default — Configurations.list returns the global value
+        # when no zone override exists. Defensive fallback to '10' (the framework default).
+        return '10'
+
     def _backup_type(self, backup):
         # Backup objects expose `type`; for chained backups it's "INCREMENTAL", else "FULL".
         return getattr(backup, 'type', 'FULL') or 'FULL'
@@ -292,6 +304,7 @@ def test_incremental_chain_cadence(self):
         FULL, INCREMENTAL, INCREMENTAL, FULL, INCREMENTAL, ...
         """
         self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+        original_full_every = self._get_full_every()
         self._set_full_every(3)
         try:
             ssh_client_vm = self.vm.get_ssh_client(reconnect=True)
@@ -318,7 +331,7 @@ def test_incremental_chain_cadence(self):
             for b in reversed(created):
                 Backup.delete(self.apiclient, b.id)
         finally:
-            self._set_full_every(10)
+            self._set_full_every(original_full_every)
             self.backup_offering.removeOffering(self.apiclient, self.vm.id)
 
     @attr(tags=["advanced", "backup"], required_hardware="true")
@@ -328,6 +341,7 @@ def test_restore_from_incremental(self):
         latest incremental and verify all three markers are present (chain flatten).
         """
         self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+        original_full_every = self._get_full_every()
         self._set_full_every(5)
         try:
             ssh_client_vm = self.vm.get_ssh_client(reconnect=True)
@@ -365,7 +379,7 @@ def test_restore_from_incremental(self):
             for b in reversed(backups):
                 Backup.delete(self.apiclient, b.id)
         finally:
-            self._set_full_every(10)
+            self._set_full_every(original_full_every)
             self.backup_offering.removeOffering(self.apiclient, self.vm.id)
 
     @attr(tags=["advanced", "backup"], required_hardware="true")
@@ -376,6 +390,7 @@ def test_delete_middle_incremental_repairs_chain(self):
         should still produce a working VM with all expected blocks.
         """
         self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+        original_full_every = self._get_full_every()
         self._set_full_every(5)
         try:
             ssh_client_vm = self.vm.get_ssh_client(reconnect=True)
@@ -416,7 +431,7 @@ def test_delete_middle_incremental_repairs_chain(self):
             Backup.delete(self.apiclient, inc2.id)
             Backup.delete(self.apiclient, full.id)
         finally:
-            self._set_full_every(10)
+            self._set_full_every(original_full_every)
             self.backup_offering.removeOffering(self.apiclient, self.vm.id)
 
     @attr(tags=["advanced", "backup"], required_hardware="true")
@@ -426,6 +441,7 @@ def test_refuse_delete_full_with_children(self):
         With forced=true it must succeed and remove the entire chain.
         """
         self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+        original_full_every = self._get_full_every()
         self._set_full_every(5)
         try:
             Backup.create(self.apiclient, self.vm.id, "rdc_full")
@@ -449,7 +465,7 @@ def test_refuse_delete_full_with_children(self):
             remaining = Backup.list(self.apiclient, self.vm.id)
             self.assertIsNone(remaining, "Forced delete of FULL should remove the entire chain")
         finally:
-            self._set_full_every(10)
+            self._set_full_every(original_full_every)
             self.backup_offering.removeOffering(self.apiclient, self.vm.id)
 
     @attr(tags=["advanced", "backup"], required_hardware="true")
@@ -460,6 +476,7 @@ def test_stopped_vm_falls_back_to_full(self):
         new chain. The incrementalFallback flag should be reflected in backup.type=FULL.
         """
         self.backup_offering.assignOffering(self.apiclient, self.vm.id)
+        original_full_every = self._get_full_every()
         self._set_full_every(2)  # next backup after the first should be incremental
         try:
             Backup.create(self.apiclient, self.vm.id, "svf_first")
@@ -482,5 +499,5 @@ def test_stopped_vm_falls_back_to_full(self):
             for b in reversed(backups):
                 Backup.delete(self.apiclient, b.id)
         finally:
-            self._set_full_every(10)
+            self._set_full_every(original_full_every)
             self.backup_offering.removeOffering(self.apiclient, self.vm.id)