feat(backup): cascade-delete + chain repair for NAS incrementals

Adds the delete-with-chain-repair semantics agreed in the RFC review:

  scripts/vm/hypervisor/kvm/nasbackup.sh
    - New '-o rebase' operation: rebases an existing on-NAS qcow2 onto
      a new backing parent. Uses a SAFE rebase (no -u) so the target
      absorbs blocks of the about-to-be-deleted parent before the
      backing pointer is moved up to the grandparent. Writes the new
      backing reference relative to the target's directory so it
      survives mount-point changes.
    - New CLI flags --rebase-target, --rebase-new-backing (both passed
      mount-relative).

  RebaseBackupCommand + LibvirtRebaseBackupCommandWrapper
    - New agent command that wraps the script's rebase operation. The
      provider sends one of these per child that needs re-pointing.

  NASBackupProvider.deleteBackup
    - Now plans the chain repair before touching files via
      computeChainRepair():
        * No chain metadata     -> single-file delete (legacy behaviour)
        * Tail incremental      -> single delete, no rebase
        * Middle incremental    -> rebase immediate child onto our
                                   parent, then delete; shift
                                   chain_position of all later
                                   descendants by -1
        * Full with descendants -> refuse unless forced=true; with
                                   forced=true delete full + every
                                   descendant newest-first
    - Updates parent_backup_id, chain_position metadata in
      backup_details after each rebase so the model in the DB matches
      the on-disk chain.

This implements the cascade-delete behaviour requested in @abh1sar's
review point #7.

Refs: #12899

Author: jmsperu

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

@@ -0,0 +1,73 @@
+//
+// 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 com.cloud.agent.api.Command;
+import com.cloud.agent.api.LogLevel;
+
+/**
+ * Tells the KVM agent to rebase a NAS backup qcow2 onto a new backing parent. Used by the
+ * NAS backup provider during chain repair when a middle incremental is being deleted: the
+ * immediate child must absorb the soon-to-be-deleted parent's blocks and then re-link to
+ * the grandparent. Both target and new-backing paths are NAS-mount-relative.
+ */
+public class RebaseBackupCommand extends Command {
+    private String targetPath;       // mount-relative path of the qcow2 to repoint
+    private String newBackingPath;   // mount-relative path of the new backing parent
+    private String backupRepoType;
+    private String backupRepoAddress;
+    @LogLevel(LogLevel.Log4jLevel.Off)
+    private String mountOptions;
+
+    public RebaseBackupCommand(String targetPath, String newBackingPath,
+                               String backupRepoType, String backupRepoAddress, String mountOptions) {
+        super();
+        this.targetPath = targetPath;
+        this.newBackingPath = newBackingPath;
+        this.backupRepoType = backupRepoType;
+        this.backupRepoAddress = backupRepoAddress;
+        this.mountOptions = mountOptions;
+    }
+
+    public String getTargetPath() {
+        return targetPath;
+    }
+
+    public String getNewBackingPath() {
+        return newBackingPath;
+    }
+
+    public String getBackupRepoType() {
+        return backupRepoType;
+    }
+
+    public String getBackupRepoAddress() {
+        return backupRepoAddress;
+    }
+
+    public String getMountOptions() {
+        return mountOptions == null ? "" : mountOptions;
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return true;
+    }
+}

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

@@ -693,24 +693,244 @@ public boolean deleteBackup(Backup backup, boolean forced) {
             throw new CloudRuntimeException(String.format("Unable to find a running KVM host in zone %d to delete backup %s", backup.getZoneId(), backup.getUuid()));
         }
 
-        DeleteBackupCommand command = new DeleteBackupCommand(backup.getExternalId(), backupRepository.getType(),
-                backupRepository.getAddress(), backupRepository.getMountOptions());
+        // Repair the chain (if any) before removing the backup file. For chained backups,
+        // children that point at this backup must be re-pointed at this backup's parent
+        // (with their blocks merged via qemu-img rebase). For a full at the head of a chain
+        // with surviving children, refuse unless forced — `forced=true` then deletes the
+        // full plus every descendant.
+        ChainRepairPlan plan = computeChainRepair(backup, forced);
+        if (!plan.proceed) {
+            throw new CloudRuntimeException(plan.reason);
+        }
 
