Refactoring retention of backup schedules (#11223)

* refactor backup schedule retention workflows

Co-authored-by: Fabricio Duarte <fabricio.duarte.jr@gmail.com>

api/src/main/java/org/apache/cloudstack/api/ApiConstants.java

@@ -1307,6 +1307,10 @@ public class ApiConstants {
             "however, the following formats are also accepted: \"yyyy-MM-dd HH:mm:ss\" (e.g.: \"2023-01-01 12:00:00\") and \"yyyy-MM-dd\" (e.g.: \"2023-01-01\" - if the time is not " +
             "added, it will be interpreted as \"23:59:59\"). If the recommended format is not used, the date will be considered in the server timezone.";
 
+    public static final String PARAMETER_DESCRIPTION_MAX_BACKUPS = "The maximum number of backups to keep for a VM. " +
+            "If \"0\", no retention policy will be applied and, thus, no backups from the schedule will be automatically deleted. " +
+            "This parameter is only supported for the Dummy, NAS and EMC Networker backup provider.";
+
     public static final String VMWARE_DC = "vmwaredc";
 
     public static final String CSS = "css";

api/src/main/java/org/apache/cloudstack/api/command/user/backup/CreateBackupCmd.java

@@ -19,7 +19,6 @@ package org.apache.cloudstack.api.command.user.backup;
 
 import javax.inject.Inject;
 
-import com.cloud.storage.Snapshot;
 import org.apache.cloudstack.acl.RoleType;
 import org.apache.cloudstack.api.APICommand;
 import org.apache.cloudstack.api.ApiCommandResourceType;
@@ -28,7 +27,6 @@ import org.apache.cloudstack.api.ApiErrorCode;
 import org.apache.cloudstack.api.BaseAsyncCreateCmd;
 import org.apache.cloudstack.api.Parameter;
 import org.apache.cloudstack.api.ServerApiException;
-import org.apache.cloudstack.api.response.BackupScheduleResponse;
 import org.apache.cloudstack.api.response.SuccessResponse;
 import org.apache.cloudstack.api.response.UserVmResponse;
 import org.apache.cloudstack.backup.BackupManager;
@@ -62,13 +60,6 @@ public class CreateBackupCmd extends BaseAsyncCreateCmd {
             description = "ID of the VM")
     private Long vmId;
 
-    @Parameter(name = ApiConstants.SCHEDULE_ID,
-            type = CommandType.LONG,
-            entityType = BackupScheduleResponse.class,
-            description = "backup schedule ID of the VM, if this is null, it indicates that it is a manual backup.",
-            since = "4.21.0")
-    private Long scheduleId;
-
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -77,14 +68,6 @@ public class CreateBackupCmd extends BaseAsyncCreateCmd {
         return vmId;
     }
 
-    public Long getScheduleId() {
-        if (scheduleId != null) {
-            return scheduleId;
-        } else {
-            return Snapshot.MANUAL_POLICY_ID;
-        }
-    }
-
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
@@ -92,7 +75,7 @@ public class CreateBackupCmd extends BaseAsyncCreateCmd {
     @Override
     public void execute() throws ResourceUnavailableException, InsufficientCapacityException, ServerApiException, ConcurrentOperationException, ResourceAllocationException, NetworkRuleConflictException {
         try {
-            boolean result = backupManager.createBackup(getVmId(), getScheduleId());
+            boolean result = backupManager.createBackup(getVmId(), getJob());
             if (result) {
                 SuccessResponse response = new SuccessResponse(getCommandName());
                 response.setResponseName(getCommandName());

api/src/main/java/org/apache/cloudstack/api/command/user/backup/CreateBackupScheduleCmd.java

@@ -75,10 +75,8 @@ public class CreateBackupScheduleCmd extends BaseCmd {
             description = "Specifies a timezone for this command. For more information on the timezone parameter, see TimeZone Format.")
     private String timezone;
 
-    @Parameter(name = ApiConstants.MAX_BACKUPS,
-            type = CommandType.INTEGER,
-            description = "maximum number of backups to retain",
-            since = "4.21.0")
+    @Parameter(name = ApiConstants.MAX_BACKUPS, type = CommandType.INTEGER,
+            since = "4.21.0", description = ApiConstants.PARAMETER_DESCRIPTION_MAX_BACKUPS)
     private Integer maxBackups;
 
     /////////////////////////////////////////////////////

api/src/main/java/org/apache/cloudstack/backup/Backup.java

@@ -33,28 +33,6 @@ public interface Backup extends ControlledEntity, InternalIdentity, Identity {
         Allocated, Queued, BackingUp, BackedUp, Error, Failed, Restoring, Removed, Expunged
     }
 
-    public enum Type {
-        MANUAL, HOURLY, DAILY, WEEKLY, MONTHLY;
-        private int max = 8;
-
-        public void setMax(int max) {
-            this.max = max;
-        }
-
-        public int getMax() {
-            return max;
-        }
-
-        @Override
-        public String toString() {
-            return this.name();
-        }
-
-        public boolean equals(String snapshotType) {
-            return this.toString().equalsIgnoreCase(snapshotType);
-        }
-    }
-
     class Metric {
         private Long backupSize = 0L;
         private Long dataSize = 0L;
@@ -166,4 +144,5 @@ public interface Backup extends ControlledEntity, InternalIdentity, Identity {
     Long getProtectedSize();
     List<VolumeInfo> getBackedUpVolumes();
     long getZoneId();
+    Long getBackupScheduleId();
 }

api/src/main/java/org/apache/cloudstack/backup/BackupManager.java

@@ -58,38 +58,6 @@ public interface BackupManager extends BackupService, Configurable, PluggableSer
             "false",
             "Enable volume attach/detach operations for VMs that are assigned to Backup Offerings.", true);
 
-    ConfigKey<Integer> BackupHourlyMax = new ConfigKey<Integer>("Advanced", Integer.class,
-            "backup.max.hourly",
-            "8",
-            "Maximum recurring hourly backups to be retained for an instance. If the limit is reached, early backups from the start of the hour are deleted so that newer ones can be saved. This limit does not apply to manual backups. If set to 0, recurring hourly backups can not be scheduled.",
-            false,
-            ConfigKey.Scope.Global,
-            null);
-
-    ConfigKey<Integer> BackupDailyMax = new ConfigKey<Integer>("Advanced", Integer.class,
-            "backup.max.daily",
-            "8",
-            "Maximum recurring daily backups to be retained for an instance. If the limit is reached, backups from the start of the day are deleted so that newer ones can be saved. This limit does not apply to manual backups. If set to 0, recurring daily backups can not be scheduled.",
-            false,
-            ConfigKey.Scope.Global,
-            null);
-
-    ConfigKey<Integer> BackupWeeklyMax = new ConfigKey<Integer>("Advanced", Integer.class,
-            "backup.max.weekly",
-            "8",
-            "Maximum recurring weekly backups to be retained for an instance. If the limit is reached, backups from the beginning of the week are deleted so that newer ones can be saved. This limit does not apply to manual backups. If set to 0, recurring weekly backups can not be scheduled.",
-            false,
-            ConfigKey.Scope.Global,
-            null);
-
-    ConfigKey<Integer> BackupMonthlyMax = new ConfigKey<Integer>("Advanced", Integer.class,
-            "backup.max.monthly",
-            "8",
-            "Maximum recurring monthly backups to be retained for an instance. If the limit is reached, backups from the beginning of the month are deleted so that newer ones can be saved. This limit does not apply to manual backups. If set to 0, recurring monthly backups can not be scheduled.",
-            false,
-             ConfigKey.Scope.Global,
-            null);
-
     ConfigKey<Long> DefaultMaxAccountBackups = new ConfigKey<Long>("Account Defaults", Long.class,
             "max.account.backups",
             "20",
@@ -201,10 +169,10 @@ public interface BackupManager extends BackupService, Configurable, PluggableSer
     /**
      * Creates backup of a VM
      * @param vmId Virtual Machine ID
-     * @param scheduleId Virtual Machine Backup Schedule ID
+     * @param job The async job associated with the backup retention
      * @return returns operation success
      */
-    boolean createBackup(final Long vmId, final Long scheduleId) throws ResourceAllocationException;
+    boolean createBackup(final Long vmId, Object job) throws ResourceAllocationException;
 
     /**
      * List existing backups for a VM

api/src/main/java/org/apache/cloudstack/backup/BackupSchedule.java

@@ -30,6 +30,6 @@ public interface BackupSchedule extends InternalIdentity {
     String getTimezone();
     Date getScheduledTimestamp();
     Long getAsyncJobId();
-    Integer getMaxBackups();
+    int getMaxBackups();
     String getUuid();
 }

engine/schema/src/main/java/org/apache/cloudstack/backup/BackupScheduleVO.java

@@ -63,12 +63,12 @@ public class BackupScheduleVO implements BackupSchedule {
     Long asyncJobId;
 
     @Column(name = "max_backups")
-    Integer maxBackups = 0;
+    private int maxBackups = 0;
 
     public BackupScheduleVO() {
     }
 
-    public BackupScheduleVO(Long vmId, DateUtil.IntervalType scheduleType, String schedule, String timezone, Date scheduledTimestamp, Integer maxBackups) {
+    public BackupScheduleVO(Long vmId, DateUtil.IntervalType scheduleType, String schedule, String timezone, Date scheduledTimestamp, int maxBackups) {
         this.vmId = vmId;
         this.scheduleType = (short) scheduleType.ordinal();
         this.schedule = schedule;
@@ -142,11 +142,11 @@ public class BackupScheduleVO implements BackupSchedule {
         this.asyncJobId = asyncJobId;
     }
 
-    public Integer getMaxBackups() {
+    public int getMaxBackups() {
         return maxBackups;
     }
 
-    public void setMaxBackups(Integer maxBackups) {
+    public void setMaxBackups(int maxBackups) {
         this.maxBackups = maxBackups;
     }
 }

engine/schema/src/main/java/org/apache/cloudstack/backup/BackupVO.java

@@ -88,12 +88,12 @@ public class BackupVO implements Backup {
     @Column(name = "zone_id")
     private long zoneId;
 
-    @Column(name = "backup_interval_type")
-    private short backupIntervalType;
-
     @Column(name = "backed_volumes", length = 65535)
     protected String backedUpVolumes;
 
+    @Column(name = "backup_schedule_id")
+    private Long backupScheduleId;
+
     public BackupVO() {
         this.uuid = UUID.randomUUID().toString();
     }
@@ -211,14 +211,6 @@ public class BackupVO implements Backup {
         this.zoneId = zoneId;
     }
 
-    public short getBackupIntervalType() {
-        return backupIntervalType;
-    }
-
-    public void setBackupIntervalType(short backupIntervalType) {
-        this.backupIntervalType = backupIntervalType;
-    }
-
     @Override
     public Class<?> getEntityType() {
         return Backup.class;
@@ -247,4 +239,13 @@ public class BackupVO implements Backup {
     public void setRemoved(Date removed) {
         this.removed = removed;
     }
+
+    @Override
+    public Long getBackupScheduleId() {
+        return backupScheduleId;
+    }
+
+    public void setBackupScheduleId(Long backupScheduleId) {
+        this.backupScheduleId = backupScheduleId;
+    }
 }

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

@@ -36,9 +36,8 @@ public interface BackupDao extends GenericDao<BackupVO, Long> {
     BackupVO getBackupVO(Backup backup);
     List<Backup> listByOfferingId(Long backupOfferingId);
 
-    List<BackupVO> listBackupsByVMandIntervalType(Long vmId, Backup.Type backupType);
-
     BackupResponse newBackupResponse(Backup backup);
     public Long countBackupsForAccount(long accountId);
     public Long calculateBackupStorageForAccount(long accountId);
+    List<BackupVO> listBySchedule(Long backupScheduleId);
 }

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

@@ -24,6 +24,7 @@ import java.util.Objects;
 import javax.annotation.PostConstruct;
 import javax.inject.Inject;
 
+import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.GenericSearchBuilder;
 import org.apache.cloudstack.api.response.BackupResponse;
 import org.apache.cloudstack.backup.Backup;
@@ -63,7 +64,7 @@ public class BackupDaoImpl extends GenericDaoBase<BackupVO, Long> implements Bac
     private SearchBuilder<BackupVO> backupSearch;
     private GenericSearchBuilder<BackupVO, Long> CountBackupsByAccount;
     private GenericSearchBuilder<BackupVO, SumCount> CalculateBackupStorageByAccount;
-    private SearchBuilder<BackupVO> ListBackupsByVMandIntervalType;
+    private SearchBuilder<BackupVO> listBackupsBySchedule;
 
     public BackupDaoImpl() {
     }
@@ -91,12 +92,11 @@ public class BackupDaoImpl extends GenericDaoBase<BackupVO, Long> implements Bac
         CalculateBackupStorageByAccount.and("removed", CalculateBackupStorageByAccount.entity().getRemoved(), SearchCriteria.Op.NULL);
         CalculateBackupStorageByAccount.done();
 
-        ListBackupsByVMandIntervalType = createSearchBuilder();
-        ListBackupsByVMandIntervalType.and("vmId", ListBackupsByVMandIntervalType.entity().getVmId(), SearchCriteria.Op.EQ);
-        ListBackupsByVMandIntervalType.and("intervalType", ListBackupsByVMandIntervalType.entity().getBackupIntervalType(), SearchCriteria.Op.EQ);
-        ListBackupsByVMandIntervalType.and("status", ListBackupsByVMandIntervalType.entity().getStatus(), SearchCriteria.Op.EQ);
-        ListBackupsByVMandIntervalType.and("removed", ListBackupsByVMandIntervalType.entity().getRemoved(), SearchCriteria.Op.NULL);
-        ListBackupsByVMandIntervalType.done();
+        listBackupsBySchedule = createSearchBuilder();
+        listBackupsBySchedule.and("backup_schedule_id", listBackupsBySchedule.entity().getBackupScheduleId(), SearchCriteria.Op.EQ);
+        listBackupsBySchedule.and("status", listBackupsBySchedule.entity().getStatus(), SearchCriteria.Op.EQ);
+        listBackupsBySchedule.and("removed", listBackupsBySchedule.entity().getRemoved(), SearchCriteria.Op.NULL);
+        listBackupsBySchedule.done();
     }
 
     @Override
@@ -184,12 +184,11 @@ public class BackupDaoImpl extends GenericDaoBase<BackupVO, Long> implements Bac
     }
 
     @Override
-    public List<BackupVO> listBackupsByVMandIntervalType(Long vmId, Backup.Type backupType) {
-        SearchCriteria<BackupVO> sc = ListBackupsByVMandIntervalType.create();
-        sc.setParameters("vmId", vmId);
-        sc.setParameters("type", backupType.ordinal());
+    public List<BackupVO> listBySchedule(Long backupScheduleId) {
+        SearchCriteria<BackupVO> sc = listBackupsBySchedule.create();
+        sc.setParameters("backup_schedule_id", backupScheduleId);
         sc.setParameters("status", Backup.Status.BackedUp);
-        return listBy(sc, null);
+        return listBy(sc, new Filter(BackupVO.class, "date", true));
     }
 
     @Override

engine/schema/src/main/resources/META-INF/db/schema-42010to42100.sql

@@ -19,9 +19,8 @@
 -- Schema upgrade from 4.20.1.0 to 4.21.0.0
 --;
 
--- Add columns max_backup and backup_interval_type to backup table
-ALTER TABLE `cloud`.`backup_schedule` ADD COLUMN `max_backups` int(8) default NULL COMMENT 'maximum number of backups to maintain';
-ALTER TABLE `cloud`.`backups` ADD COLUMN `backup_interval_type` int(5) COMMENT 'type of backup, e.g. manual, recurring - hourly, daily, weekly or monthly';
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.backup_schedule', 'max_backups', 'INT(8) UNSIGNED NOT NULL DEFAULT 0 COMMENT ''Maximum number of backups to be retained''');
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.backups', 'backup_schedule_id', 'BIGINT(20) UNSIGNED');
 
 -- Update default value for the config 'vm.network.nic.max.secondary.ipaddresses' (and value to default value if value is null)
 UPDATE `cloud`.`configuration` SET default_value = '10' WHERE name = 'vm.network.nic.max.secondary.ipaddresses';

server/src/main/java/com/cloud/api/dispatch/ParamGenericValidationWorker.java

@@ -69,6 +69,7 @@ public class ParamGenericValidationWorker implements DispatchWorker {
         defaultParamNames.add(ApiConstants.ID);
         defaultParamNames.add(ApiConstants.SIGNATURE_VERSION);
         defaultParamNames.add(ApiConstants.EXPIRES);
+        defaultParamNames.add(ApiConstants.SCHEDULE_ID);
         defaultParamNames.add("_");
     }
 

server/src/main/java/org/apache/cloudstack/backup/BackupManagerImpl.java

@@ -33,6 +33,8 @@ import java.util.stream.Collectors;
 import javax.inject.Inject;
 import javax.naming.ConfigurationException;
 
+import com.cloud.serializer.GsonHelper;
+import com.google.gson.reflect.TypeToken;
 import org.apache.cloudstack.api.ApiCommandResourceType;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.InternalIdentity;
@@ -71,6 +73,7 @@ import org.apache.cloudstack.poll.BackgroundPollTask;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
+import org.apache.commons.lang.math.NumberUtils;
 import org.apache.commons.lang3.BooleanUtils;
 import org.apache.commons.lang3.ObjectUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -97,7 +100,6 @@ import com.cloud.hypervisor.HypervisorGuru;
 import com.cloud.hypervisor.HypervisorGuruManager;
 import com.cloud.projects.Project;
 import com.cloud.storage.ScopeType;
-import com.cloud.storage.Snapshot;
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeApiService;
 import com.cloud.storage.VolumeVO;
@@ -430,7 +432,6 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         final DateUtil.IntervalType intervalType = cmd.getIntervalType();
         final String scheduleString = cmd.getSchedule();
         final TimeZone timeZone = TimeZone.getTimeZone(cmd.getTimezone());
-        final Integer maxBackups = cmd.getMaxBackups();
 
         if (intervalType == null) {
             throw new CloudRuntimeException("Invalid interval type provided");
@@ -443,40 +444,13 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         if (vm.getBackupOfferingId() == null) {
             throw new CloudRuntimeException("Cannot configure backup schedule for the VM without having any backup offering");
         }
-        if (maxBackups != null && maxBackups <= 0) {
-            throw new InvalidParameterValueException(String.format("maxBackups [%s] for instance %s should be greater than 0.", maxBackups, vm.getName()));
-        }
-
-        Backup.Type backupType = Backup.Type.valueOf(intervalType.name());
-        int intervalMaxBackups = backupType.getMax();
-        if (maxBackups != null && maxBackups > intervalMaxBackups) {
-            throw new InvalidParameterValueException(String.format("maxBackups [%s] for instance %s exceeds limit [%s] for interval type [%s].", maxBackups, vm.getName(),
-                    intervalMaxBackups, intervalType));
-        }
-
-        Account owner = accountManager.getAccount(vm.getAccountId());
-
-        long accountLimit = resourceLimitMgr.findCorrectResourceLimitForAccount(owner, Resource.ResourceType.backup, null);
-        long domainLimit = resourceLimitMgr.findCorrectResourceLimitForDomain(domainManager.getDomain(owner.getDomainId()), Resource.ResourceType.backup, null);
-        if (maxBackups != null && !accountManager.isRootAdmin(owner.getId()) && ((accountLimit != -1 && maxBackups > accountLimit) || (domainLimit != -1 && maxBackups > domainLimit))) {
-            String message = "domain/account";
-            if (owner.getType() == Account.Type.PROJECT) {
-                message = "domain/project";
-            }
-            throw new InvalidParameterValueException("Max number of backups shouldn't exceed the " + message + " level backup limit");
-        }
 
         final BackupOffering offering = backupOfferingDao.findById(vm.getBackupOfferingId());
         if (offering == null || !offering.isUserDrivenBackupAllowed()) {
             throw new CloudRuntimeException("The selected backup offering does not allow user-defined backup schedule");
         }
 
-        if (maxBackups == null && !"veeam".equals(offering.getProvider())) {
-            throw new CloudRuntimeException("Please specify the maximum number of buckets to retain.");
-        }
-        if (maxBackups != null && "veeam".equals(offering.getProvider())) {
-            throw new CloudRuntimeException("The maximum backups to retain cannot be configured through CloudStack for Veeam. Retention is managed directly in Veeam based on the settings specified when creating the backup job.");
-        }
+        final int maxBackups = validateAndGetDefaultBackupRetentionIfRequired(cmd.getMaxBackups(), offering, vm);
 
         final String timezoneId = timeZone.getID();
         if (!timezoneId.equals(cmd.getTimezone())) {
@@ -504,6 +478,43 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         return backupScheduleDao.findById(schedule.getId());
     }
 
+    /**
+     * Validates the provided backup retention value and returns 0 as the default value if required.
+     *
+     * @param maxBackups The number of backups to retain, can be null
+     * @param offering The backup offering
+     * @param vm The VM associated with the backup schedule
+     * @return The validated number of backups to retain. If maxBackups is null, returns 0 as the default value
+     * @throws InvalidParameterValueException if the backup offering's provider is Veeam, or maxBackups is less than 0 or greater than the account and domain backup limits
+     */
+    protected int validateAndGetDefaultBackupRetentionIfRequired(Integer maxBackups, BackupOffering offering, VirtualMachine vm) {
+        if (maxBackups == null) {
+            return 0;
+        }
+        if ("veeam".equals(offering.getProvider())) {
+            throw new InvalidParameterValueException("The maximum amount of backups to retain cannot be directly configured via Apache CloudStack for Veeam. " +
+                    "Retention is managed directly in Veeam based on the settings specified when creating the backup job.");
+        }
+        if (maxBackups < 0) {
+            throw new InvalidParameterValueException("maxbackups value for backup schedule must be a non-negative integer.");
+        }
+
+        Account owner = accountManager.getAccount(vm.getAccountId());
+        long accountLimit = resourceLimitMgr.findCorrectResourceLimitForAccount(owner, Resource.ResourceType.backup, null);
+        boolean exceededAccountLimit = accountLimit != -1 && maxBackups > accountLimit;
+
+        long domainLimit = resourceLimitMgr.findCorrectResourceLimitForDomain(domainManager.getDomain(owner.getDomainId()), Resource.ResourceType.backup, null);
+        boolean exceededDomainLimit = domainLimit != -1 && maxBackups > domainLimit;
+
+        if (!accountManager.isRootAdmin(owner.getId()) && (exceededAccountLimit || exceededDomainLimit)) {
+            throw new InvalidParameterValueException(
+                    String.format("'maxbackups' should not exceed the domain/%s backup limit.", owner.getType() == Account.Type.PROJECT ? "project" : "account")
+            );
+        }
+
+        return maxBackups;
+    }
+
     @Override
     public List<BackupSchedule> listBackupSchedule(final Long vmId) {
         final VMInstanceVO vm = findVmById(vmId);
@@ -564,30 +575,9 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         return success;
     }
 
-    private void postCreateScheduledBackup(Backup.Type backupType, Long vmId) {
-        DateUtil.IntervalType intervalType = DateUtil.IntervalType.valueOf(backupType.name());
-        final BackupScheduleVO schedule = backupScheduleDao.findByVMAndIntervalType(vmId, intervalType);
-        if (schedule == null) {
-            return;
-        }
-        Integer maxBackups = schedule.getMaxBackups();
-        if (maxBackups == null) {
-            return;
-        }
-        List<BackupVO> backups = backupDao.listBackupsByVMandIntervalType(vmId, backupType);
-        while (backups.size() > maxBackups) {
-            BackupVO oldestBackup = backups.get(0);
-            if (deleteBackup(oldestBackup.getId(), false)) {
-                ActionEventUtils.onCompletedActionEvent(User.UID_SYSTEM, oldestBackup.getAccountId(), EventVO.LEVEL_INFO, EventTypes.EVENT_VM_BACKUP_DELETE,
-                        "Successfully deleted oldest backup: " + oldestBackup.getId(), oldestBackup.getId(), ApiCommandResourceType.Backup.toString(), 0);
-            }
-            backups.remove(oldestBackup);
-        }
-    }
-
     @Override
     @ActionEvent(eventType = EventTypes.EVENT_VM_BACKUP_CREATE, eventDescription = "creating VM backup", async = true)
-    public boolean createBackup(final Long vmId, final Long scheduleId) throws ResourceAllocationException {
+    public boolean createBackup(final Long vmId, Object job) throws ResourceAllocationException {
         final VMInstanceVO vm = findVmById(vmId);
         validateForZone(vm.getDataCenterId());
         accountManager.checkAccess(CallContext.current().getCallingAccount(), null, true, vm);
@@ -605,16 +595,14 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
             throw new CloudRuntimeException("The assigned backup offering does not allow ad-hoc user backup");
         }
 
-        Backup.Type type = getBackupType(scheduleId);
+        Long backupScheduleId = getBackupScheduleId(job);
+        boolean isScheduledBackup = backupScheduleId != null;
         Account owner = accountManager.getAccount(vm.getAccountId());
         try {
             resourceLimitMgr.checkResourceLimit(owner, Resource.ResourceType.backup);
         } catch (ResourceAllocationException e) {
-            if (type != Backup.Type.MANUAL) {
-                String msg = "Backup resource limit exceeded for account id : " + owner.getId() + ". Failed to create backup";
-                logger.warn(msg);
-                alertManager.sendAlert(AlertManager.AlertType.ALERT_TYPE_UPDATE_RESOURCE_COUNT, 0L, 0L, msg, "Backup resource limit exceeded for account id : " + owner.getId()
-                        + ". Failed to create backups; please use updateResourceLimit to increase the limit");
+            if (isScheduledBackup) {
+                sendExceededBackupLimitAlert(owner.getUuid(), Resource.ResourceType.backup);
             }
             throw e;
         }
@@ -632,11 +620,8 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         try {
             resourceLimitMgr.checkResourceLimit(owner, Resource.ResourceType.backup_storage, backupSize);
         } catch (ResourceAllocationException e) {
-            if (type != Backup.Type.MANUAL) {
-                String msg = "Backup storage space resource limit exceeded for account id : " + owner.getId() + ". Failed to create backup";
-                logger.warn(msg);
-                alertManager.sendAlert(AlertManager.AlertType.ALERT_TYPE_UPDATE_RESOURCE_COUNT, 0L, 0L, msg, "Backup storage space resource limit exceeded for account id : " + owner.getId()
-                        + ". Failed to create backups; please use updateResourceLimit to increase the limit");
+            if (isScheduledBackup) {
+                sendExceededBackupLimitAlert(owner.getUuid(), Resource.ResourceType.backup_storage);
             }
             throw e;
         }
@@ -646,7 +631,6 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
                 vmId, ApiCommandResourceType.VirtualMachine.toString(),
                 true, 0);
 
-
         final BackupProvider backupProvider = getBackupProvider(offering.getProvider());
         if (backupProvider != null) {
             Pair<Boolean, Backup> result = backupProvider.takeBackup(vm);
@@ -656,19 +640,108 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
             Backup backup = result.second();
             if (backup != null) {
                 BackupVO vmBackup = backupDao.findById(result.second().getId());
-                vmBackup.setBackupIntervalType((short) type.ordinal());
+                vmBackup.setBackupScheduleId(backupScheduleId);
                 backupDao.update(vmBackup.getId(), vmBackup);
                 resourceLimitMgr.incrementResourceCount(vm.getAccountId(), Resource.ResourceType.backup);
                 resourceLimitMgr.incrementResourceCount(vm.getAccountId(), Resource.ResourceType.backup_storage, backup.getSize());
             }
-            if (type != Backup.Type.MANUAL) {
-                postCreateScheduledBackup(type, vm.getId());
+            if (isScheduledBackup) {
+                deleteOldestBackupFromScheduleIfRequired(vmId, backupScheduleId);
             }
             return true;
         }
         throw new CloudRuntimeException("Failed to create VM backup");
     }
 
+    /**
+     * Sends an alert when the backup limit has been exceeded for a given account.
+     *
+     * @param ownerUuid The UUID of the account owner that exceeded the limit
+     * @param resourceType The type of resource limit that was exceeded (either {@link Resource.ResourceType#backup} or {@link Resource.ResourceType#backup_storage})
+     *
+     */
+    protected void sendExceededBackupLimitAlert(String ownerUuid, Resource.ResourceType resourceType) {
+        String message = String.format("Failed to create backup: backup %s limit exceeded for account with ID: %s.",
+                resourceType == Resource.ResourceType.backup ? "resource" : "storage space resource" , ownerUuid);
+        logger.warn(message);
+        alertManager.sendAlert(AlertManager.AlertType.ALERT_TYPE_UPDATE_RESOURCE_COUNT, 0L, 0L,
+                message, message + " Please, use the 'updateResourceLimit' API to increase the backup limit.");
+    }
+
+    /**
+     * Gets the backup schedule ID from the async job's payload.
+     *
+     * @param job The asynchronous job associated with the creation of the backup
+     * @return The backup schedule ID. Returns null if the backup has been manually created
+     */
+    protected Long getBackupScheduleId(Object job) {
+        if (!(job instanceof AsyncJobVO)) {
+            return null;
+        }
+
+        AsyncJobVO asyncJob = (AsyncJobVO) job;
+        logger.debug("Trying to retrieve [{}] parameter from the job [ID: {}] parameters.", ApiConstants.SCHEDULE_ID, asyncJob.getId());
+        String jobParamsRaw = asyncJob.getCmdInfo();
+
+        if (!jobParamsRaw.contains(ApiConstants.SCHEDULE_ID)) {
+            logger.info("Job [ID: {}] parameters do not include the [{}] parameter. Thus, the current backup is a manual backup.", asyncJob.getId(), ApiConstants.SCHEDULE_ID);
+            return null;
+        }
+
+        TypeToken<Map<String, String>> jobParamsType = new TypeToken<>(){};
+        Map<String, String> jobParams = GsonHelper.getGson().fromJson(jobParamsRaw, jobParamsType.getType());
+        long backupScheduleId = NumberUtils.toLong(jobParams.get(ApiConstants.SCHEDULE_ID));
+        logger.info("Job [ID: {}] parameters include the [{}] parameter, whose value is equal to [{}]. Thus, the current backup is a scheduled backup.", asyncJob.getId(), ApiConstants.SCHEDULE_ID, backupScheduleId);
+        return backupScheduleId == 0L ? null : backupScheduleId;
+    }
+
+    /**
+     * Deletes the oldest backups from the schedule. If the backup schedule is not active, the schedule's retention is equal to 0,
+     * or the number of backups to be deleted is lower than one, then no backups are deleted.
+     *
+     * @param vmId The ID of the VM associated with the backups
+     * @param backupScheduleId Backup schedule ID of the backups
+     */
+    protected void deleteOldestBackupFromScheduleIfRequired(Long vmId, long backupScheduleId) {
+        BackupScheduleVO backupScheduleVO = backupScheduleDao.findById(backupScheduleId);
+        if (backupScheduleVO == null || backupScheduleVO.getMaxBackups() == 0) {
+            logger.info("The schedule does not have a retention specified and, hence, not deleting any backups from it.", vmId);
+            return;
+        }
+
+        logger.debug("Checking if it is required to delete the oldest backups from the schedule with ID [{}], to meet its retention requirement of [{}] backups.", backupScheduleId, backupScheduleVO.getMaxBackups());
+        List<BackupVO> backups = backupDao.listBySchedule(backupScheduleId);
+        int amountOfBackupsToDelete = backups.size() - backupScheduleVO.getMaxBackups();
+        if (amountOfBackupsToDelete > 0) {
+            deleteExcessBackups(backups, amountOfBackupsToDelete, backupScheduleId);
+        } else {
+            logger.debug("Not required to delete any backups from the schedule [ID: {}]: [backups size: {}] and [retention: {}].", backupScheduleId, backups.size(), backupScheduleVO.getMaxBackups());
+        }
+    }
+
+    /**
+     * Deletes a certain number of backups associated with a schedule.
+     *
+     * @param backups List of backups associated with a schedule
+     * @param amountOfBackupsToDelete Number of backups to be deleted from the list of backups
+     * @param backupScheduleId ID of the backup schedule associated with the backups
+     */
+    protected void deleteExcessBackups(List<BackupVO> backups, int amountOfBackupsToDelete, long backupScheduleId) {
+        logger.debug("Deleting the [{}] oldest backups from the schedule [ID: {}].", amountOfBackupsToDelete, backupScheduleId);
+
+        for (int i = 0; i < amountOfBackupsToDelete; i++) {
+            BackupVO backup = backups.get(i);
+            if (deleteBackup(backup.getId(), false)) {
+                String eventDescription = String.format("Successfully deleted backup for VM [ID: %s], suiting the retention specified in the backup schedule [ID: %s]", backup.getVmId(), backupScheduleId);
+                logger.info(eventDescription);
+                ActionEventUtils.onCompletedActionEvent(
+                        User.UID_SYSTEM, backup.getAccountId(), EventVO.LEVEL_INFO,
+                        EventTypes.EVENT_VM_BACKUP_DELETE, eventDescription, backup.getId(), ApiCommandResourceType.Backup.toString(), 0
+                );
+            }
+        }
+    }
+
     @Override
     public Pair<List<Backup>, Integer> listBackups(final ListBackupsCmd cmd) {
         final Long id = cmd.getId();
@@ -839,29 +912,6 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         }
     }
 
-    private Backup.Type getBackupType(Long scheduleId) {
-        if (scheduleId.equals(Snapshot.MANUAL_POLICY_ID)) {
-            return Backup.Type.MANUAL;
-        } else {
-            BackupScheduleVO scheduleVO = backupScheduleDao.findById(scheduleId);
-            DateUtil.IntervalType intvType = scheduleVO.getScheduleType();
-            return getBackupType(intvType);
-        }
-    }
-
-    private Backup.Type getBackupType(DateUtil.IntervalType intvType) {
-        if (intvType.equals(DateUtil.IntervalType.HOURLY)) {
-            return Backup.Type.HOURLY;
-        } else if (intvType.equals(DateUtil.IntervalType.DAILY)) {
-            return Backup.Type.DAILY;
-        } else if (intvType.equals(DateUtil.IntervalType.WEEKLY)) {
-            return Backup.Type.WEEKLY;
-        } else if (intvType.equals(DateUtil.IntervalType.MONTHLY)) {
-            return Backup.Type.MONTHLY;
-        }
-        return null;
-    }
-
     /**
      * Tries to update the state of given VM, given specified event
      * @param vm The VM to update its state
@@ -1107,10 +1157,6 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
     public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
         super.configure(name, params);
         backgroundPollManager.submitTask(new BackupSyncTask(this));
-        Backup.Type.HOURLY.setMax(BackupHourlyMax.value());
-        Backup.Type.DAILY.setMax(BackupDailyMax.value());
-        Backup.Type.WEEKLY.setMax(BackupWeeklyMax.value());
-        Backup.Type.MONTHLY.setMax(BackupMonthlyMax.value());
         return true;
     }
 
@@ -1191,10 +1237,6 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
                 BackupProviderPlugin,
                 BackupSyncPollingInterval,
                 BackupEnableAttachDetachVolumes,
-                BackupHourlyMax,
-                BackupDailyMax,
-                BackupWeeklyMax,
-                BackupMonthlyMax,
                 DefaultMaxAccountBackups,
                 DefaultMaxAccountBackupStorage,
                 DefaultMaxProjectBackups,
@@ -1333,7 +1375,7 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
                         true, 0);
                 final Map<String, String> params = new HashMap<String, String>();
                 params.put(ApiConstants.VIRTUAL_MACHINE_ID, "" + vmId);
-                params.put(ApiConstants.SCHEDULE_ID, "" + backupScheduleId);
+                params.put(ApiConstants.SCHEDULE_ID, String.valueOf(backupScheduleId));
                 params.put("ctxUserId", "1");
                 params.put("ctxAccountId", "" + vm.getAccountId());
                 params.put("ctxStartEventId", String.valueOf(eventId));

server/src/test/java/org/apache/cloudstack/backup/BackupManagerTest.java

@@ -44,6 +44,8 @@ import com.cloud.vm.VMInstanceVO;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineManager;
 import com.cloud.vm.dao.VMInstanceDao;
+import com.google.gson.Gson;
+import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.ServerApiException;
 import org.apache.cloudstack.api.command.admin.backup.UpdateBackupOfferingCmd;
 import org.apache.cloudstack.api.command.user.backup.CreateBackupScheduleCmd;
@@ -54,6 +56,7 @@ import org.apache.cloudstack.backup.dao.BackupScheduleDao;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.impl.ConfigDepotImpl;
+import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -74,9 +77,11 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TimeZone;
+import java.util.UUID;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -128,25 +133,39 @@ public class BackupManagerTest {
     DataCenterDao dataCenterDao;
 
     @Mock
-    AlertManager alertManager;
+    private AlertManager alertManagerMock;
+
+    @Mock
+    private Domain domainMock;
 
     @Mock
     private VMInstanceVO vmInstanceVOMock;
 
     @Mock
-    private CallContext callContextMock;
+    private CreateBackupScheduleCmd createBackupScheduleCmdMock;
+
+    @Mock
+    private BackupOfferingVO backupOfferingVOMock;
+
+    @Mock
+    private AsyncJobVO asyncJobVOMock;
+
+    @Mock
+    private BackupScheduleVO backupScheduleVOMock;
 
     @Mock
     private AccountVO accountVOMock;
 
     @Mock
-    private DeleteBackupScheduleCmd deleteBackupScheduleCmdMock;
+    private CallContext callContextMock;
 
     @Mock
-    private BackupScheduleVO backupScheduleVOMock;
+    private DeleteBackupScheduleCmd deleteBackupScheduleCmdMock;
 
     private UserVO user;
 
+    private Gson gson;
+
     private String[] hostPossibleValues = {"127.0.0.1", "hostname"};
     private String[] datastoresPossibleValues = {"e9804933-8609-4de3-bccc-6278072a496c", "datastore-name"};
     private AutoCloseable closeable;
@@ -155,6 +174,8 @@ public class BackupManagerTest {
 
     @Before
     public void setup() throws Exception {
+        gson = new Gson();
+
         closeable = MockitoAnnotations.openMocks(this);
         when(backupOfferingDao.findById(null)).thenReturn(null);
         when(backupOfferingDao.findById(123l)).thenReturn(null);
@@ -452,98 +473,84 @@ public class BackupManagerTest {
     }
 
     @Test
-    public void testConfigureBackupScheduleLimitReached() {
-        Long vmId = 1L;
-        Long zoneId = 2L;
-        Long accountId = 3L;
-        Long domainId = 4L;
+    public void configureBackupScheduleTestEnsureLimitCheckIsPerformed() {
+        long vmId = 1L;
+        long zoneId = 2L;
+        long accountId = 3L;
+        long domainId = 4L;
+        long backupOfferingId = 5L;
 
-        CreateBackupScheduleCmd cmd = Mockito.mock(CreateBackupScheduleCmd.class);
-        when(cmd.getVmId()).thenReturn(vmId);
-        when(cmd.getTimezone()).thenReturn("GMT");
-        when(cmd.getIntervalType()).thenReturn(DateUtil.IntervalType.DAILY);
-        when(cmd.getMaxBackups()).thenReturn(8);
+        when(createBackupScheduleCmdMock.getVmId()).thenReturn(vmId);
+        when(createBackupScheduleCmdMock.getTimezone()).thenReturn("GMT");
+        when(createBackupScheduleCmdMock.getIntervalType()).thenReturn(DateUtil.IntervalType.DAILY);
+        when(createBackupScheduleCmdMock.getMaxBackups()).thenReturn(8);
 
-        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
-        when(vmInstanceDao.findById(vmId)).thenReturn(vm);
-        when(vm.getDataCenterId()).thenReturn(zoneId);
-        when(vm.getAccountId()).thenReturn(accountId);
+        when(vmInstanceDao.findById(vmId)).thenReturn(vmInstanceVOMock);
+        when(vmInstanceVOMock.getDataCenterId()).thenReturn(zoneId);
+        when(vmInstanceVOMock.getAccountId()).thenReturn(accountId);
+        when(vmInstanceVOMock.getBackupOfferingId()).thenReturn(backupOfferingId);
+
+        when(backupOfferingDao.findById(backupOfferingId)).thenReturn(backupOfferingVOMock);
+        when(backupOfferingVOMock.isUserDrivenBackupAllowed()).thenReturn(true);
 
         overrideBackupFrameworkConfigValue();
 
-        Account account = Mockito.mock(Account.class);
-        when(accountManager.getAccount(accountId)).thenReturn(account);
-        when(account.getDomainId()).thenReturn(domainId);
-        Domain domain = Mockito.mock(Domain.class);
-        when(domainManager.getDomain(domainId)).thenReturn(domain);
-        when(resourceLimitMgr.findCorrectResourceLimitForAccount(account, Resource.ResourceType.backup, null)).thenReturn(10L);
-        when(resourceLimitMgr.findCorrectResourceLimitForDomain(domain, Resource.ResourceType.backup, null)).thenReturn(1L);
+        when(accountManager.getAccount(accountId)).thenReturn(accountVOMock);
+        when(accountVOMock.getDomainId()).thenReturn(domainId);
+        when(domainManager.getDomain(domainId)).thenReturn(domainMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForAccount(accountVOMock, Resource.ResourceType.backup, null)).thenReturn(10L);
+        when(resourceLimitMgr.findCorrectResourceLimitForDomain(domainMock, Resource.ResourceType.backup, null)).thenReturn(1L);
 
         InvalidParameterValueException exception = Assert.assertThrows(InvalidParameterValueException.class,
-                () -> backupManager.configureBackupSchedule(cmd));
-        Assert.assertEquals(exception.getMessage(), "Max number of backups shouldn't exceed the domain/account level backup limit");
+                () -> backupManager.configureBackupSchedule(createBackupScheduleCmdMock));
+        Assert.assertEquals("'maxbackups' should not exceed the domain/account backup limit.", exception.getMessage());
     }
 
     @Test
-    public void testCreateScheduledBackup() throws ResourceAllocationException {
+    public void createBackupTestCreateScheduledBackup() throws ResourceAllocationException {
         Long vmId = 1L;
         Long zoneId = 2L;
         Long scheduleId = 3L;
         Long backupOfferingId = 4L;
         Long accountId = 5L;
         Long backupId = 6L;
-        Long oldestBackupId = 7L;
         Long newBackupSize = 1000000000L;
-        Long oldBackupSize = 400000000L;
 
-        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
-        when(vmInstanceDao.findById(vmId)).thenReturn(vm);
-        when(vmInstanceDao.findByIdIncludingRemoved(vmId)).thenReturn(vm);
-        when(vm.getId()).thenReturn(vmId);
-        when(vm.getDataCenterId()).thenReturn(zoneId);
-        when(vm.getBackupOfferingId()).thenReturn(backupOfferingId);
-        when(vm.getAccountId()).thenReturn(accountId);
+        when(vmInstanceDao.findById(vmId)).thenReturn(vmInstanceVOMock);
+        when(vmInstanceVOMock.getDataCenterId()).thenReturn(zoneId);
+        when(vmInstanceVOMock.getBackupOfferingId()).thenReturn(backupOfferingId);
+        when(vmInstanceVOMock.getAccountId()).thenReturn(accountId);
 
         overrideBackupFrameworkConfigValue();
-        BackupOfferingVO offering = Mockito.mock(BackupOfferingVO.class);
-        when(backupOfferingDao.findById(backupOfferingId)).thenReturn(offering);
-        when(offering.isUserDrivenBackupAllowed()).thenReturn(true);
-        when(offering.getProvider()).thenReturn("test");
+        when(backupOfferingDao.findById(backupOfferingId)).thenReturn(backupOfferingVOMock);
+        when(backupOfferingVOMock.isUserDrivenBackupAllowed()).thenReturn(true);
+        when(backupOfferingVOMock.getProvider()).thenReturn("test");
 
-        Account account = Mockito.mock(Account.class);
-        when(accountManager.getAccount(accountId)).thenReturn(account);
+        Mockito.doReturn(scheduleId).when(backupManager).getBackupScheduleId(asyncJobVOMock);
+
+        when(accountManager.getAccount(accountId)).thenReturn(accountVOMock);
 
         BackupScheduleVO schedule = mock(BackupScheduleVO.class);
-        when(schedule.getScheduleType()).thenReturn(DateUtil.IntervalType.DAILY);
-        when(schedule.getMaxBackups()).thenReturn(0);
         when(backupScheduleDao.findById(scheduleId)).thenReturn(schedule);
-        when(backupScheduleDao.findByVMAndIntervalType(vmId, DateUtil.IntervalType.DAILY)).thenReturn(schedule);
+        when(schedule.getMaxBackups()).thenReturn(2);
 
         BackupProvider backupProvider = mock(BackupProvider.class);
         Backup backup = mock(Backup.class);
         when(backup.getId()).thenReturn(backupId);
         when(backup.getSize()).thenReturn(newBackupSize);
         when(backupProvider.getName()).thenReturn("test");
-        when(backupProvider.takeBackup(vm)).thenReturn(new Pair<>(true, backup));
+        when(backupProvider.takeBackup(vmInstanceVOMock)).thenReturn(new Pair<>(true, backup));
         Map<String, BackupProvider> backupProvidersMap = new HashMap<>();
         backupProvidersMap.put(backupProvider.getName().toLowerCase(), backupProvider);
         ReflectionTestUtils.setField(backupManager, "backupProvidersMap", backupProvidersMap);
 
         BackupVO backupVO = mock(BackupVO.class);
         when(backupVO.getId()).thenReturn(backupId);
-        BackupVO oldestBackupVO = mock(BackupVO.class);
-        when(oldestBackupVO.getSize()).thenReturn(oldBackupSize);
-        when(oldestBackupVO.getId()).thenReturn(oldestBackupId);
-        when(oldestBackupVO.getVmId()).thenReturn(vmId);
-        when(oldestBackupVO.getBackupOfferingId()).thenReturn(backupOfferingId);
+        BackupVO oldestBackupVO = mock(BackupVO.class);;
 
         when(backupDao.findById(backupId)).thenReturn(backupVO);
         List<BackupVO> backups = new ArrayList<>(List.of(oldestBackupVO));
-        when(backupDao.listBackupsByVMandIntervalType(vmId, Backup.Type.DAILY)).thenReturn(backups);
-        when(backupDao.findByIdIncludingRemoved(oldestBackupId)).thenReturn(oldestBackupVO);
-        when(backupOfferingDao.findByIdIncludingRemoved(backupOfferingId)).thenReturn(offering);
-        when(backupProvider.deleteBackup(oldestBackupVO, false)).thenReturn(true);
-        when(backupDao.remove(oldestBackupVO.getId())).thenReturn(true);
+        when(backupDao.listBySchedule(scheduleId)).thenReturn(backups);
 
         try (MockedStatic<ActionEventUtils> ignored = Mockito.mockStatic(ActionEventUtils.class)) {
             Mockito.when(ActionEventUtils.onActionEvent(Mockito.anyLong(), Mockito.anyLong(),
@@ -551,51 +558,39 @@ public class BackupManagerTest {
                     Mockito.anyString(), Mockito.anyString(),
                     Mockito.anyLong(), Mockito.anyString())).thenReturn(1L);
 
-            Assert.assertEquals(backupManager.createBackup(vmId, scheduleId), true);
-
+            assertTrue(backupManager.createBackup(vmId, asyncJobVOMock));
             Mockito.verify(resourceLimitMgr, times(1)).incrementResourceCount(accountId, Resource.ResourceType.backup);
             Mockito.verify(resourceLimitMgr, times(1)).incrementResourceCount(accountId, Resource.ResourceType.backup_storage, newBackupSize);
             Mockito.verify(backupDao, times(1)).update(backupVO.getId(), backupVO);
-
-            Mockito.verify(resourceLimitMgr, times(1)).decrementResourceCount(accountId, Resource.ResourceType.backup);
-            Mockito.verify(resourceLimitMgr, times(1)).decrementResourceCount(accountId, Resource.ResourceType.backup_storage, oldBackupSize);
-            Mockito.verify(backupDao, times(1)).remove(oldestBackupId);
+            Mockito.verify(backupManager, times(1)).deleteOldestBackupFromScheduleIfRequired(vmId, scheduleId);
         }
     }
 
-    @Test (expected = ResourceAllocationException.class)
-    public void testCreateBackupLimitReached() throws ResourceAllocationException {
+    @Test(expected = ResourceAllocationException.class)
+    public void createBackupTestResourceLimitReached() throws ResourceAllocationException {
         Long vmId = 1L;
         Long zoneId = 2L;
         Long scheduleId = 3L;
         Long backupOfferingId = 4L;
         Long accountId = 5L;
 
-        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
-        when(vmInstanceDao.findById(vmId)).thenReturn(vm);
-        when(vm.getDataCenterId()).thenReturn(zoneId);
-        when(vm.getBackupOfferingId()).thenReturn(backupOfferingId);
-        when(vm.getAccountId()).thenReturn(accountId);
+        when(vmInstanceDao.findById(vmId)).thenReturn(vmInstanceVOMock);
+        when(vmInstanceVOMock.getDataCenterId()).thenReturn(zoneId);
+        when(vmInstanceVOMock.getBackupOfferingId()).thenReturn(backupOfferingId);
+        when(vmInstanceVOMock.getAccountId()).thenReturn(accountId);
 
         overrideBackupFrameworkConfigValue();
         BackupOfferingVO offering = Mockito.mock(BackupOfferingVO.class);
         when(backupOfferingDao.findById(backupOfferingId)).thenReturn(offering);
         when(offering.isUserDrivenBackupAllowed()).thenReturn(true);
 
-        BackupScheduleVO schedule = mock(BackupScheduleVO.class);
-        when(schedule.getScheduleType()).thenReturn(DateUtil.IntervalType.DAILY);
-        when(backupScheduleDao.findById(scheduleId)).thenReturn(schedule);
+        Mockito.doReturn(scheduleId).when(backupManager).getBackupScheduleId(asyncJobVOMock);
 
         Account account = Mockito.mock(Account.class);
-        when(account.getId()).thenReturn(accountId);
         when(accountManager.getAccount(accountId)).thenReturn(account);
         Mockito.doThrow(new ResourceAllocationException("", Resource.ResourceType.backup_storage)).when(resourceLimitMgr).checkResourceLimit(account, Resource.ResourceType.backup_storage, 0L);
 
-        backupManager.createBackup(vmId, scheduleId);
-
-        String msg = "Backup storage space resource limit exceeded for account id : " + accountId + ". Failed to create backup";
-        Mockito.verify(alertManager, times(1)).sendAlert(AlertManager.AlertType.ALERT_TYPE_UPDATE_RESOURCE_COUNT, 0L, 0L, msg, "Backup storage space resource limit exceeded for account id : " + accountId
-                + ". Failed to create backups; please use updateResourceLimit to increase the limit");
+        backupManager.createBackup(vmId, asyncJobVOMock);
     }
 
     @Test
@@ -766,4 +761,212 @@ public class BackupManagerTest {
         boolean success = backupManager.deleteBackupSchedule(deleteBackupScheduleCmdMock);
         assertTrue(success);
     }
+
+    @Test
+    public void validateAndGetDefaultBackupRetentionIfRequiredTestReturnZeroAsDefaultValue() {
+        int retention = backupManager.validateAndGetDefaultBackupRetentionIfRequired(null, backupOfferingVOMock, null);
+        assertEquals(0, retention);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void validateAndGetDefaultBackupRetentionIfRequiredTestThrowExceptionWhenBackupOfferingProviderIsVeeam() {
+        Mockito.when(backupOfferingVOMock.getProvider()).thenReturn("veeam");
+        backupManager.validateAndGetDefaultBackupRetentionIfRequired(1, backupOfferingVOMock, vmInstanceVOMock);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void validateAndGetDefaultBackupRetentionIfRequiredTestThrowExceptionWhenMaxBackupsIsLessThanZero() {
+        backupManager.validateAndGetDefaultBackupRetentionIfRequired(-1, backupOfferingVOMock, vmInstanceVOMock);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void validateAndGetDefaultBackupRetentionIfRequiredTestThrowExceptionWhenMaxBackupsExceedsAccountLimit() {
+        int maxBackups = 6;
+        long accountId = 1L;
+        long accountLimit = 5L;
+        long domainId = 10L;
+        long domainLimit = -1L;
+
+        when(vmInstanceVOMock.getAccountId()).thenReturn(accountId);
+        when(accountManager.getAccount(accountId)).thenReturn(accountVOMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForAccount(accountVOMock, Resource.ResourceType.backup, null)).thenReturn(accountLimit);
+        when(accountVOMock.getDomainId()).thenReturn(domainId);
+        when(domainManager.getDomain(domainId)).thenReturn(domainMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForDomain(domainMock, Resource.ResourceType.backup, null)).thenReturn(domainLimit);
+        when(accountVOMock.getId()).thenReturn(accountId);
+        when(accountManager.isRootAdmin(accountId)).thenReturn(false);
+
+        backupManager.validateAndGetDefaultBackupRetentionIfRequired(maxBackups, backupOfferingVOMock, vmInstanceVOMock);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void validateAndGetDefaultBackupRetentionIfRequiredTestThrowExceptionWhenMaxBackupsExceedsDomainLimit() {
+        int maxBackups = 6;
+        long accountId = 1L;
+        long accountLimit = -1L;
+        long domainId = 10L;
+        long domainLimit = 5L;
+
+        when(vmInstanceVOMock.getAccountId()).thenReturn(accountId);
+        when(accountManager.getAccount(accountId)).thenReturn(accountVOMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForAccount(accountVOMock, Resource.ResourceType.backup, null)).thenReturn(accountLimit);
+        when(accountVOMock.getDomainId()).thenReturn(domainId);
+        when(domainManager.getDomain(domainId)).thenReturn(domainMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForDomain(domainMock, Resource.ResourceType.backup, null)).thenReturn(domainLimit);
+        when(accountVOMock.getId()).thenReturn(accountId);
+        when(accountManager.isRootAdmin(accountId)).thenReturn(false);
+
+        backupManager.validateAndGetDefaultBackupRetentionIfRequired(maxBackups, backupOfferingVOMock, vmInstanceVOMock);
+    }
+
+    @Test
+    public void validateAndGetDefaultBackupRetentionIfRequiredTestIgnoreLimitCheckWhenAccountIsRootAdmin() {
+        int maxBackups = 6;
+        long accountId = 1L;
+        long accountLimit = 5L;
+        long domainId = 10L;
+        long domainLimit = 5L;
+
+        when(vmInstanceVOMock.getAccountId()).thenReturn(accountId);
+        when(accountManager.getAccount(accountId)).thenReturn(accountVOMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForAccount(accountVOMock, Resource.ResourceType.backup, null)).thenReturn(accountLimit);
+        when(accountVOMock.getDomainId()).thenReturn(domainId);
+        when(domainManager.getDomain(domainId)).thenReturn(domainMock);
+        when(resourceLimitMgr.findCorrectResourceLimitForDomain(domainMock, Resource.ResourceType.backup, null)).thenReturn(domainLimit);
+        when(accountVOMock.getId()).thenReturn(accountId);
+        when(accountManager.isRootAdmin(accountId)).thenReturn(true);
+
+        int retention = backupManager.validateAndGetDefaultBackupRetentionIfRequired(maxBackups, backupOfferingVOMock, vmInstanceVOMock);
+        assertEquals(maxBackups, retention);
+    }
+
+    @Test
+    public void getBackupScheduleTestReturnNullWhenBackupIsManual() {
+        String jobParams = "{}";
+        when(asyncJobVOMock.getCmdInfo()).thenReturn(jobParams);
+        when(asyncJobVOMock.getId()).thenReturn(1L);
+
+        Long backupScheduleId = backupManager.getBackupScheduleId(asyncJobVOMock);
+        assertNull(backupScheduleId);
+    }
+
+    @Test
+    public void getBackupScheduleTestReturnBackupScheduleIdWhenBackupIsScheduled() {
+        Map<String, String> params = Map.of(
+                ApiConstants.SCHEDULE_ID, "100"
+        );
+        String jobParams = gson.toJson(params);
+        when(asyncJobVOMock.getCmdInfo()).thenReturn(jobParams);
+        when(asyncJobVOMock.getId()).thenReturn(1L);
+
+        Long backupScheduleId = backupManager.getBackupScheduleId(asyncJobVOMock);
+        assertEquals(Long.valueOf("100"), backupScheduleId);
+    }
+
+    @Test
+    public void getBackupScheduleTestReturnNullWhenSpecifiedBackupScheduleIdIsNotALongValue() {
+        Map<String, String> params = Map.of(
+                ApiConstants.SCHEDULE_ID, "InvalidValue"
+        );
+        String jobParams = gson.toJson(params);
+        when(asyncJobVOMock.getCmdInfo()).thenReturn(jobParams);
+        when(asyncJobVOMock.getId()).thenReturn(1L);
+
+        Long backupScheduleId = backupManager.getBackupScheduleId(asyncJobVOMock);
+        assertNull(backupScheduleId);
+    }
+
+    @Test
+    public void deleteOldestBackupFromScheduleIfRequiredTestSkipDeletionWhenBackupScheduleIsNotFound() {
+        backupManager.deleteOldestBackupFromScheduleIfRequired(1L, 1L);
+        Mockito.verify(backupManager, Mockito.never()).deleteExcessBackups(Mockito.anyList(), Mockito.anyInt(), Mockito.anyLong());
+    }
+
+    @Test
+    public void deleteOldestBackupFromScheduleIfRequiredTestSkipDeletionWhenRetentionIsEqualToZero() {
+        Mockito.when(backupScheduleDao.findById(1L)).thenReturn(backupScheduleVOMock);
+        Mockito.when(backupScheduleVOMock.getMaxBackups()).thenReturn(0);
+        backupManager.deleteOldestBackupFromScheduleIfRequired(1L, 1L);
+        Mockito.verify(backupManager, Mockito.never()).deleteExcessBackups(Mockito.anyList(), Mockito.anyInt(), Mockito.anyLong());
+    }
+
+    @Test
+    public void deleteOldestBackupFromScheduleIfRequiredTestSkipDeletionWhenAmountOfBackupsToBeDeletedIsLessThanOne() {
+        List<BackupVO> backups = List.of(Mockito.mock(BackupVO.class), Mockito.mock(BackupVO.class));
+        Mockito.when(backupScheduleDao.findById(1L)).thenReturn(backupScheduleVOMock);
+        Mockito.when(backupScheduleVOMock.getMaxBackups()).thenReturn(2);
+        Mockito.when(backupDao.listBySchedule(1L)).thenReturn(backups);
+        backupManager.deleteOldestBackupFromScheduleIfRequired(1L, 1L);
+        Mockito.verify(backupManager, Mockito.never()).deleteExcessBackups(Mockito.anyList(), Mockito.anyInt(), Mockito.anyLong());
+    }
+
+    @Test
+    public void deleteOldestBackupFromScheduleIfRequiredTestDeleteBackupsWhenRequired() {
+        List<BackupVO> backups = List.of(Mockito.mock(BackupVO.class), Mockito.mock(BackupVO.class));
+        Mockito.when(backupScheduleDao.findById(1L)).thenReturn(backupScheduleVOMock);
+        Mockito.when(backupScheduleVOMock.getMaxBackups()).thenReturn(1);
+        Mockito.when(backupDao.listBySchedule(1L)).thenReturn(backups);
+        Mockito.doNothing().when(backupManager).deleteExcessBackups(Mockito.anyList(), Mockito.anyInt(), Mockito.anyLong());
+        backupManager.deleteOldestBackupFromScheduleIfRequired(1L, 1L);
+        Mockito.verify(backupManager).deleteExcessBackups(Mockito.anyList(), Mockito.anyInt(), Mockito.anyLong());
+    }
+
+    @Test
+    public void deleteExcessBackupsTestEnsureBackupsAreDeletedWhenMethodIsCalled() {
+        try (MockedStatic<ActionEventUtils> actionEventUtils = Mockito.mockStatic(ActionEventUtils.class)) {
+            List<BackupVO> backups = List.of(Mockito.mock(BackupVO.class),
+                    Mockito.mock(BackupVO.class),
+                    Mockito.mock(BackupVO.class));
+
+            Mockito.when(backups.get(0).getId()).thenReturn(1L);
+            Mockito.when(backups.get(1).getId()).thenReturn(2L);
+            Mockito.when(backups.get(0).getAccountId()).thenReturn(1L);
+            Mockito.when(backups.get(1).getAccountId()).thenReturn(2L);
+            Mockito.doReturn(true).when(backupManager).deleteBackup(Mockito.anyLong(), Mockito.eq(false));
+
+            actionEventUtils.when(() -> ActionEventUtils.onStartedActionEvent(
+                    Mockito.anyLong(), Mockito.anyLong(), Mockito.anyString(),
+                    Mockito.anyString(), Mockito.anyLong(), Mockito.anyString(),
+                    Mockito.anyBoolean(), Mockito.anyInt())).thenReturn(1L);
+            actionEventUtils.when(() -> ActionEventUtils.onCompletedActionEvent(
+                    Mockito.anyLong(), Mockito.anyLong(), Mockito.anyString(),
+                    Mockito.anyString(), Mockito.anyString(), Mockito.anyLong(),
+                    Mockito.anyString(), Mockito.anyInt())).thenReturn(2L);
+
+            backupManager.deleteExcessBackups(backups, 2, 1L);
+            Mockito.verify(backupManager, times(2)).deleteBackup(Mockito.anyLong(), Mockito.eq(false));
+        }
+    }
+
+    @Test
+    public void sendExceededBackupLimitAlertTestSendAlertForBackupResourceType() {
+        String accountUuid = UUID.randomUUID().toString();
+        String expectedMessage = "Failed to create backup: backup resource limit exceeded for account with ID: " + accountUuid + ".";
+        String expectedAlertDetails = expectedMessage + " Please, use the 'updateResourceLimit' API to increase the backup limit.";
+
+        backupManager.sendExceededBackupLimitAlert(accountUuid, Resource.ResourceType.backup);
+        verify(alertManagerMock).sendAlert(
+                AlertManager.AlertType.ALERT_TYPE_UPDATE_RESOURCE_COUNT,
+                0L,
+                0L,
+                expectedMessage,
+                expectedAlertDetails
+        );
+    }
+
+    @Test
+    public void sendExceededBackupLimitAlertTestSendAlertForBackupStorageResourceType() {
+        String accountUuid = UUID.randomUUID().toString();
+        String expectedMessage = "Failed to create backup: backup storage space resource limit exceeded for account with ID: " + accountUuid + ".";
+        String expectedAlertDetails = expectedMessage + " Please, use the 'updateResourceLimit' API to increase the backup limit.";
+
+        backupManager.sendExceededBackupLimitAlert(accountUuid, Resource.ResourceType.backup_storage);
+        verify(alertManagerMock).sendAlert(
+                AlertManager.AlertType.ALERT_TYPE_UPDATE_RESOURCE_COUNT,
+                0L,
+                0L,
+                expectedMessage,
+                expectedAlertDetails
+        );
+    }
 }

ui/src/views/compute/backup/FormSchedule.vue

@@ -112,7 +112,7 @@
                   <a-input-number
                     style="width: 100%"
                     v-model:value="form.maxbackups"
-                    :min="1" />
+                    :min="0" />
                 </a-tooltip>
               </a-form-item>
             </a-col>