-        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 delete backup timed out, please try again");
+        // Issue rebase commands for each child that needs re-pointing (ordered so each rebase
+        // operates on a chain that still resolves: children first if there are nested ones).
+        for (RebaseStep step : plan.rebaseSteps) {
+            RebaseBackupCommand rebase = new RebaseBackupCommand(step.targetMountRelativePath,
+                    step.newBackingMountRelativePath, backupRepository.getType(),
+                    backupRepository.getAddress(), backupRepository.getMountOptions());
+            BackupAnswer rebaseAnswer;
+            try {
+                rebaseAnswer = (BackupAnswer) agentManager.send(host.getId(), rebase);
+            } catch (AgentUnavailableException e) {
+                throw new CloudRuntimeException("Unable to contact backend control plane to repair backup chain");
+            } catch (OperationTimedoutException e) {
+                throw new CloudRuntimeException("Backup chain repair (rebase) timed out, please try again");
+            }
+            if (rebaseAnswer == null || !rebaseAnswer.getResult()) {
+                throw new CloudRuntimeException(String.format(
+                        "Backup chain repair failed: rebase of %s onto %s returned %s",
+                        step.targetMountRelativePath, step.newBackingMountRelativePath,
+                        rebaseAnswer == null ? "no answer" : rebaseAnswer.getDetails()));
+            }
+            // Update the rebased child's parent reference + position in backup_details.
+            BackupDetailVO parentDetail = backupDetailsDao.findDetail(step.childBackupId, NASBackupChainKeys.PARENT_BACKUP_ID);
+            if (parentDetail != null) {
+                parentDetail.setValue(step.newParentUuid == null ? "" : step.newParentUuid);
+                backupDetailsDao.update(parentDetail.getId(), parentDetail);
+            } else if (step.newParentUuid != null) {
+                backupDetailsDao.persist(new BackupDetailVO(step.childBackupId,
+                        NASBackupChainKeys.PARENT_BACKUP_ID, step.newParentUuid, true));
+            }
+            BackupDetailVO posDetail = backupDetailsDao.findDetail(step.childBackupId, NASBackupChainKeys.CHAIN_POSITION);
+            if (posDetail != null) {
+                posDetail.setValue(String.valueOf(step.newChainPosition));
+                backupDetailsDao.update(posDetail.getId(), posDetail);
+            }
         }
 
-        if (answer != null && answer.getResult()) {
-            return backupDao.remove(backup.getId());
+        // Now delete this backup's files. For a forced delete of a full with descendants we
+        // also delete all descendants' files (newest first so each rm targets a leaf).
+        for (Backup victim : plan.toDelete) {
+            DeleteBackupCommand command = new DeleteBackupCommand(victim.getExternalId(), backupRepository.getType(),
+                    backupRepository.getAddress(), backupRepository.getMountOptions());
+            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 delete backup timed out, please try again");
+            }
+            if (answer == null || !answer.getResult()) {
+                logger.warn("Failed to delete backup file for {} ({}); leaving DB row intact", victim.getUuid(), victim.getExternalId());
+                return false;
+            }
+            backupDao.remove(victim.getId());
         }
 
-        logger.debug("There was an error removing the backup with id {}", backup.getId());
-        return false;
+        // Shift chain_position down by 1 for any survivors deeper in the chain than the
+        // backup we just removed (their direct parent reference is unchanged, but their
+        // numeric position needs to stay consistent so future full-every cadence math works).
+        if (plan.shiftPositionsBelow != null) {
+            for (Backup b : backupDao.listByVmId(null, backup.getVmId())) {
+                if (!plan.shiftPositionsBelow.chainId.equals(readDetail(b, NASBackupChainKeys.CHAIN_ID))) {
+                    continue;
+                }
+                int pos = chainPosition(b);
+                if (pos > plan.shiftPositionsBelow.afterPosition && pos != Integer.MAX_VALUE) {
+                    BackupDetailVO posDetail = backupDetailsDao.findDetail(b.getId(), NASBackupChainKeys.CHAIN_POSITION);
+                    if (posDetail != null) {
+                        posDetail.setValue(String.valueOf(pos - 1));
+                        backupDetailsDao.update(posDetail.getId(), posDetail);
+                    }
+                }
+            }
+        }
+
+        return true;
+    }
+
+    private static final class PositionShift {
+        final String chainId;
+        final int afterPosition; // shift positions strictly greater than this by -1
+        PositionShift(String chainId, int afterPosition) {
+            this.chainId = chainId;
+            this.afterPosition = afterPosition;
+        }
+    }
+
+    /**
+     * Result of {@link #computeChainRepair}: whether to proceed, what to rebase, what to delete.
+     */
+    private static final class ChainRepairPlan {
+        final boolean proceed;
+        final String reason;
+        final List<RebaseStep> rebaseSteps;
+        final List<Backup> toDelete;
+        final PositionShift shiftPositionsBelow;
+
+        private ChainRepairPlan(boolean proceed, String reason, List<RebaseStep> rebaseSteps, List<Backup> toDelete,
+                                PositionShift shiftPositionsBelow) {
+            this.proceed = proceed;
+            this.reason = reason;
+            this.rebaseSteps = rebaseSteps;
+            this.toDelete = toDelete;
+            this.shiftPositionsBelow = shiftPositionsBelow;
+        }
+
+        static ChainRepairPlan refuse(String reason) {
+            return new ChainRepairPlan(false, reason, Collections.emptyList(), Collections.emptyList(), null);
+        }
+
+        static ChainRepairPlan proceed(List<RebaseStep> rebaseSteps, List<Backup> toDelete) {
+            return new ChainRepairPlan(true, null, rebaseSteps, toDelete, null);
+        }
+
+        static ChainRepairPlan proceed(List<RebaseStep> rebaseSteps, List<Backup> toDelete, PositionShift shift) {
+            return new ChainRepairPlan(true, null, rebaseSteps, toDelete, shift);
+        }
+    }
+
+    private static final class RebaseStep {
+        final long childBackupId;
+        final String targetMountRelativePath;
+        final String newBackingMountRelativePath;
+        final String newParentUuid;        // null when re-pointed onto an existing full's UUID is desired but unavailable
+        final int newChainPosition;
+
+        RebaseStep(long childBackupId, String targetMountRelativePath, String newBackingMountRelativePath,
+                   String newParentUuid, int newChainPosition) {
+            this.childBackupId = childBackupId;
+            this.targetMountRelativePath = targetMountRelativePath;
+            this.newBackingMountRelativePath = newBackingMountRelativePath;
+            this.newParentUuid = newParentUuid;
+            this.newChainPosition = newChainPosition;
+        }
+    }
+
+    /**
+     * Compute the chain-repair plan for deleting {@code backup}. Conservative semantics:
+     *   - Backups outside any tracked chain (no NAS chain metadata) are deleted as-is.
+     *   - A standalone backup with no children is deleted as-is.
+     *   - A middle incremental: rebase its immediate child onto its own parent, then delete it.
+     *     Descendants of that child are unaffected (their backing chain still resolves).
+     *   - A full with surviving descendants: refuse unless {@code forced=true}; then delete
+     *     full + every descendant (newest first).
+     */
+    private ChainRepairPlan computeChainRepair(Backup backup, boolean forced) {
+        String chainId = readDetail(backup, NASBackupChainKeys.CHAIN_ID);
+        if (chainId == null) {
+            // Pre-incremental backups (or callers that never wrote chain metadata) — single delete.
+            return ChainRepairPlan.proceed(Collections.emptyList(), Collections.singletonList(backup));
+        }
+
+        // Gather every backup in the same chain for this VM.
+        List<Backup> chain = new ArrayList<>();
+        for (Backup b : backupDao.listByVmId(null, backup.getVmId())) {
+            if (chainId.equals(readDetail(b, NASBackupChainKeys.CHAIN_ID))) {
+                chain.add(b);
+            }
+        }
+        chain.sort(Comparator.comparingInt(b -> chainPosition(b)));
+
+        int targetPos = chainPosition(backup);
+        boolean isFull = targetPos == 0;
+        List<Backup> descendants = chain.stream()
+                .filter(b -> chainPosition(b) > targetPos)
+                .collect(Collectors.toList());
+
+        if (isFull) {
+            if (descendants.isEmpty()) {
+                return ChainRepairPlan.proceed(Collections.emptyList(), Collections.singletonList(backup));
+            }
+            if (!forced) {
+                return ChainRepairPlan.refuse(String.format(
+                        "Backup %s is the full anchor of a chain with %d incremental(s). Delete the incrementals first, " +
+                                "or pass forced=true to remove the entire chain.",
+                        backup.getUuid(), descendants.size()));
+            }
+            // Forced delete: remove descendants newest first, then the full.
+            List<Backup> victims = new ArrayList<>(descendants);
+            victims.sort(Comparator.comparingInt((Backup b) -> chainPosition(b)).reversed());
+            victims.add(backup);
+            return ChainRepairPlan.proceed(Collections.emptyList(), victims);
+        }
+
+        // Middle (or tail) incremental.
+        if (descendants.isEmpty()) {
+            // Tail: nothing to rebase, just delete.
+            return ChainRepairPlan.proceed(Collections.emptyList(), Collections.singletonList(backup));
+        }
+
+        // Middle: only the immediate child needs to absorb our blocks and rebase onto our parent.
+        Backup immediateChild = descendants.stream()
+                .min(Comparator.comparingInt(b -> chainPosition(b)))
+                .orElseThrow(() -> new CloudRuntimeException("Internal error: no immediate child found for chain repair"));
+        Backup ourParent = chain.stream()
+                .filter(b -> chainPosition(b) == targetPos - 1)
+                .findFirst()
+                .orElseThrow(() -> new CloudRuntimeException(String.format(
+                        "Cannot delete %s: its parent (chain_position=%d) is missing from the chain",
+                        backup.getUuid(), targetPos - 1)));
+
+        VolumeVO rootVolume = volumeDao.getInstanceRootVolume(backup.getVmId());
+        String volUuid = rootVolume == null ? "root" : rootVolume.getUuid();
+        String childPath = immediateChild.getExternalId() + "/root." + volUuid + ".qcow2";
+        String parentPath = ourParent.getExternalId() + "/root." + volUuid + ".qcow2";
+
+        RebaseStep step = new RebaseStep(immediateChild.getId(), childPath, parentPath,
+                ourParent.getUuid(), chainPosition(immediateChild) - 1);
+
+        // After we delete the middle backup, every descendant's numeric chain_position
+        // becomes stale (off by one). Their backing-file pointers don't need re-writing
+        // (only the immediate child changed parents) but their position metadata does.
+        return ChainRepairPlan.proceed(
+                Collections.singletonList(step),
+                Collections.singletonList(backup),
+                new PositionShift(chainId, targetPos));
+    }
+
+    private int chainPosition(Backup b) {
+        String s = readDetail(b, NASBackupChainKeys.CHAIN_POSITION);
+        if (s == null) {
+            return Integer.MAX_VALUE; // no metadata => sort to end
+        }
+        try {
+            return Integer.parseInt(s);
+        } catch (NumberFormatException e) {
+            return Integer.MAX_VALUE;
+        }
     }
 
     public void syncBackupMetrics(Long zoneId) {

plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtRebaseBackupCommandWrapper.java

@@ -0,0 +1,59 @@
+//
+// 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 com.cloud.hypervisor.kvm.resource.wrapper;
+
+import com.cloud.agent.api.Answer;
+import com.cloud.hypervisor.kvm.resource.LibvirtComputingResource;
+import com.cloud.resource.CommandWrapper;
+import com.cloud.resource.ResourceWrapper;
+import com.cloud.utils.Pair;
+import com.cloud.utils.script.Script;
+import org.apache.cloudstack.backup.BackupAnswer;
+import org.apache.cloudstack.backup.RebaseBackupCommand;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@ResourceWrapper(handles = RebaseBackupCommand.class)
+public class LibvirtRebaseBackupCommandWrapper extends CommandWrapper<RebaseBackupCommand, Answer, LibvirtComputingResource> {
+    @Override
+    public Answer execute(RebaseBackupCommand command, LibvirtComputingResource libvirtComputingResource) {
+        List<String[]> commands = new ArrayList<>();
+        commands.add(new String[]{
+                libvirtComputingResource.getNasBackupPath(),
+                "-o", "rebase",
+                "-t", command.getBackupRepoType(),
+                "-s", command.getBackupRepoAddress(),
+                "-m", command.getMountOptions(),
+                "--rebase-target", command.getTargetPath(),
+                "--rebase-new-backing", command.getNewBackingPath()
+        });
+
+        Pair<Integer, String> result = Script.executePipedCommands(commands, libvirtComputingResource.getCmdsTimeout());
+        logger.debug("Backup rebase result: {} , exit code: {}", result.second(), result.first());
+
+        if (result.first() != 0) {
+            logger.warn("Failed to rebase backup file {} onto {}: {}",
+                    command.getTargetPath(), command.getNewBackingPath(), result.second());
+            return new BackupAnswer(command, false, result.second());
+        }
+        return new BackupAnswer(command, true, null);
+    }
+}