API: Add support to list all snapshot policies & backup schedules (#11587)

* API: Add support to list all snapshot policies & backup schedules

* Add support for backup policy listing without tying it to the vmid

* add tests for snapshot policy listing

* update tests for listbackupschedules

* remove trailing spaces and fix lint failure

* Add upgrade test

* remove unused import

* add create policy - snap/backup in the list view with resource (volume/vm) selection

* add translations

* refresh parent list

* remove unnecessary alert info

* fix checks for UI backup schedule list view

* fix checks for UI backup schedule list view

* add back access checks

* add since param

* fix failing test

* update snapshot policy and backup schedule ownership when VM is moved

* fix issue with showing vm selection

* fix unit test failure

* Update list snappolicy & backup schedule logic to list only those that belong to a proj or for root admin those that belong to it, unless listall & projid is passed

* fix test

* support snap / backup policy search using keyword

* fix tests

api/src/main/java/com/cloud/storage/snapshot/SnapshotApiService.java

@@ -85,7 +85,7 @@ public interface SnapshotApiService {
      *            the command that specifies the volume criteria
      * @return list of snapshot policies
      */
-    Pair<List<? extends SnapshotPolicy>, Integer> listPoliciesforVolume(ListSnapshotPoliciesCmd cmd);
+    Pair<List<? extends SnapshotPolicy>, Integer> listSnapshotPolicies(ListSnapshotPoliciesCmd cmd);
 
     boolean deleteSnapshotPolicies(DeleteSnapshotPoliciesCmd cmd);
 

api/src/main/java/com/cloud/storage/snapshot/SnapshotPolicy.java

@@ -16,11 +16,12 @@
 // under the License.
 package com.cloud.storage.snapshot;
 
+import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.api.Displayable;
 import org.apache.cloudstack.api.Identity;
 import org.apache.cloudstack.api.InternalIdentity;
 
-public interface SnapshotPolicy extends Identity, InternalIdentity, Displayable {
+public interface SnapshotPolicy extends ControlledEntity, Identity, InternalIdentity, Displayable {
 
     long getVolumeId();
 

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

@@ -24,7 +24,7 @@ import org.apache.cloudstack.acl.RoleType;
 import org.apache.cloudstack.api.APICommand;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.ApiErrorCode;
-import org.apache.cloudstack.api.BaseCmd;
+import org.apache.cloudstack.api.BaseListProjectAndAccountResourcesCmd;
 import org.apache.cloudstack.api.Parameter;
 import org.apache.cloudstack.api.ServerApiException;
 import org.apache.cloudstack.api.response.BackupScheduleResponse;
@@ -39,7 +39,6 @@ import com.cloud.exception.InsufficientCapacityException;
 import com.cloud.exception.NetworkRuleConflictException;
 import com.cloud.exception.ResourceAllocationException;
 import com.cloud.exception.ResourceUnavailableException;
-import com.cloud.utils.exception.CloudRuntimeException;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -48,10 +47,10 @@ import java.util.List;
         description = "List backup schedule of a VM",
         responseObject = BackupScheduleResponse.class, since = "4.14.0",
         authorized = {RoleType.Admin, RoleType.ResourceAdmin, RoleType.DomainAdmin, RoleType.User})
-public class ListBackupScheduleCmd extends BaseCmd {
+public class ListBackupScheduleCmd extends BaseListProjectAndAccountResourcesCmd {
 
     @Inject
-    private BackupManager backupManager;
+    BackupManager backupManager;
 
     /////////////////////////////////////////////////////
     //////////////// API parameters /////////////////////
@@ -60,10 +59,16 @@ public class ListBackupScheduleCmd extends BaseCmd {
     @Parameter(name = ApiConstants.VIRTUAL_MACHINE_ID,
             type = CommandType.UUID,
             entityType = UserVmResponse.class,
-            required = true,
             description = "ID of the VM")
     private Long vmId;
 
+    @Parameter(name = ApiConstants.ID,
+            type = CommandType.UUID,
+            entityType = BackupScheduleResponse.class,
+            description = "the ID of the backup schedule",
+            since = "4.22.0")
+    private Long id;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -72,6 +77,10 @@ public class ListBackupScheduleCmd extends BaseCmd {
         return vmId;
     }
 
+    public Long getId() {
+        return id;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
@@ -79,19 +88,18 @@ public class ListBackupScheduleCmd extends BaseCmd {
     @Override
     public void execute() throws ResourceUnavailableException, InsufficientCapacityException, ServerApiException, ConcurrentOperationException, ResourceAllocationException, NetworkRuleConflictException {
         try{
-            List<BackupSchedule> schedules = backupManager.listBackupSchedule(getVmId());
+            List<BackupSchedule> schedules = backupManager.listBackupSchedules(this);
             ListResponse<BackupScheduleResponse> response = new ListResponse<>();
             List<BackupScheduleResponse> scheduleResponses = new ArrayList<>();
+
             if (!CollectionUtils.isNullOrEmpty(schedules)) {
                 for (BackupSchedule schedule : schedules) {
                     scheduleResponses.add(_responseGenerator.createBackupScheduleResponse(schedule));
                 }
+            }
             response.setResponses(scheduleResponses, schedules.size());
             response.setResponseName(getCommandName());
             setResponseObject(response);
-            } else {
-                throw new CloudRuntimeException("No backup schedule exists for the VM");
-            }
         } catch (Exception e) {
             throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, e.getMessage());
         }

api/src/main/java/org/apache/cloudstack/api/command/user/snapshot/ListSnapshotPoliciesCmd.java

@@ -23,7 +23,7 @@ import org.apache.cloudstack.acl.RoleType;
 
 import org.apache.cloudstack.api.APICommand;
 import org.apache.cloudstack.api.ApiConstants;
-import org.apache.cloudstack.api.BaseListCmd;
+import org.apache.cloudstack.api.BaseListProjectAndAccountResourcesCmd;
 import org.apache.cloudstack.api.Parameter;
 import org.apache.cloudstack.api.response.ListResponse;
 import org.apache.cloudstack.api.response.SnapshotPolicyResponse;
@@ -34,7 +34,7 @@ import com.cloud.utils.Pair;
 
 @APICommand(name = "listSnapshotPolicies", description = "Lists snapshot policies.", responseObject = SnapshotPolicyResponse.class,
         requestHasSensitiveInfo = false, responseHasSensitiveInfo = false)
-public class ListSnapshotPoliciesCmd extends BaseListCmd {
+public class ListSnapshotPoliciesCmd extends BaseListProjectAndAccountResourcesCmd {
 
 
     /////////////////////////////////////////////////////
@@ -69,13 +69,14 @@ public class ListSnapshotPoliciesCmd extends BaseListCmd {
     public Long getId() {
         return id;
     }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
 
     @Override
     public void execute() {
-        Pair<List<? extends SnapshotPolicy>, Integer> result = _snapshotService.listPoliciesforVolume(this);
+        Pair<List<? extends SnapshotPolicy>, Integer> result = _snapshotService.listSnapshotPolicies(this);
         ListResponse<SnapshotPolicyResponse> response = new ListResponse<SnapshotPolicyResponse>();
         List<SnapshotPolicyResponse> policyResponses = new ArrayList<SnapshotPolicyResponse>();
         for (SnapshotPolicy policy : result.first()) {

api/src/main/java/org/apache/cloudstack/api/response/SnapshotPolicyResponse.java

@@ -37,6 +37,10 @@ public class SnapshotPolicyResponse extends BaseResponseWithTagInformation {
     @Param(description = "the ID of the disk volume")
     private String volumeId;
 
+    @SerializedName("volumename")
+    @Param(description = "the name of the disk volume")
+    private String volumeName;
+
     @SerializedName("schedule")
     @Param(description = "time the snapshot is scheduled to be taken.")
     private String schedule;
@@ -87,6 +91,10 @@ public class SnapshotPolicyResponse extends BaseResponseWithTagInformation {
         this.volumeId = volumeId;
     }
 
+    public void setVolumeName(String volumeName) {
+        this.volumeName = volumeName;
+    }
+
     public String getSchedule() {
         return schedule;
     }

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

@@ -28,6 +28,7 @@ import org.apache.cloudstack.api.command.user.backup.CreateBackupCmd;
 import org.apache.cloudstack.api.command.user.backup.CreateBackupScheduleCmd;
 import org.apache.cloudstack.api.command.user.backup.DeleteBackupScheduleCmd;
 import org.apache.cloudstack.api.command.user.backup.ListBackupOfferingsCmd;
+import org.apache.cloudstack.api.command.user.backup.ListBackupScheduleCmd;
 import org.apache.cloudstack.api.command.user.backup.ListBackupsCmd;
 import org.apache.cloudstack.api.response.BackupResponse;
 import org.apache.cloudstack.framework.config.ConfigKey;
@@ -174,7 +175,7 @@ public interface BackupManager extends BackupService, Configurable, PluggableSer
      * @param vmId
      * @return
      */
-    List<BackupSchedule> listBackupSchedule(Long vmId);
+    List<BackupSchedule> listBackupSchedules(ListBackupScheduleCmd cmd);
 
     /**
      * Deletes VM backup schedule for a VM

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

@@ -19,11 +19,12 @@ package org.apache.cloudstack.backup;
 
 import java.util.Date;
 
+import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.api.InternalIdentity;
 
 import com.cloud.utils.DateUtil;
 
-public interface BackupSchedule extends InternalIdentity {
+public interface BackupSchedule extends ControlledEntity, InternalIdentity {
     Long getVmId();
     DateUtil.IntervalType getScheduleType();
     String getSchedule();

api/src/test/java/org/apache/cloudstack/api/command/user/backup/ListBackupScheduleCmdTest.java

@@ -0,0 +1,98 @@
+// 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.api.command.user.backup;
+
+import com.cloud.exception.InsufficientCapacityException;
+import com.cloud.exception.NetworkRuleConflictException;
+import com.cloud.exception.ResourceAllocationException;
+import com.cloud.exception.ResourceUnavailableException;
+import com.cloud.user.Account;
+import org.apache.cloudstack.api.ResponseGenerator;
+import org.apache.cloudstack.api.response.BackupScheduleResponse;
+import org.apache.cloudstack.api.response.ListResponse;
+import org.apache.cloudstack.backup.BackupManager;
+import org.apache.cloudstack.backup.BackupSchedule;
+import org.apache.cloudstack.context.CallContext;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ListBackupScheduleCmdTest {
+
+    @Mock
+    private BackupManager backupManager;
+
+    @Mock
+    private ResponseGenerator responseGenerator;
+
+    private ListBackupScheduleCmd cmd;
+
+    @Before
+    public void setUp() {
+        cmd = new ListBackupScheduleCmd();
+        cmd.backupManager = backupManager;
+        cmd._responseGenerator = responseGenerator;
+    }
+
+    @Test
+    public void testExecuteWithSchedules() throws ResourceUnavailableException, InsufficientCapacityException, ResourceAllocationException, NetworkRuleConflictException {
+        BackupSchedule schedule = Mockito.mock(BackupSchedule.class);
+        BackupScheduleResponse scheduleResponse = Mockito.mock(BackupScheduleResponse.class);
+        List<BackupSchedule> schedules = new ArrayList<>();
+        schedules.add(schedule);
+
+        Mockito.when(backupManager.listBackupSchedules(cmd)).thenReturn(schedules);
+        Mockito.when(responseGenerator.createBackupScheduleResponse(schedule)).thenReturn(scheduleResponse);
+
+        Account mockAccount = Mockito.mock(Account.class);
+        CallContext callContext = Mockito.mock(CallContext.class);
+        try (org.mockito.MockedStatic<CallContext> mocked = Mockito.mockStatic(CallContext.class)) {
+            cmd.execute();
+        }
+
+        ListResponse<?> response = (ListResponse<?>) cmd.getResponseObject();
+        Assert.assertNotNull(response);
+        Assert.assertEquals(1, response.getResponses().size());
+        Assert.assertEquals(scheduleResponse, response.getResponses().get(0));
+    }
+
+    @Test
+    public void testExecuteWithNoSchedules() {
+        Mockito.when(backupManager.listBackupSchedules(cmd)).thenReturn(new ArrayList<>());
+        CallContext callContext = Mockito.mock(CallContext.class);
+
+        try (org.mockito.MockedStatic<CallContext> mocked = Mockito.mockStatic(CallContext.class)) {
+            mocked.when(CallContext::current).thenReturn(callContext);
+            cmd.execute();
+        } catch (ResourceUnavailableException | InsufficientCapacityException | ResourceAllocationException |
+                 NetworkRuleConflictException e) {
+            throw new RuntimeException(e);
+        }
+
+        ListResponse<?> response = (ListResponse<?>) cmd.getResponseObject();
+        Assert.assertNotNull(response);
+        Assert.assertEquals(0, response.getResponses().size());
+    }
+}

api/src/test/java/org/apache/cloudstack/api/command/user/snapshot/ListSnapshotPoliciesCmdTest.java

@@ -0,0 +1,79 @@
+// 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.api.command.user.snapshot;
+
+import com.cloud.storage.snapshot.SnapshotApiService;
+import com.cloud.storage.snapshot.SnapshotPolicy;
+import com.cloud.utils.Pair;
+import org.apache.cloudstack.api.ResponseGenerator;
+import org.apache.cloudstack.api.response.ListResponse;
+import org.apache.cloudstack.api.response.SnapshotPolicyResponse;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class ListSnapshotPoliciesCmdTest {
+    private ListSnapshotPoliciesCmd cmd;
+    private SnapshotApiService snapshotService;
+    private ResponseGenerator responseGenerator;
+
+    @Before
+    public void setUp() {
+        cmd = new ListSnapshotPoliciesCmd();
+        snapshotService = Mockito.mock(SnapshotApiService.class);
+        responseGenerator = Mockito.mock(ResponseGenerator.class);
+
+        cmd._snapshotService = snapshotService;
+        cmd._responseGenerator = responseGenerator;
+    }
+
+    @Test
+    public void testExecuteWithPolicies() {
+        SnapshotPolicy policy = Mockito.mock(SnapshotPolicy.class);
+        SnapshotPolicyResponse policyResponse = Mockito.mock(SnapshotPolicyResponse.class);
+        List<SnapshotPolicy> policies = new ArrayList<>();
+        policies.add(policy);
+
+        Mockito.when(snapshotService.listSnapshotPolicies(cmd))
+                .thenReturn(new Pair<>(policies, 1));
+        Mockito.when(responseGenerator.createSnapshotPolicyResponse(policy))
+                .thenReturn(policyResponse);
+
+        cmd.execute();
+
+        ListResponse<?> response = (ListResponse<?>) cmd.getResponseObject();
+        Assert.assertNotNull(response);
+        Assert.assertEquals(1, response.getResponses().size());
+        Assert.assertEquals(policyResponse, response.getResponses().get(0));
+    }
+
+    @Test
+    public void testExecuteWithNoPolicies() {
+        Mockito.when(snapshotService.listSnapshotPolicies(cmd))
+                .thenReturn(new Pair<>(new ArrayList<>(), 0));
+
+        cmd.execute();
+
+        ListResponse<?> response = (ListResponse<?>) cmd.getResponseObject();
+        Assert.assertNotNull(response);
+        Assert.assertTrue(response.getResponses().isEmpty());
+    }
+}

engine/schema/src/main/java/com/cloud/storage/SnapshotPolicyVO.java

@@ -59,6 +59,12 @@ public class SnapshotPolicyVO implements SnapshotPolicy {
     @Column(name = "uuid")
     String uuid;
 
+    @Column(name = "account_id")
+    long accountId;
+
+    @Column(name = "domain_id")
+    long domainId;
+
     @Column(name = "display", updatable = true, nullable = false)
     protected boolean display = true;
 
@@ -66,7 +72,7 @@ public class SnapshotPolicyVO implements SnapshotPolicy {
         this.uuid = UUID.randomUUID().toString();
     }
 
-    public SnapshotPolicyVO(long volumeId, String schedule, String timezone, IntervalType intvType, int maxSnaps, boolean display) {
+    public SnapshotPolicyVO(long volumeId, String schedule, String timezone, IntervalType intvType, int maxSnaps, long accountId, long domainId, boolean display) {
         this.volumeId = volumeId;
         this.schedule = schedule;
         this.timezone = timezone;
@@ -75,6 +81,8 @@ public class SnapshotPolicyVO implements SnapshotPolicy {
         this.active = true;
         this.display = display;
         this.uuid = UUID.randomUUID().toString();
+        this.accountId = accountId;
+        this.domainId = domainId;
     }
 
     @Override
@@ -160,4 +168,32 @@ public class SnapshotPolicyVO implements SnapshotPolicy {
     public void setDisplay(boolean display) {
         this.display = display;
     }
+
+    @Override
+    public long getAccountId() {
+        return accountId;
+    }
+
+    public void setAccountId(long accountId) {
+        this.accountId = accountId;
+    }
+
+    @Override
+    public long getDomainId() {
+        return domainId;
+    }
+
+    public void setDomainId(long domainId) {
+        this.domainId = domainId;
+    }
+
+    @Override
+    public Class<?> getEntityType() {
+        return SnapshotPolicy.class;
+    }
+
+    @Override
+    public String getName() {
+        return null;
+    }
 }

engine/schema/src/main/java/com/cloud/upgrade/dao/Upgrade42100to42200.java

@@ -16,6 +16,14 @@
 // under the License.
 package com.cloud.upgrade.dao;
 
+import com.cloud.utils.exception.CloudRuntimeException;
+
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
 public class Upgrade42100to42200 extends DbUpgradeAbstractImpl implements DbUpgrade, DbUpgradeSystemVmTemplate {
 
     @Override
@@ -27,4 +35,69 @@ public class Upgrade42100to42200 extends DbUpgradeAbstractImpl implements DbUpgr
     public String getUpgradedVersion() {
         return "4.22.0.0";
     }
+
+    @Override
+    public InputStream[] getPrepareScripts() {
+        final String scriptFile = "META-INF/db/schema-42100to42200.sql";
+        final InputStream script = Thread.currentThread().getContextClassLoader().getResourceAsStream(scriptFile);
+        if (script == null) {
+            throw new CloudRuntimeException("Unable to find " + scriptFile);
+        }
+
+        return new InputStream[] {script};
+    }
+
+    @Override
+    public void performDataMigration(Connection conn) {
+        updateSnapshotPolicyOwnership(conn);
+        updateBackupScheduleOwnership(conn);
+    }
+
+    protected void updateSnapshotPolicyOwnership(Connection conn) {
+        // set account_id and domain_id in snapshot_policy table from volume table
+        String selectSql = "SELECT sp.id, v.account_id, v.domain_id FROM snapshot_policy sp, volumes v WHERE sp.volume_id = v.id AND (sp.account_id IS NULL AND sp.domain_id IS NULL)";
+        String updateSql = "UPDATE snapshot_policy SET account_id = ?, domain_id = ? WHERE id = ?";
+
+        try (PreparedStatement selectPstmt = conn.prepareStatement(selectSql);
+             ResultSet rs = selectPstmt.executeQuery();
+             PreparedStatement updatePstmt = conn.prepareStatement(updateSql)) {
+
+            while (rs.next()) {
+                long policyId = rs.getLong(1);
+                long accountId = rs.getLong(2);
+                long domainId = rs.getLong(3);
+
+                updatePstmt.setLong(1, accountId);
+                updatePstmt.setLong(2, domainId);
+                updatePstmt.setLong(3, policyId);
+                updatePstmt.executeUpdate();
+            }
+        } catch (SQLException e) {
+            throw new CloudRuntimeException("Unable to update snapshot_policy table with account_id and domain_id", e);
+        }
+    }
+
+    protected void updateBackupScheduleOwnership(Connection conn) {
+        // Set account_id and domain_id in backup_schedule table from vm_instance table
+        String selectSql = "SELECT bs.id, vm.account_id, vm.domain_id FROM backup_schedule bs, vm_instance vm WHERE bs.vm_id = vm.id AND (bs.account_id IS NULL AND bs.domain_id IS NULL)";
+        String updateSql = "UPDATE backup_schedule SET account_id = ?, domain_id = ? WHERE id = ?";
+
+        try (PreparedStatement selectPstmt = conn.prepareStatement(selectSql);
+             ResultSet rs = selectPstmt.executeQuery();
+             PreparedStatement updatePstmt = conn.prepareStatement(updateSql)) {
+
+            while (rs.next()) {
+                long scheduleId = rs.getLong(1);
+                long accountId = rs.getLong(2);
+                long domainId = rs.getLong(3);
+
+                updatePstmt.setLong(1, accountId);
+                updatePstmt.setLong(2, domainId);
+                updatePstmt.setLong(3, scheduleId);
+                updatePstmt.executeUpdate();
+            }
+        } catch (SQLException e) {
+            throw new CloudRuntimeException("Unable to update backup_schedule table with account_id and domain_id", e);
+        }
+    }
 }

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

@@ -68,10 +68,16 @@ public class BackupScheduleVO implements BackupSchedule {
     @Column(name = "quiescevm")
     Boolean quiesceVM = false;
 
+    @Column(name = "account_id")
+    Long accountId;
+
+    @Column(name = "domain_id")
+    Long domainId;
+
     public BackupScheduleVO() {
     }
 
-    public BackupScheduleVO(Long vmId, DateUtil.IntervalType scheduleType, String schedule, String timezone, Date scheduledTimestamp, int maxBackups, Boolean quiesceVM) {
+    public BackupScheduleVO(Long vmId, DateUtil.IntervalType scheduleType, String schedule, String timezone, Date scheduledTimestamp, int maxBackups, Boolean quiesceVM, Long accountId, Long domainId) {
         this.vmId = vmId;
         this.scheduleType = (short) scheduleType.ordinal();
         this.schedule = schedule;
@@ -79,6 +85,8 @@ public class BackupScheduleVO implements BackupSchedule {
         this.scheduledTimestamp = scheduledTimestamp;
         this.maxBackups = maxBackups;
         this.quiesceVM = quiesceVM;
+        this.accountId = accountId;
+        this.domainId = domainId;
     }
 
     @Override
@@ -161,4 +169,32 @@ public class BackupScheduleVO implements BackupSchedule {
     public Boolean getQuiesceVM() {
         return quiesceVM;
     }
+
+    @Override
+    public Class<?> getEntityType() {
+        return BackupSchedule.class;
+    }
+
+    @Override
+    public String getName() {
+        return null;
+    }
+
+    @Override
+    public long getDomainId() {
+        return domainId;
+    }
+
+    @Override
+    public long getAccountId() {
+        return accountId;
+    }
+
+    public void setAccountId(Long accountId) {
+        this.accountId = accountId;
+    }
+
+    public void setDomainId(Long domainId) {
+        this.domainId = domainId;
+    }
 }

engine/schema/src/main/resources/META-INF/db/schema-42100to42200.sql

@@ -26,6 +26,12 @@ CALL `cloud`.`IDEMPOTENT_CHANGE_COLUMN`('router_health_check', 'check_result', '
 -- Increase length of scripts_version column to 128 due to md5sum to sha512sum change
 CALL `cloud`.`IDEMPOTENT_CHANGE_COLUMN`('cloud.domain_router', 'scripts_version', 'scripts_version', 'VARCHAR(128)');
 
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.snapshot_policy','domain_id', 'BIGINT(20) DEFAULT NULL');
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.snapshot_policy','account_id', 'BIGINT(20) DEFAULT NULL');
+
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.backup_schedule','domain_id', 'BIGINT(20) DEFAULT NULL');
+CALL `cloud`.`IDEMPOTENT_ADD_COLUMN`('cloud.backup_schedule','account_id', 'BIGINT(20) DEFAULT NULL');
+
 -- Increase the cache_mode column size from cloud.disk_offering table
 CALL `cloud`.`IDEMPOTENT_CHANGE_COLUMN`('cloud.disk_offering', 'cache_mode', 'cache_mode', 'varchar(18) DEFAULT "none" COMMENT "The disk cache mode to use for disks created with this offering"');
 

engine/schema/src/test/java/com/cloud/upgrade/dao/Upgrade42100to42200Test.java

@@ -0,0 +1,242 @@
+// 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.upgrade.dao;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.inOrder;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class Upgrade42100to42200Test {
+
+    @Spy
+    Upgrade42100to42200 upgrade;
+
+    @Mock
+    private Connection conn;
+
+    @Mock
+    private PreparedStatement selectStmt;
+
+    @Mock
+    private PreparedStatement updateStmt;
+
+    @Mock
+    private ResultSet resultSet;
+
+    @Test
+    public void testUpdateSnapshotPolicyOwnership() throws SQLException {
+        // Setup mock data for snapshot policies without ownership
+        when(conn.prepareStatement("SELECT sp.id, v.account_id, v.domain_id FROM snapshot_policy sp, volumes v WHERE sp.volume_id = v.id AND (sp.account_id IS NULL AND sp.domain_id IS NULL)"))
+                .thenReturn(selectStmt);
+        when(conn.prepareStatement("UPDATE snapshot_policy SET account_id = ?, domain_id = ? WHERE id = ?"))
+                .thenReturn(updateStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+
+        when(resultSet.next())
+                .thenReturn(true)
+                .thenReturn(true)
+                .thenReturn(false);
+
+        when(resultSet.getLong(1))
+                .thenReturn(1L)
+                .thenReturn(2L);
+        when(resultSet.getLong(2))
+                .thenReturn(100L)
+                .thenReturn(200L);
+        when(resultSet.getLong(3))
+                .thenReturn(1L)
+                .thenReturn(2L);
+
+        upgrade.updateSnapshotPolicyOwnership(conn);
+
+        verify(conn).prepareStatement("SELECT sp.id, v.account_id, v.domain_id FROM snapshot_policy sp, volumes v WHERE sp.volume_id = v.id AND (sp.account_id IS NULL AND sp.domain_id IS NULL)");
+        verify(conn).prepareStatement("UPDATE snapshot_policy SET account_id = ?, domain_id = ? WHERE id = ?");
+
+        InOrder inOrder = inOrder(updateStmt);
+
+        inOrder.verify(updateStmt).setLong(1, 100L); // account_id
+        inOrder.verify(updateStmt).setLong(2, 1L);   // domain_id
+        inOrder.verify(updateStmt).setLong(3, 1L);   // policy_id
+        inOrder.verify(updateStmt).executeUpdate();
+
+        inOrder.verify(updateStmt).setLong(1, 200L); // account_id
+        inOrder.verify(updateStmt).setLong(2, 2L);   // domain_id
+        inOrder.verify(updateStmt).setLong(3, 2L);   // policy_id
+        inOrder.verify(updateStmt).executeUpdate();
+
+        verify(updateStmt, times(2)).executeUpdate();
+    }
+
+    @Test
+    public void testUpdateBackupScheduleOwnership() throws SQLException {
+        when(conn.prepareStatement("SELECT bs.id, vm.account_id, vm.domain_id FROM backup_schedule bs, vm_instance vm WHERE bs.vm_id = vm.id AND (bs.account_id IS NULL AND bs.domain_id IS NULL)"))
+                .thenReturn(selectStmt);
+        when(conn.prepareStatement("UPDATE backup_schedule SET account_id = ?, domain_id = ? WHERE id = ?"))
+                .thenReturn(updateStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+
+        when(resultSet.next())
+                .thenReturn(true)
+                .thenReturn(true)
+                .thenReturn(true)
+                .thenReturn(false);
+
+        when(resultSet.getLong(1))
+                .thenReturn(10L)
+                .thenReturn(20L)
+                .thenReturn(30L);
+        when(resultSet.getLong(2))
+                .thenReturn(500L)
+                .thenReturn(600L)
+                .thenReturn(700L);
+        when(resultSet.getLong(3))
+                .thenReturn(5L)
+                .thenReturn(6L)
+                .thenReturn(7L);
+
+        upgrade.updateBackupScheduleOwnership(conn);
+
+        verify(conn).prepareStatement("SELECT bs.id, vm.account_id, vm.domain_id FROM backup_schedule bs, vm_instance vm WHERE bs.vm_id = vm.id AND (bs.account_id IS NULL AND bs.domain_id IS NULL)");
+        verify(conn).prepareStatement("UPDATE backup_schedule SET account_id = ?, domain_id = ? WHERE id = ?");
+
+        InOrder inOrder = inOrder(updateStmt);
+
+        inOrder.verify(updateStmt).setLong(1, 500L);
+        inOrder.verify(updateStmt).setLong(2, 5L);
+        inOrder.verify(updateStmt).setLong(3, 10L);
+        inOrder.verify(updateStmt).executeUpdate();
+
+        inOrder.verify(updateStmt).setLong(1, 600L);
+        inOrder.verify(updateStmt).setLong(2, 6L);
+        inOrder.verify(updateStmt).setLong(3, 20L);
+        inOrder.verify(updateStmt).executeUpdate();
+
+        inOrder.verify(updateStmt).setLong(1, 700L);
+        inOrder.verify(updateStmt).setLong(2, 7L);
+        inOrder.verify(updateStmt).setLong(3, 30L);
+        inOrder.verify(updateStmt).executeUpdate();
+
+        verify(updateStmt, times(3)).executeUpdate();
+    }
+
+    @Test
+    public void testUpdateSnapshotPolicyOwnershipNoResults() throws SQLException {
+        when(conn.prepareStatement("SELECT sp.id, v.account_id, v.domain_id FROM snapshot_policy sp, volumes v WHERE sp.volume_id = v.id AND (sp.account_id IS NULL AND sp.domain_id IS NULL)"))
+                .thenReturn(selectStmt);
+        when(conn.prepareStatement("UPDATE snapshot_policy SET account_id = ?, domain_id = ? WHERE id = ?"))
+                .thenReturn(updateStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+
+        when(resultSet.next()).thenReturn(false);
+
+        upgrade.updateSnapshotPolicyOwnership(conn);
+
+        verify(selectStmt).executeQuery();
+        verify(updateStmt, times(0)).executeUpdate();
+    }
+
+    @Test
+    public void testUpdateBackupScheduleOwnershipNoResults() throws SQLException {
+        when(conn.prepareStatement("SELECT bs.id, vm.account_id, vm.domain_id FROM backup_schedule bs, vm_instance vm WHERE bs.vm_id = vm.id AND (bs.account_id IS NULL AND bs.domain_id IS NULL)"))
+                .thenReturn(selectStmt);
+        when(conn.prepareStatement("UPDATE backup_schedule SET account_id = ?, domain_id = ? WHERE id = ?"))
+                .thenReturn(updateStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+
+        when(resultSet.next()).thenReturn(false);
+
+        upgrade.updateBackupScheduleOwnership(conn);
+
+        verify(selectStmt).executeQuery();
+        verify(updateStmt, times(0)).executeUpdate();
+    }
+
+    @Test
+    public void testPerformDataMigration() throws SQLException {
+        when(conn.prepareStatement(anyString())).thenReturn(selectStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+        when(resultSet.next()).thenReturn(false);
+
+        upgrade.performDataMigration(conn);
+
+        verify(conn).prepareStatement("SELECT sp.id, v.account_id, v.domain_id FROM snapshot_policy sp, volumes v WHERE sp.volume_id = v.id AND (sp.account_id IS NULL AND sp.domain_id IS NULL)");
+        verify(conn).prepareStatement("SELECT bs.id, vm.account_id, vm.domain_id FROM backup_schedule bs, vm_instance vm WHERE bs.vm_id = vm.id AND (bs.account_id IS NULL AND bs.domain_id IS NULL)");
+    }
+
+    @Test
+    public void testUpdateSnapshotPolicyOwnershipSingleRecord() throws SQLException {
+        when(conn.prepareStatement("SELECT sp.id, v.account_id, v.domain_id FROM snapshot_policy sp, volumes v WHERE sp.volume_id = v.id AND (sp.account_id IS NULL AND sp.domain_id IS NULL)"))
+                .thenReturn(selectStmt);
+        when(conn.prepareStatement("UPDATE snapshot_policy SET account_id = ?, domain_id = ? WHERE id = ?"))
+                .thenReturn(updateStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+
+        when(resultSet.next())
+                .thenReturn(true)
+                .thenReturn(false);
+
+        when(resultSet.getLong(1)).thenReturn(42L);
+        when(resultSet.getLong(2)).thenReturn(999L);
+        when(resultSet.getLong(3)).thenReturn(10L);
+
+        upgrade.updateSnapshotPolicyOwnership(conn);
+
+        verify(updateStmt).setLong(1, 999L);
+        verify(updateStmt).setLong(2, 10L);
+        verify(updateStmt).setLong(3, 42L);
+        verify(updateStmt, times(1)).executeUpdate();
+    }
+
+    @Test
+    public void testUpdateBackupScheduleOwnershipSingleRecord() throws SQLException {
+        when(conn.prepareStatement("SELECT bs.id, vm.account_id, vm.domain_id FROM backup_schedule bs, vm_instance vm WHERE bs.vm_id = vm.id AND (bs.account_id IS NULL AND bs.domain_id IS NULL)"))
+                .thenReturn(selectStmt);
+        when(conn.prepareStatement("UPDATE backup_schedule SET account_id = ?, domain_id = ? WHERE id = ?"))
+                .thenReturn(updateStmt);
+        when(selectStmt.executeQuery()).thenReturn(resultSet);
+
+        when(resultSet.next())
+                .thenReturn(true)
+                .thenReturn(false);
+
+        when(resultSet.getLong(1)).thenReturn(55L);
+        when(resultSet.getLong(2)).thenReturn(888L);
+        when(resultSet.getLong(3)).thenReturn(15L);
+
+        upgrade.updateBackupScheduleOwnership(conn);
+
+        verify(updateStmt).setLong(1, 888L);
+        verify(updateStmt).setLong(2, 15L);
+        verify(updateStmt).setLong(3, 55L);
+        verify(updateStmt, times(1)).executeUpdate();
+    }
+}

engine/storage/integration-test/src/test/java/org/apache/cloudstack/storage/test/SnapshotTestWithFakeData.java

@@ -305,7 +305,7 @@ public class SnapshotTestWithFakeData {
     }
 
     protected SnapshotPolicyVO createSnapshotPolicy(Long volId) {
-        SnapshotPolicyVO policyVO = new SnapshotPolicyVO(volId, "jfkd", "fdfd", DateUtil.IntervalType.DAILY, 8, true);
+        SnapshotPolicyVO policyVO = new SnapshotPolicyVO(volId, "jfkd", "fdfd", DateUtil.IntervalType.DAILY, 8, 1, 1, true);
             policyVO = snapshotPolicyDao.persist(policyVO);
             return policyVO;
         }

server/src/main/java/com/cloud/api/ApiResponseHelper.java

@@ -853,6 +853,7 @@ public class ApiResponseHelper implements ResponseGenerator {
         Volume vol = ApiDBUtils.findVolumeById(policy.getVolumeId());
         if (vol != null) {
             policyResponse.setVolumeId(vol.getUuid());
+            policyResponse.setVolumeName(vol.getName());
         }
         policyResponse.setSchedule(policy.getSchedule());
         policyResponse.setIntervalType(policy.getInterval());

server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java

@@ -1303,7 +1303,8 @@ public class SnapshotManagerImpl extends MutualExclusiveIdsManagerBase implement
     }
 
     protected SnapshotPolicyVO createSnapshotPolicy(long volumeId, String schedule, String timezone, IntervalType intervalType, int maxSnaps, boolean display, List<Long> zoneIds, List<Long> poolIds) {
-        SnapshotPolicyVO policy = new SnapshotPolicyVO(volumeId, schedule, timezone, intervalType, maxSnaps, display);
+        VolumeVO volume = _volsDao.findById(volumeId);
+        SnapshotPolicyVO policy = new SnapshotPolicyVO(volumeId, schedule, timezone, intervalType, maxSnaps, volume.getAccountId(), volume.getDomainId(), display);
         policy = _snapshotPolicyDao.persist(policy);
         if (CollectionUtils.isNotEmpty(zoneIds)) {
             List<SnapshotPolicyDetailVO> details = new ArrayList<>();
@@ -1388,28 +1389,54 @@ public class SnapshotManagerImpl extends MutualExclusiveIdsManagerBase implement
     }
 
     @Override
-    public Pair<List<? extends SnapshotPolicy>, Integer> listPoliciesforVolume(ListSnapshotPoliciesCmd cmd) {
+    public Pair<List<? extends SnapshotPolicy>, Integer> listSnapshotPolicies(ListSnapshotPoliciesCmd cmd) {
         Long volumeId = cmd.getVolumeId();
-        boolean display = cmd.isDisplay();
         Long id = cmd.getId();
-        Pair<List<SnapshotPolicyVO>, Integer> result = null;
+        Account caller = CallContext.current().getCallingAccount();
-        // TODO - Have a better way of doing this.
+        List<Long> permittedAccounts = new ArrayList<>();
-        if (id != null) {
+        String keyword = cmd.getKeyword();
-            result = _snapshotPolicyDao.listAndCountById(id, display, null);
+
-            if (result != null && result.first() != null && !result.first().isEmpty()) {
+        // Verify parameters
-                SnapshotPolicyVO snapshotPolicy = result.first().get(0);
+        if (volumeId != null) {
-                volumeId = snapshotPolicy.getVolumeId();
-            }
-        }
             VolumeVO volume = _volsDao.findById(volumeId);
-        if (volume == null) {
+            if (volume != null) {
-            throw new InvalidParameterValueException("Unable to find a volume with id " + volumeId);
-        }
                 _accountMgr.checkAccess(CallContext.current().getCallingAccount(), null, true, volume);
-        if (result != null)
+            }
-            return new Pair<List<? extends SnapshotPolicy>, Integer>(result.first(), result.second());
+        }
-        result = _snapshotPolicyDao.listAndCountByVolumeId(volumeId, display);
+
-        return new Pair<List<? extends SnapshotPolicy>, Integer>(result.first(), result.second());
+        Ternary<Long, Boolean, ListProjectResourcesCriteria> domainIdRecursiveListProject =
+                new Ternary<>(cmd.getDomainId(), cmd.isRecursive(), null);
+        _accountMgr.buildACLSearchParameters(caller, id, cmd.getAccountName(), cmd.getProjectId(), permittedAccounts, domainIdRecursiveListProject, cmd.listAll(), false);
+        Long domainId = domainIdRecursiveListProject.first();
+        Boolean isRecursive = domainIdRecursiveListProject.second();
+        ListProjectResourcesCriteria listProjectResourcesCriteria = domainIdRecursiveListProject.third();
+
+        Filter searchFilter = new Filter(SnapshotPolicyVO.class, "id", false, cmd.getStartIndex(), cmd.getPageSizeVal());
+        SearchBuilder<SnapshotPolicyVO> policySearch = _snapshotPolicyDao.createSearchBuilder();
+        _accountMgr.buildACLSearchBuilder(policySearch, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
+
+        policySearch.and("id", policySearch.entity().getId(), SearchCriteria.Op.EQ);
+        policySearch.and("volumeId", policySearch.entity().getVolumeId(), SearchCriteria.Op.EQ);
+
+        SearchBuilder<VolumeVO> volumeSearch = _volsDao.createSearchBuilder();
+        volumeSearch.and("name", volumeSearch.entity().getName(), SearchCriteria.Op.LIKE);
+        policySearch.join("volumeJoin", volumeSearch, policySearch.entity().getVolumeId(), volumeSearch.entity().getId(), JoinBuilder.JoinType.INNER);
+
+        SearchCriteria<SnapshotPolicyVO> sc = policySearch.create();
+        _accountMgr.buildACLSearchCriteria(sc, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
+
+        if (volumeId != null) {
+            sc.setParameters("volumeId", volumeId);
+        }
+        if (id != null) {
+            sc.setParameters("id", id);
+        }
+        if (keyword != null) {
+            sc.setJoinParameters("volumeJoin", "name", "%" + keyword + "%");
+        }
+
+        Pair<List<SnapshotPolicyVO>, Integer> result = _snapshotPolicyDao.searchAndCount(sc, searchFilter);
+        return new Pair<>(result.first(), result.second());
     }
 
     private List<SnapshotPolicyVO> listPoliciesforVolume(long volumeId) {

server/src/main/java/com/cloud/vm/UserVmManagerImpl.java

@@ -60,6 +60,8 @@ import javax.naming.ConfigurationException;
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.ParserConfigurationException;
 
+import com.cloud.storage.SnapshotPolicyVO;
+import com.cloud.storage.dao.SnapshotPolicyDao;
 import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.acl.ControlledEntity.ACLType;
 import org.apache.cloudstack.acl.SecurityChecker.AccessType;
@@ -103,8 +105,10 @@ import org.apache.cloudstack.api.command.user.vmgroup.DeleteVMGroupCmd;
 import org.apache.cloudstack.api.command.user.volume.ChangeOfferingForVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.ResizeVolumeCmd;
 import org.apache.cloudstack.backup.BackupManager;
+import org.apache.cloudstack.backup.BackupScheduleVO;
 import org.apache.cloudstack.backup.BackupVO;
 import org.apache.cloudstack.backup.dao.BackupDao;
+import org.apache.cloudstack.backup.dao.BackupScheduleDao;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.cloud.entity.api.VirtualMachineEntity;
 import org.apache.cloudstack.engine.cloud.entity.api.db.dao.VMNetworkMapDao;
@@ -607,6 +611,10 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     ReservationDao reservationDao;
     @Inject
     ResourceLimitService resourceLimitService;
+    @Inject
+    SnapshotPolicyDao snapshotPolicyDao;
+    @Inject
+    BackupScheduleDao backupScheduleDao;
 
     @Inject
     private StatsCollector statsCollector;
@@ -8045,6 +8053,9 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
 
         updateVolumesOwner(volumes, oldAccount, newAccount, newAccountId);
 
+        updateSnapshotPolicyOwnership(volumes, newAccount);
+        updateBackupScheduleOwnership(vm, newAccount);
+
         try {
             updateVmNetwork(cmd, caller, vm, newAccount, template);
         } catch (InsufficientCapacityException | ResourceAllocationException e) {
@@ -8519,6 +8530,36 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         }
     }
 
+    protected void updateSnapshotPolicyOwnership(List<VolumeVO> volumes, Account newAccount) {
+        logger.debug("Updating snapshot policy ownership for volumes of VM being assigned to account [{}]", newAccount);
+
+        for (VolumeVO volume : volumes) {
+            List<SnapshotPolicyVO> snapshotPolicies = snapshotPolicyDao.listByVolumeId(volume.getId());
+            for (SnapshotPolicyVO policy : snapshotPolicies) {
+                logger.trace("Updating snapshot policy [{}] ownership from account [{}] to account [{}]",
+                        policy.getId(), policy.getAccountId(), newAccount.getAccountId());
+
+                policy.setAccountId(newAccount.getAccountId());
+                policy.setDomainId(newAccount.getDomainId());
+                snapshotPolicyDao.update(policy.getId(), policy);
+            }
+        }
+    }
+
+    protected void updateBackupScheduleOwnership(UserVmVO vm, Account newAccount) {
+        logger.debug("Updating backup schedule ownership for VM [{}] being assigned to account [{}]", vm, newAccount);
+
+        List<BackupScheduleVO> backupSchedules = backupScheduleDao.listByVM(vm.getId());
+        for (BackupScheduleVO schedule : backupSchedules) {
+            logger.trace("Updating backup schedule [{}] ownership from account [{}] to account [{}]",
+                    schedule.getId(), schedule.getAccountId(), newAccount.getAccountId());
+
+            schedule.setAccountId(newAccount.getAccountId());
+            schedule.setDomainId(newAccount.getDomainId());
+            backupScheduleDao.update(schedule.getId(), schedule);
+        }
+    }
+
     /**
      * Attempts to create a network suitable for the creation of a VM ({@link NetworkOrchestrationService#createGuestNetwork}).
      * If no physical network is found, throws a {@link InvalidParameterValueException}.

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

@@ -588,7 +588,7 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
 
         final BackupScheduleVO schedule = backupScheduleDao.findByVMAndIntervalType(vmId, intervalType);
         if (schedule == null) {
-            return backupScheduleDao.persist(new BackupScheduleVO(vmId, intervalType, scheduleString, timezoneId, nextDateTime, maxBackups, cmd.getQuiesceVM()));
+            return backupScheduleDao.persist(new BackupScheduleVO(vmId, intervalType, scheduleString, timezoneId, nextDateTime, maxBackups, cmd.getQuiesceVM(), vm.getAccountId(), vm.getDomainId()));
         }
 
         schedule.setScheduleType((short) intervalType.ordinal());
@@ -638,13 +638,59 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
         return maxBackups;
     }
 
-    @Override
+    public List<BackupSchedule> listBackupSchedules(ListBackupScheduleCmd cmd) {
-    public List<BackupSchedule> listBackupSchedule(final Long vmId) {
+        Account caller = CallContext.current().getCallingAccount();
+        Long id = cmd.getId();
+        Long vmId = cmd.getVmId();
+        List<Long> permittedAccounts = new ArrayList<>();
+        Long domainId = null;
+        Boolean isRecursive = null;
+        String keyword = cmd.getKeyword();
+        Project.ListProjectResourcesCriteria listProjectResourcesCriteria = null;
+
+        if (vmId != null) {
             final VMInstanceVO vm = findVmById(vmId);
             validateBackupForZone(vm.getDataCenterId());
             accountManager.checkAccess(CallContext.current().getCallingAccount(), null, true, vm);
+        }
 
-        return backupScheduleDao.listByVM(vmId).stream().map(BackupSchedule.class::cast).collect(Collectors.toList());
+        Ternary<Long, Boolean, Project.ListProjectResourcesCriteria> domainIdRecursiveListProject =
+                new Ternary<>(cmd.getDomainId(), cmd.isRecursive(), null);
+        accountManager.buildACLSearchParameters(caller, id, cmd.getAccountName(), cmd.getProjectId(), permittedAccounts, domainIdRecursiveListProject, true, false);
+        domainId = domainIdRecursiveListProject.first();
+        isRecursive = domainIdRecursiveListProject.second();
+        listProjectResourcesCriteria = domainIdRecursiveListProject.third();
+
+        Filter searchFilter = new Filter(BackupScheduleVO.class, "id", false, null, null);
+        SearchBuilder<BackupScheduleVO> searchBuilder = backupScheduleDao.createSearchBuilder();
+
+        accountManager.buildACLSearchBuilder(searchBuilder, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
+
+        searchBuilder.and("id", searchBuilder.entity().getId(), SearchCriteria.Op.EQ);
+        if (vmId != null) {
+            searchBuilder.and("vmId", searchBuilder.entity().getVmId(), SearchCriteria.Op.EQ);
+        }
+        if (keyword != null && !keyword.isEmpty()) {
+            SearchBuilder<VMInstanceVO> vmSearch = vmInstanceDao.createSearchBuilder();
+            vmSearch.and("hostName", vmSearch.entity().getHostName(), SearchCriteria.Op.LIKE);
+            searchBuilder.join("vmJoin", vmSearch, searchBuilder.entity().getVmId(), vmSearch.entity().getId(), JoinBuilder.JoinType.INNER);
+        }
+
+        SearchCriteria<BackupScheduleVO> sc = searchBuilder.create();
+        accountManager.buildACLSearchCriteria(sc, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
+
+        if (id != null) {
+            sc.setParameters("id", id);
+        }
+        if (vmId != null) {
+            sc.setParameters("vmId", vmId);
+        }
+        if (keyword != null && !keyword.isEmpty()) {
+            sc.setJoinParameters("vmJoin", "hostName", "%" + keyword + "%");
+        }
+
+        Pair<List<BackupScheduleVO>, Integer> result = backupScheduleDao.searchAndCount(sc, searchFilter);
+        return new ArrayList<>(result.first());
     }
 
     @Override

server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerImplTest.java

@@ -27,18 +27,25 @@ import com.cloud.exception.ResourceUnavailableException;
 import com.cloud.org.Grouping;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.Snapshot;
+import com.cloud.storage.SnapshotPolicyVO;
 import com.cloud.storage.SnapshotVO;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.SnapshotDao;
+import com.cloud.storage.dao.SnapshotPolicyDao;
 import com.cloud.storage.dao.SnapshotZoneDao;
 import com.cloud.storage.dao.VolumeDao;
 import com.cloud.user.Account;
 import com.cloud.user.AccountManager;
 import com.cloud.user.AccountVO;
 import com.cloud.user.ResourceLimitService;
+import com.cloud.user.User;
 import com.cloud.user.dao.AccountDao;
 import com.cloud.utils.Pair;
 
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+import org.apache.cloudstack.api.command.user.snapshot.ListSnapshotPoliciesCmd;
+import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.subsystem.api.storage.CreateCmdResult;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
@@ -51,6 +58,8 @@ import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreVO;
 
 import org.junit.Assert;
+import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -89,9 +98,23 @@ public class SnapshotManagerImplTest {
     SnapshotZoneDao snapshotZoneDao;
     @Mock
     VolumeDao volumeDao;
+    @Mock
+    SnapshotPolicyDao snapshotPolicyDao;
     @InjectMocks
     SnapshotManagerImpl snapshotManager = new SnapshotManagerImpl();
 
+    @Before
+    public void setUp() {
+        snapshotManager._snapshotPolicyDao = snapshotPolicyDao;
+        snapshotManager._volsDao = volumeDao;
+        snapshotManager._accountMgr = accountManager;
+    }
+
+    @After
+    public void tearDown() {
+        CallContext.unregister();
+    }
+
     @Test
     public void testGetSnapshotZoneImageStoreValid() {
         final long snapshotId = 1L;
@@ -395,4 +418,106 @@ public class SnapshotManagerImplTest {
         Mockito.when(dataCenterDao.findById(zoneId)).thenReturn(dataCenterVO);
         Assert.assertNotNull(snapshotManager.getCheckedDestinationZoneForSnapshotCopy(zoneId, false));
     }
+
+    @Test
+    public void testListSnapshotPolicies() {
+        long volumeId = 42L;
+        ListSnapshotPoliciesCmd cmd = Mockito.mock(ListSnapshotPoliciesCmd.class);
+        Mockito.when(cmd.getVolumeId()).thenReturn(volumeId);
+        Mockito.when(cmd.getId()).thenReturn(null);
+        Mockito.when(cmd.getStartIndex()).thenReturn(0L);
+        Mockito.when(cmd.getPageSizeVal()).thenReturn(10L);
+
+        Account caller = Mockito.mock(Account.class);
+        Mockito.when(caller.getId()).thenReturn(1L);
+        CallContext.register(Mockito.mock(User.class), caller);
+
+        SnapshotPolicyVO policy1 = Mockito.mock(SnapshotPolicyVO.class);
+        SnapshotPolicyVO policy2 = Mockito.mock(SnapshotPolicyVO.class);
+        List<SnapshotPolicyVO> mockPolicies = List.of(policy1, policy2);
+
+        SearchBuilder<SnapshotPolicyVO> mockSearchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchBuilder<VolumeVO> mockVolumeSearchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchCriteria<SnapshotPolicyVO> mockSearchCriteria = Mockito.mock(SearchCriteria.class);
+
+        Mockito.when(snapshotPolicyDao.createSearchBuilder()).thenReturn(mockSearchBuilder);
+        Mockito.when(mockSearchBuilder.entity()).thenReturn(Mockito.mock(SnapshotPolicyVO.class));
+        Mockito.when(mockSearchBuilder.create()).thenReturn(mockSearchCriteria);
+        Mockito.when(volumeDao.createSearchBuilder()).thenReturn(mockVolumeSearchBuilder);
+        Mockito.when(mockVolumeSearchBuilder.entity()).thenReturn(Mockito.mock(VolumeVO.class));
+        Mockito.when(snapshotPolicyDao.searchAndCount(Mockito.any(), Mockito.any())).thenReturn(new Pair<>(mockPolicies, 2));
+
+        Pair<List<? extends SnapshotPolicy>, Integer> result = snapshotManager.listSnapshotPolicies(cmd);
+
+        Assert.assertNotNull(result);
+        Assert.assertEquals(2, result.first().size());
+        Assert.assertEquals(Integer.valueOf(2), result.second());
+        Assert.assertEquals(mockPolicies, result.first());
+    }
+
+    @Test
+    public void testListSnapshotPolicies_NonRootAdmin() {
+        ListSnapshotPoliciesCmd cmd = Mockito.mock(ListSnapshotPoliciesCmd.class);
+        Mockito.when(cmd.getVolumeId()).thenReturn(1L);
+        Mockito.when(cmd.getId()).thenReturn(null);
+        Mockito.when(cmd.getStartIndex()).thenReturn(0L);
+        Mockito.when(cmd.getPageSizeVal()).thenReturn(10L);
+
+        Account caller = Mockito.mock(Account.class);
+        Mockito.when(caller.getId()).thenReturn(2L);
+        CallContext.register(Mockito.mock(User.class), caller);
+
+        SnapshotPolicyVO policy1 = Mockito.mock(SnapshotPolicyVO.class);
+        SnapshotPolicyVO policy2 = Mockito.mock(SnapshotPolicyVO.class);
+        List<SnapshotPolicyVO> mockPolicies = List.of(policy1, policy2);
+
+        SearchBuilder<SnapshotPolicyVO> mockSearchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchBuilder<VolumeVO> mockVolumeSearchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchCriteria<SnapshotPolicyVO> mockSearchCriteria = Mockito.mock(SearchCriteria.class);
+
+        Mockito.when(snapshotPolicyDao.createSearchBuilder()).thenReturn(mockSearchBuilder);
+        Mockito.when(mockSearchBuilder.entity()).thenReturn(Mockito.mock(SnapshotPolicyVO.class));
+        Mockito.when(mockSearchBuilder.create()).thenReturn(mockSearchCriteria);
+        Mockito.when(volumeDao.createSearchBuilder()).thenReturn(mockVolumeSearchBuilder);
+        Mockito.when(mockVolumeSearchBuilder.entity()).thenReturn(Mockito.mock(VolumeVO.class));
+        Mockito.when(snapshotPolicyDao.searchAndCount(Mockito.any(), Mockito.any())).thenReturn(new Pair<>(mockPolicies, 2));
+
+        Pair<List<? extends SnapshotPolicy>, Integer> result = snapshotManager.listSnapshotPolicies(cmd);
+
+        Assert.assertNotNull(result);
+        Assert.assertEquals(2, result.first().size());
+        Assert.assertEquals(Integer.valueOf(2), result.second());
+        Assert.assertEquals(mockPolicies, result.first());
+    }
+
+    @Test
+    public void testListSnapshotPolicies_RootAdmin() {
+        ListSnapshotPoliciesCmd cmd = Mockito.mock(ListSnapshotPoliciesCmd.class);
+        Mockito.when(cmd.getVolumeId()).thenReturn(1L);
+        Mockito.when(cmd.getId()).thenReturn(null);
+        Mockito.when(cmd.getStartIndex()).thenReturn(0L);
+        Mockito.when(cmd.getPageSizeVal()).thenReturn(10L);
+
+        Account caller = Mockito.mock(Account.class);
+        Mockito.when(caller.getId()).thenReturn(1L);
+        CallContext.register(Mockito.mock(User.class), caller);
+
+        SnapshotPolicyVO policy = Mockito.mock(SnapshotPolicyVO.class);
+        SearchBuilder<SnapshotPolicyVO> mockSearchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchBuilder<VolumeVO> mockVolumeSearchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchCriteria<SnapshotPolicyVO> mockSearchCriteria = Mockito.mock(SearchCriteria.class);
+
+        Mockito.when(snapshotPolicyDao.createSearchBuilder()).thenReturn(mockSearchBuilder);
+        Mockito.when(mockSearchBuilder.entity()).thenReturn(Mockito.mock(SnapshotPolicyVO.class));
+        Mockito.when(mockSearchBuilder.create()).thenReturn(mockSearchCriteria);
+        Mockito.when(volumeDao.createSearchBuilder()).thenReturn(mockVolumeSearchBuilder);
+        Mockito.when(mockVolumeSearchBuilder.entity()).thenReturn(Mockito.mock(VolumeVO.class));
+        Mockito.when(snapshotPolicyDao.searchAndCount(Mockito.any(), Mockito.any())).thenReturn(new Pair<>(List.of(policy), 1));
+
+        Pair<List<? extends SnapshotPolicy>, Integer> result = snapshotManager.listSnapshotPolicies(cmd);
+
+        Assert.assertNotNull(result);
+        Assert.assertEquals(1, result.first().size());
+        Assert.assertEquals(Integer.valueOf(1), result.second());
+    }
 }

server/src/test/java/com/cloud/storage/snapshot/SnapshotManagerTest.java

@@ -209,6 +209,8 @@ public class SnapshotManagerTest {
     private static final String TEST_SNAPSHOT_POLICY_TIMEZONE = "";
     private static final IntervalType TEST_SNAPSHOT_POLICY_INTERVAL = IntervalType.MONTHLY;
     private static final int TEST_SNAPSHOT_POLICY_MAX_SNAPS = 1;
+    private static final long TEST_SNAPSHOT_POLICY_ACCOUNT_ID = 1;
+    private static final long TEST_SNAPSHOT_POLICY_DOMAIN_ID = 1;
     private static final boolean TEST_SNAPSHOT_POLICY_DISPLAY = true;
     private static final boolean TEST_SNAPSHOT_POLICY_ACTIVE = true;
     private static final long TEST_ZONE_ID = 7L;
@@ -251,7 +253,7 @@ public class SnapshotManagerTest {
         when(_resourceMgr.listAllUpAndEnabledHostsInOneZoneByHypervisor(any(HypervisorType.class), anyLong())).thenReturn(null);
 
         snapshotPolicyVoInstance = new SnapshotPolicyVO(TEST_VOLUME_ID, TEST_SNAPSHOT_POLICY_SCHEDULE, TEST_SNAPSHOT_POLICY_TIMEZONE, TEST_SNAPSHOT_POLICY_INTERVAL,
-          TEST_SNAPSHOT_POLICY_MAX_SNAPS, TEST_SNAPSHOT_POLICY_DISPLAY);
+          TEST_SNAPSHOT_POLICY_MAX_SNAPS, TEST_SNAPSHOT_POLICY_ACCOUNT_ID, TEST_SNAPSHOT_POLICY_DOMAIN_ID, TEST_SNAPSHOT_POLICY_DISPLAY);
 
         apiDBUtilsMock = Mockito.mockStatic(ApiDBUtils.class);
     }
@@ -442,7 +444,7 @@ public class SnapshotManagerTest {
         Mockito.doReturn(true).when(taggedResourceServiceMock).deleteTags(any(), any(), any());
 
         SnapshotPolicyVO snapshotPolicyVo = new SnapshotPolicyVO(TEST_VOLUME_ID, TEST_SNAPSHOT_POLICY_SCHEDULE, TEST_SNAPSHOT_POLICY_TIMEZONE, TEST_SNAPSHOT_POLICY_INTERVAL,
-          TEST_SNAPSHOT_POLICY_MAX_SNAPS, TEST_SNAPSHOT_POLICY_DISPLAY);
+          TEST_SNAPSHOT_POLICY_MAX_SNAPS, TEST_SNAPSHOT_POLICY_ACCOUNT_ID, TEST_SNAPSHOT_POLICY_DOMAIN_ID, TEST_SNAPSHOT_POLICY_DISPLAY);
 
         _snapshotMgr.updateSnapshotPolicy(snapshotPolicyVo, TEST_SNAPSHOT_POLICY_SCHEDULE, TEST_SNAPSHOT_POLICY_TIMEZONE,
           TEST_SNAPSHOT_POLICY_INTERVAL, TEST_SNAPSHOT_POLICY_MAX_SNAPS, TEST_SNAPSHOT_POLICY_DISPLAY, TEST_SNAPSHOT_POLICY_ACTIVE, null, null);

server/src/test/java/com/cloud/vm/UserVmManagerImplTest.java

@@ -59,6 +59,7 @@ import java.util.Map;
 import java.util.TimeZone;
 import java.util.UUID;
 
+import com.cloud.storage.dao.SnapshotPolicyDao;
 import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.acl.SecurityChecker;
 import org.apache.cloudstack.api.ApiCommandResourceType;
@@ -79,6 +80,7 @@ import org.apache.cloudstack.api.command.user.volume.ResizeVolumeCmd;
 import org.apache.cloudstack.backup.BackupManager;
 import org.apache.cloudstack.backup.BackupVO;
 import org.apache.cloudstack.backup.dao.BackupDao;
+import org.apache.cloudstack.backup.dao.BackupScheduleDao;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
@@ -435,6 +437,13 @@ public class UserVmManagerImplTest {
     @Mock
     private UUIDManager uuidMgr;
 
+
+    @Mock
+    private SnapshotPolicyDao snapshotPolicyDao;
+
+    @Mock
+    private BackupScheduleDao backupScheduleDao;
+
     MockedStatic<UnmanagedVMsManager> unmanagedVMsManagerMockedStatic;
 
     private static final long vmId = 1l;

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

@@ -60,6 +60,8 @@ import com.cloud.user.User;
 import com.cloud.user.dao.AccountDao;
 import com.cloud.utils.DateUtil;
 import com.cloud.utils.Pair;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.fsm.NoTransitionException;
 import com.cloud.vm.VMInstanceDetailVO;
@@ -77,6 +79,7 @@ import org.apache.cloudstack.api.command.admin.backup.UpdateBackupOfferingCmd;
 import org.apache.cloudstack.api.command.user.backup.CreateBackupCmd;
 import org.apache.cloudstack.api.command.user.backup.CreateBackupScheduleCmd;
 import org.apache.cloudstack.api.command.user.backup.DeleteBackupScheduleCmd;
+import org.apache.cloudstack.api.command.user.backup.ListBackupScheduleCmd;
 import org.apache.cloudstack.api.response.BackupResponse;
 import org.apache.cloudstack.backup.dao.BackupDao;
 import org.apache.cloudstack.backup.dao.BackupDetailsDao;
@@ -1815,6 +1818,78 @@ public class BackupManagerTest {
         );
     }
 
+    @Test
+    public void testListBackupSchedulesAsRootAdmin() {
+        long vmId = 1L;
+        ListBackupScheduleCmd cmd = Mockito.mock(ListBackupScheduleCmd.class);
+        Mockito.when(cmd.getVmId()).thenReturn(vmId);
+        Mockito.when(cmd.getId()).thenReturn(1L);
+
+        // Mock VM for validation
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        Mockito.when(vmInstanceDao.findById(vmId)).thenReturn(vm);
+        Mockito.when(vm.getDataCenterId()).thenReturn(1L);
+        overrideBackupFrameworkConfigValue();
+        Mockito.doNothing().when(accountManager).checkAccess(Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any());
+
+        BackupScheduleVO schedule1 = Mockito.mock(BackupScheduleVO.class);
+        BackupScheduleVO schedule2 = Mockito.mock(BackupScheduleVO.class);
+        List<BackupScheduleVO> schedules = List.of(schedule1, schedule2);
+
+        SearchBuilder<BackupScheduleVO> searchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchCriteria<BackupScheduleVO> searchCriteria = Mockito.mock(SearchCriteria.class);
+        BackupScheduleVO entity = Mockito.mock(BackupScheduleVO.class);
+
+        Mockito.when(backupScheduleDao.createSearchBuilder()).thenReturn(searchBuilder);
+        Mockito.when(searchBuilder.entity()).thenReturn(entity);
+        Mockito.when(searchBuilder.and(Mockito.anyString(), (Object) any(), Mockito.any())).thenReturn(searchBuilder);
+        Mockito.when(searchBuilder.create()).thenReturn(searchCriteria);
+
+        Mockito.when(backupScheduleDao.searchAndCount(Mockito.any(), Mockito.any())).thenReturn(new Pair<>(schedules, schedules.size()));
+        List<BackupSchedule> result = backupManager.listBackupSchedules(cmd);
+
+        assertEquals(2, result.size());
+        assertTrue(result.contains(schedule1));
+        assertTrue(result.contains(schedule2));
+    }
+
+    @Test
+    public void testListBackupSchedulesAsNonAdmin() {
+        long vmId = 1L;
+        ListBackupScheduleCmd cmd = Mockito.mock(ListBackupScheduleCmd.class);
+        Mockito.when(cmd.getVmId()).thenReturn(vmId);
+        Mockito.when(cmd.getId()).thenReturn(1L);
+
+        // Mock VM for validation
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        Mockito.when(vmInstanceDao.findById(vmId)).thenReturn(vm);
+        Mockito.when(vm.getDataCenterId()).thenReturn(1L);
+        overrideBackupFrameworkConfigValue();
+        Mockito.doNothing().when(accountManager).checkAccess(Mockito.any(), Mockito.any(), Mockito.anyBoolean(), Mockito.any());
+
+        BackupScheduleVO schedule = Mockito.mock(BackupScheduleVO.class);
+        List<BackupScheduleVO> schedules = List.of(schedule);
+
+        SearchBuilder<BackupScheduleVO> searchBuilder = Mockito.mock(SearchBuilder.class);
+        SearchCriteria<BackupScheduleVO> searchCriteria = Mockito.mock(SearchCriteria.class);
+        BackupScheduleVO entity = Mockito.mock(BackupScheduleVO.class);
+
+        Mockito.when(backupScheduleDao.createSearchBuilder()).thenReturn(searchBuilder);
+        Mockito.when(searchBuilder.create()).thenReturn(searchCriteria);
+        Mockito.when(searchBuilder.entity()).thenReturn(entity);
+        Mockito.when(searchBuilder.and(Mockito.anyString(), (Object) any(), Mockito.any())).thenReturn(searchBuilder);
+        Mockito.lenient().when(backupScheduleDao.search(Mockito.eq(searchCriteria), Mockito.any())).thenReturn(schedules);
+
+        Mockito.doNothing().when(accountManager).buildACLSearchBuilder(Mockito.any(), Mockito.anyLong(), Mockito.anyBoolean(), Mockito.anyList(), Mockito.any());
+        Mockito.doNothing().when(accountManager).buildACLSearchCriteria(Mockito.any(), Mockito.anyLong(), Mockito.anyBoolean(), Mockito.anyList(), Mockito.any());
+
+        Mockito.when(backupScheduleDao.searchAndCount(Mockito.any(), Mockito.any())).thenReturn(new Pair<>(schedules, schedules.size()));
+        List<BackupSchedule> result = backupManager.listBackupSchedules(cmd);
+
+        assertEquals(1, result.size());
+        assertTrue(result.contains(schedule));
+    }
+
     @Test
     public void testCanCreateInstanceFromBackupAcrossZonesSuccess() {
         Long backupId = 1L;

ui/public/locales/en.json

@@ -78,6 +78,8 @@
 "label.action.copy.iso": "Copy ISO",
 "label.action.copy.snapshot": "Copy Snapshot",
 "label.action.copy.template": "Copy Template",
+"label.action.create.backup.schedule": "Create Backup Schedule",
+"label.action.create.recurring.snapshot": "Create Recurring Snapshot",
 "label.action.create.snapshot.from.vmsnapshot": "Create Snapshot from Instance Snapshot",
 "label.action.create.template.from.volume": "Create Template from volume",
 "label.action.create.volume": "Create Volume",
@@ -455,6 +457,7 @@
 "label.backup.restore": "Restore Instance backup",
 "label.backup.schedule.create.failed": "Failed to create Backup Schedule",
 "label.backuplimit": "Backup Limits",
+"label.backup.schedules": "Backup Schedules",
 "label.backup.storage": "Backup Storage",
 "label.backupstoragelimit": "Backup Storage Limits (GiB)",
 "label.backupofferingid": "Backup Offering ID",
@@ -749,6 +752,7 @@
 "label.delete.asnrange": "Delete AS Range",
 "label.delete.autoscale.vmgroup": "Delete AutoScaling Group",
 "label.delete.backup": "Delete backup",
+"label.delete.backup.schedule": "Delete backup schedule",
 "label.delete.bgp.peer": "Delete BGP peer",
 "label.delete.bigswitchbcf": "Remove BigSwitch BCF controller",
 "label.delete.brocadevcs": "Remove Brocade Vcs switch",
@@ -1510,7 +1514,7 @@
 "label.max.primary.storage": "Max. primary (GiB)",
 "label.max.secondary.storage": "Max. secondary (GiB)",
 "label.max.migrations": "Max. migrations",
-"label.maxbackup": "Max. Backups",
+"label.maxbackups": "Max. Backups",
 "label.maxbackupstorage": "Max. Backup Storage (GiB)",
 "label.maxbackups.to.retain": "Max. Backups to retain",
 "label.maxbucket": "Max. Buckets",
@@ -1536,6 +1540,7 @@
 "label.maxresolutiony": "Max. resolution Y",
 "label.maxsecondarystorage": "Max. secondary storage (GiB)",
 "label.maxsize": "Maximum size",
+"label.maxsnaps": "Max. Snapshots",
 "label.maxsnapshot": "Max. Snapshots",
 "label.maxtemplate": "Max. Templates",
 "label.maxuservm": "Max. User Instances",
@@ -2214,6 +2219,8 @@
 "label.select.root.disk": "Select the ROOT disk",
 "label.select.source.vcenter.datacenter": "Select the source VMware vCenter Datacenter",
 "label.select.tier": "Select Network Tier",
+"label.select.vm": "Select Instance",
+"label.select.volume": "Select Volume",
 "label.select.zones": "Select zones",
 "label.select.storagepools": "Select storage pools",
 "label.select.2fa.provider": "Select the provider",
@@ -2284,6 +2291,8 @@
 "label.snapshot.name": "Snapshot name",
 "label.snapshotlimit": "Snapshot limits",
 "label.snapshotmemory": "Snapshot memory",
+"label.snapshotpolicy": "Snapshot policy",
+"label.snapshotpolicies": "Snapshot policies",
 "label.snapshots": "Volume Snapshots",
 "label.snapshottype": "Snapshot Type",
 "label.sockettimeout": "Socket timeout",
@@ -2880,6 +2889,7 @@
 "message.action.delete.autoscale.vmgroup": "Please confirm that you want to delete this autoscaling group.",
 "message.action.delete.backup.offering": "Please confirm that you want to delete this backup offering?",
 "message.action.delete.backup.repository": "Please confirm that you want to delete this backup repository?",
+"message.action.delete.backup.schedule": "Please confirm that you want to delete this backup schedule?",
 "message.action.delete.cluster": "Please confirm that you want to delete this Cluster.",
 "message.action.delete.custom.action": "Please confirm that you want to delete this custom action.",
 "message.action.delete.domain": "Please confirm that you want to delete this domain.",
@@ -2905,6 +2915,7 @@
 "message.action.delete.secondary.storage": "Please confirm that you want to delete this secondary storage.",
 "message.action.delete.security.group": "Please confirm that you want to delete this security group.",
 "message.action.delete.snapshot": "Please confirm that you want to delete this Snapshot.",
+"message.action.delete.snapshot.policy": "Please confirm that you want to delete the selected Snapshot Policy.",
 "message.action.delete.template": "Please confirm that you want to delete this Template.",
 "message.action.delete.tungsten.router.table": "Please confirm that you want to remove Route Table from this Network?",
 "message.action.delete.vgpu.profile": "Please confirm that you want to delete this vGPU profile.",
@@ -3731,6 +3742,8 @@
 "message.select.security.groups": "Please select security group(s) for your new Instance.",
 "message.select.start.date.and.time": "Select a start date & time.",
 "message.select.temporary.storage.instance.conversion": "(Optional) Select a Storage temporary destination for the converted disks through virt-v2v",
+"message.select.volume.to.continue": "Please select a volume to continue.",
+"message.select.vm.to.continue": "Please select an Instance to continue.",
 "message.select.zone.description": "Select type of Zone basic/advanced.",
 "message.select.zone.hint": "This is the type of Zone deployment that you want to use. Basic zone: provides a single Network where each Instance is assigned an IP directly from the Network. Guest isolation can be provided through layer-3 means such as security groups (IP address source filtering). Advanced zone: For more sophisticated Network topologies. This Network model provides the most flexibility in defining guest Networks and providing custom Network offerings such as firewall, VPN, or load balancer support.",
 "message.server": "Server : ",

ui/src/components/view/DetailsTab.vue

@@ -161,6 +161,13 @@
           <div>{{ dataResource[item] }}</div>
         </div>
       </a-list-item>
+      <a-list-item v-else-if="(item === 'zoneid' && $route.path.includes('/snapshotpolicy'))">
+        <div>
+          <strong>{{ $t('label.' + String(item).toLowerCase()) }}</strong>
+          <br/>
+          <div>{{ dataResource[item] }}</div>
+        </div>
+      </a-list-item>
       <a-list-item v-else-if="['startdate', 'enddate'].includes(item)">
         <div>
           <strong>{{ $t('label.' + item.replace('date', '.date.and.time'))}}</strong>

ui/src/components/view/ListView.vue

@@ -233,11 +233,49 @@
         >{{ $t(text.toLowerCase()) }}</span>
         <span v-else>{{ text }}</span>
       </template>
-
       <template v-if="column.key === 'schedule'">
+        <div v-if="['/snapshotpolicy', '/backupschedule'].some(path => $route.path.endsWith(path))">
+          <label class="interval-content">
+            <span v-if="record.intervaltype===0 || record.intervaltype==='HOURLY'">{{ record.schedule + $t('label.min.past.hour') }}</span>
+            <span v-else>{{ record.schedule.split(':')[1] + ':' + record.schedule.split(':')[0] }}</span>
+          </label>
+          <span v-if="record.intervaltype===2 || record.intervaltype==='WEEKLY'">
+            {{ ` ${$t('label.every')} ${$t(listDayOfWeek[record.schedule.split(':')[2] - 1])}` }}
+          </span>
+          <span v-else-if="record.intervaltype===3 || record.intervaltype==='MONTHLY'">
+            {{ ` ${$t('label.day')} ${record.schedule.split(':')[2]} ${$t('label.of.month')}` }}
+          </span>
+        </div>
+        <div v-else>
           {{ text }}
           <br />
           ({{ generateHumanReadableSchedule(text) }})
+        </div>
+      </template>
+      <template v-if="column.key === 'intervaltype' && ['/snapshotpolicy', '/backupschedule'].some(path => $route.path.endsWith(path))">
+        <QuickView
+          style="margin-right: 8px"
+          :actions="actions"
+          :resource="record"
+          :enabled="quickViewEnabled() && actions.length > 0"
+          @exec-action="$parent.execAction"
+        />
+        <span v-if="record.intervaltype===0">
+          <clock-circle-outlined />
+        </span>
+        <span class="custom-icon icon-daily" v-else-if="record.intervaltype===1">
+          <calendar-outlined />
+        </span>
+        <span class="custom-icon icon-weekly" v-else-if="record.intervaltype===2">
+          <calendar-outlined />
+        </span>
+        <span class="custom-icon icon-monthly" v-else-if="record.intervaltype===3">
+          <calendar-outlined />
+        </span>
+        {{ getIntervalTypeText(record.intervaltype) }}
+      </template>
+      <template v-if="column.key === 'timezone'">
+        <label>{{ getTimeZone(record.timezone) }}</label>
       </template>
       <template v-if="column.key === 'displayname'">
         <QuickView
@@ -1011,6 +1049,7 @@ import { createPathBasedOnVmType } from '@/utils/plugins'
 import { validateLinks } from '@/utils/links'
 import cronstrue from 'cronstrue/i18n'
 import moment from 'moment-timezone'
+import { timeZoneName } from '@/utils/timezone'
 import { FileTextOutlined } from '@ant-design/icons-vue'
 
 export default {
@@ -1111,7 +1150,8 @@ export default {
         }
       },
       usageTypeMap: {},
-      resourceIdToValidLinksMap: {}
+      resourceIdToValidLinksMap: {},
+      listDayOfWeek: ['sunday', 'monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday']
     }
   },
   watch: {
@@ -1148,7 +1188,7 @@ export default {
         '/zone', '/pod', '/cluster', '/host', '/storagepool', '/imagestore', '/systemvm', '/router', '/ilbvm', '/annotation',
         '/computeoffering', '/systemoffering', '/diskoffering', '/backupoffering', '/networkoffering', '/vpcoffering',
         '/tungstenfabric', '/oauthsetting', '/guestos', '/guestoshypervisormapping', '/webhook', 'webhookdeliveries', '/quotatariff', '/sharedfs',
-        '/ipv4subnets', '/managementserver', '/gpucard', '/gpudevices', '/vgpuprofile', '/extension'].join('|'))
+        '/ipv4subnets', '/managementserver', '/gpucard', '/gpudevices', '/vgpuprofile', '/extension', '/snapshotpolicy', '/backupschedule'].join('|'))
         .test(this.$route.path)
     },
     enableGroupAction () {
@@ -1162,6 +1202,13 @@ export default {
     getDateAtTimeZone (date, timezone) {
       return date ? moment(date).tz(timezone).format('YYYY-MM-DD HH:mm:ss') : null
     },
+    getIntervalTypeText (intervaltype) {
+      const types = { 0: 'HOURLY', 1: 'DAILY', 2: 'WEEKLY', 3: 'MONTHLY' }
+      return types[intervaltype] || intervaltype
+    },
+    getTimeZone (timeZone) {
+      return timeZoneName(timeZone)
+    },
     fetchColumns () {
       if (this.isOrderUpdatable()) {
         return this.columns
@@ -1563,4 +1610,31 @@ export default {
     color: #f50000;
     padding: 10%;
   }
+
+  .custom-icon:before {
+      font-size: 8px;
+      position: absolute;
+      top: 8px;
+      left: 3.5px;
+      color: #000;
+      font-weight: 700;
+      line-height: 1.7;
+    }
+
+    .icon-daily:before {
+      content: "01";
+      left: 5px;
+      top: 7px;
+      line-height: 1.9;
+    }
+
+    .icon-weekly:before {
+      content: "1-7";
+      left: 3px;
+      line-height: 1.7;
+    }
+
+    .icon-monthly:before {
+      content: "***";
+    }
 </style>

ui/src/components/view/SearchView.vue

@@ -314,7 +314,7 @@ export default {
         if (item === 'usagetype' && !('listUsageTypes' in this.$store.getters.apis)) {
           return true
         }
-        if (item === 'isencrypted' && !('listVolumes' in this.$store.getters.apis)) {
+        if (['isencrypted', 'volumeid'].includes(item) && !('listVolumes' in this.$store.getters.apis)) {
           return true
         }
         if (item === 'backupofferingid' && !('listBackupOfferings' in this.$store.getters.apis)) {
@@ -325,7 +325,7 @@ export default {
           'type', 'scope', 'managementserverid', 'serviceofferingid',
           'diskofferingid', 'networkid', 'usagetype', 'restartrequired', 'gpuenabled',
           'displaynetwork', 'guestiptype', 'usersource', 'arch', 'oscategoryid', 'templatetype', 'gpucardid', 'vgpuprofileid',
-          'extensionid', 'backupoffering'].includes(item)
+          'extensionid', 'backupoffering', 'volumeid', 'virtualmachineid'].includes(item)
         ) {
           type = 'list'
         } else if (item === 'tags') {
@@ -510,6 +510,7 @@ export default {
       let networkIndex = -1
       let usageTypeIndex = -1
       let volumeIndex = -1
+      let virtualmachineIndex = -1
       let backupOfferingIndex = -1
       let osCategoryIndex = -1
       let gpuCardIndex = -1
@@ -588,6 +589,12 @@ export default {
         promises.push(await this.fetchInstanceGroups(searchKeyword))
       }
 
+      if (arrayField.includes('virtualmachineid')) {
+        virtualmachineIndex = this.fields.findIndex(item => item.name === 'virtualmachineid')
+        this.fields[virtualmachineIndex].loading = true
+        promises.push(await this.fetchVirtualMachines(searchKeyword))
+      }
+
       if (arrayField.includes('managementserverid')) {
         managementServerIdIndex = this.fields.findIndex(item => item.name === 'managementserverid')
         this.fields[managementServerIdIndex].loading = true
@@ -648,6 +655,12 @@ export default {
         promises.push(await this.fetchVgpuProfiles(searchKeyword))
       }
 
+      if (arrayField.includes('volumeid')) {
+        volumeIndex = this.fields.findIndex(item => item.name === 'volumeid')
+        this.fields[volumeIndex].loading = true
+        promises.push(await this.fetchVolumes(searchKeyword))
+      }
+
       Promise.all(promises).then(response => {
         if (typeIndex > -1) {
           const types = response.filter(item => item.type === 'type')
@@ -778,6 +791,20 @@ export default {
             this.fields[vgpuProfileIndex].opts = this.sortArray(vgpuProfiles[0].data)
           }
         }
+
+        if (volumeIndex > -1) {
+          const volumes = response.filter(item => ['volumeid', 'isencrypted'].includes(item.type))
+          if (volumes && volumes.length > 0) {
+            this.fields[volumeIndex].opts = this.sortArray(volumes[0].data)
+          }
+        }
+
+        if (virtualmachineIndex > -1) {
+          const virtualMachines = response.filter(item => item.type === 'virtualmachineid')
+          if (virtualMachines && virtualMachines.length > 0) {
+            this.fields[virtualmachineIndex].opts = this.sortArray(virtualMachines[0].data)
+          }
+        }
       }).finally(() => {
         if (typeIndex > -1) {
           this.fields[typeIndex].loading = false
@@ -839,6 +866,12 @@ export default {
         if (vgpuProfileIndex > -1) {
           this.fields[vgpuProfileIndex].loading = false
         }
+        if (volumeIndex > -1) {
+          this.fields[volumeIndex].loading = false
+        }
+        if (virtualmachineIndex > -1) {
+          this.fields[virtualmachineIndex].loading = false
+        }
         if (Array.isArray(arrayField)) {
           this.fillFormFieldValues()
         }
@@ -1123,6 +1156,19 @@ export default {
         })
       })
     },
+    fetchVirtualMachines (searchKeyword) {
+      return new Promise((resolve, reject) => {
+        getAPI('listVirtualMachines', { listAll: true, keyword: searchKeyword }).then(json => {
+          const virtualMachines = json.listvirtualmachinesresponse.virtualmachine
+          resolve({
+            type: 'virtualmachineid',
+            data: virtualMachines
+          })
+        }).catch(error => {
+          reject(error.response.headers['x-description'])
+        })
+      })
+    },
     fetchManagementServers (searchKeyword) {
       return new Promise((resolve, reject) => {
         getAPI('listManagementServers', { listAll: true, keyword: searchKeyword }).then(json => {

ui/src/config/section/storage.js

@@ -413,6 +413,51 @@ export default {
         }
       ]
     },
+    {
+      name: 'snapshotpolicy',
+      title: 'label.snapshotpolicies',
+      icon: 'build-outlined',
+      docHelp: 'adminguide/storage.html#working-with-volume-snapshots',
+      permission: ['listSnapshotPolicies'],
+      resourceType: 'SnapshotPolicy',
+      params: { listall: true },
+      columns: () => {
+        var fields = ['intervaltype', 'maxsnaps', 'schedule', 'timezone', 'volumename']
+        return fields
+      },
+      searchFilters: ['volumeid'],
+      actions: [
+        {
+          api: 'createSnapshotPolicy',
+          icon: 'plus-outlined',
+          docHelp: 'adminguide/storage.html#working-with-volume-snapshots',
+          label: 'label.action.create.recurring.snapshot',
+          listView: true,
+          show: () => { return 'createSnapshotPolicy' in store.getters.apis },
+          popup: true,
+          component: shallowRef(defineAsyncComponent(() => import('@/views/storage/RecurringSnapshotVolume.vue'))),
+          mapping: {
+            intervaltype: {
+              options: ['HOURLY', 'DAILY', 'WEEKLY', 'MONTHLY']
+            }
+          }
+        },
+        {
+          api: 'deleteSnapshotPolicies',
+          icon: 'delete-outlined',
+          label: 'label.delete.snapshot.policy',
+          message: 'message.action.delete.snapshot.policy',
+          dataView: true,
+          show: (record) => true,
+          args: ['id'],
+          mapping: {
+            id: {
+              value: (record) => record.id
+            }
+          }
+        }
+      ]
+    },
     {
       name: 'backup',
       title: 'label.backups',
@@ -497,6 +542,51 @@ export default {
         }
       ]
     },
+    {
+      name: 'backupschedule',
+      title: 'label.backup.schedules',
+      icon: 'build-outlined',
+      docHelp: 'adminguide/storage.html#working-with-volume-snapshots',
+      permission: ['listBackupSchedule'],
+      resourceType: 'backupSchedule',
+      params: { listall: true },
+      columns: () => {
+        var fields = ['intervaltype', 'maxbackups', 'schedule', 'timezone', 'virtualmachinename']
+        return fields
+      },
+      searchFilters: ['virtualmachineid'],
+      actions: [
+        {
+          api: 'createBackupSchedule',
+          icon: 'plus-outlined',
+          docHelp: 'adminguide/storage.html#working-with-volume-snapshots',
+          label: 'label.action.create.backup.schedule',
+          listView: true,
+          show: () => { return 'createBackupSchedule' in store.getters.apis },
+          popup: true,
+          component: shallowRef(defineAsyncComponent(() => import('@/views/compute/backup/CreateBackupSchedule.vue'))),
+          mapping: {
+            intervaltype: {
+              options: ['HOURLY', 'DAILY', 'WEEKLY', 'MONTHLY']
+            }
+          }
+        },
+        {
+          api: 'deleteBackupSchedule',
+          icon: 'delete-outlined',
+          label: 'label.delete.backup.schedule',
+          message: 'message.action.delete.backup.schedule',
+          dataView: true,
+          show: (record) => true,
+          args: ['id'],
+          mapping: {
+            id: {
+              value: (record) => record.id
+            }
+          }
+        }
+      ]
+    },
     {
       name: 'buckets',
       title: 'label.buckets',

ui/src/views/compute/BackupScheduleWizard.vue

@@ -21,12 +21,18 @@
       <a-tab-pane :tab="$t('label.schedule')" key="1">
         <FormSchedule
           :loading="loading"
-          :resource="resource"/>
+          :resource="resource"
+          :dataSource="dataSource"
+          @close-action="closeAction"
+          @refresh="handleRefresh"/>
       </a-tab-pane>
       <a-tab-pane :tab="$t('label.scheduled.backups')" key="2">
         <BackupSchedule
           :loading="loading"
-          :dataSource="dataSource" />
+          :resource="resource"
+          :dataSource="dataSource"
+          @refresh="handleRefresh"
+          @close-action="closeAction" />
       </a-tab-pane>
     </a-tabs>
   </div>
@@ -52,7 +58,7 @@ export default {
   data () {
     return {
       loading: false,
-      dataSource: {}
+      dataSource: []
     }
   },
   provide () {
@@ -67,16 +73,28 @@ export default {
   methods: {
     fetchData () {
       const params = {}
-      this.dataSource = {}
+      this.dataSource = []
       this.loading = true
-      params.virtualmachineid = this.resource.id
+      params.virtualmachineid = this.resource.id || this.resource.virtualmachineid
+
+      if (!params.virtualmachineid) {
+        console.error('No VM ID found in resource:', this.resource)
+        this.loading = false
+        return
+      }
+
       getAPI('listBackupSchedule', params).then(json => {
-        this.dataSource = json.listbackupscheduleresponse.backupschedule || {}
+        this.dataSource = json.listbackupscheduleresponse.backupschedule || []
       }).finally(() => {
         this.loading = false
       })
     },
+    handleRefresh () {
+      this.fetchData()
+      this.$emit('refresh')
+    },
     closeAction () {
+      this.$emit('refresh')
       this.$emit('close-action')
     }
   }

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

@@ -0,0 +1,183 @@
+  // 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.
+
+  <template>
+  <div class="create-backup-schedule-layout">
+    <div v-if="!isVMResource" class="vm-selection">
+      <a-form layout="vertical">
+        <a-form-item :label="$t('label.virtualmachine')" required>
+          <a-select
+            v-model:value="selectedVMId"
+            :placeholder="$t('label.select.vm')"
+            :loading="vmsLoading"
+            show-search
+            :filter-option="filterOption"
+            @change="onVMChange"
+          >
+            <a-select-option
+              v-for="vm in vms"
+              :key="vm.id"
+              :value="vm.id"
+            >
+              {{ vm.name }} ({{ vm.account }})
+            </a-select-option>
+          </a-select>
+        </a-form-item>
+      </a-form>
+    </div>
+    <div v-if="currentVMResource && currentVMResource.id">
+      <BackupScheduleWizard
+        ref="backupScheduleWizard"
+        :resource="currentVMResource"
+        @close-action="closeAction"
+        @refresh="handleRefresh"
+      />
+    </div>
+    <div v-if="!currentVMResource || !currentVMResource.id" class="no-vm-selected">
+      <div class="empty-state">
+        <p>{{ $t('message.select.vm.to.continue') }}</p>
+      </div>
+    </div>
+  </div>
+  </template>
+
+<script>
+import { getAPI } from '@/api'
+import BackupScheduleWizard from '@/views/compute/BackupScheduleWizard'
+
+export default {
+  name: 'CreateBackupSchedule',
+  components: {
+    BackupScheduleWizard
+  },
+  props: {
+    resource: {
+      type: Object,
+      required: false,
+      default: () => null
+    }
+  },
+  inject: ['parentFetchData'],
+  data () {
+    return {
+      vms: [],
+      vmsLoading: false,
+      selectedVMId: null,
+      selectedVM: null
+    }
+  },
+  computed: {
+    resourceType () {
+      if (!this.resource) return 'none'
+
+      if (this.resource.vmstate !== undefined ||
+        this.resource.guestosid !== undefined ||
+        this.resource.hypervisor !== undefined ||
+        this.resource.backupofferingid !== undefined ||
+        this.resource.serviceofferingid !== undefined) {
+        return 'vm'
+      }
+      if (this.resource.intervaltype !== undefined &&
+          this.resource.schedule !== undefined) {
+        return 'backupschedule'
+      }
+
+      if (this.resource.virtualmachineid !== undefined) {
+        return 'backupschedule'
+      }
+
+      return 'unknown'
+    },
+    isVMResource () {
+      return this.resourceType === 'vm'
+    },
+    currentVMResource () {
+      if (this.isVMResource) {
+        return this.resource
+      } else {
+        return this.selectedVM
+      }
+    }
+  },
+  created () {
+    this.fetchVMs()
+  },
+  methods: {
+    async fetchVMs () {
+      this.vmsLoading = true
+      try {
+        const response = await getAPI('listVirtualMachines', { listAll: true })
+        const vms = response.listvirtualmachinesresponse.virtualmachine || []
+        this.vms = vms.filter(vm => {
+          return vm.backupofferingid && ['Running', 'Stopped'].includes(vm.state)
+        })
+      } catch (error) {
+        this.$message.error(this.$t('message.error.fetch.vms'))
+        console.error('Error fetching VMs:', error)
+      } finally {
+        this.vmsLoading = false
+      }
+    },
+    onVMChange (vmId) {
+      const vm = this.vms.find(v => v.id === vmId)
+      if (vm) {
+        this.selectedVM = vm
+        this.selectedVMId = vmId
+      }
+    },
+    closeAction () {
+      this.$emit('refresh')
+      this.$emit('close-action')
+    },
+    handleRefresh () {
+      this.$emit('refresh')
+      this.parentFetchData()
+    },
+    filterOption (input, option) {
+      return option.children[0].children.toLowerCase().indexOf(input.toLowerCase()) >= 0
+    }
+  }
+}
+</script>
+
+<style lang="less" scoped>
+.create-backup-schedule-layout {
+  .vm-selection {
+    margin-bottom: 20px;
+    padding: 16px;
+    border: 1px solid #d9d9d9;
+    border-radius: 6px;
+    background-color: #fafafa;
+    .ant-form-item {
+      margin-bottom: 0;
+      .ant-select {
+        width: 100%;
+        min-width: 400px;
+      }
+    }
+  }
+
+  .current-vm-info {
+    margin-bottom: 16px;
+  }
+
+  .no-vm-selected {
+    text-align: center;
+    padding: 40px 20px;
+  }
+}
+</style>

ui/src/views/storage/RecurringSnapshotVolume.vue

@@ -17,11 +17,33 @@
 
 <template>
   <div class="snapshot-layout">
-    <a-tabs defaultActiveKey="1" :animated="false">
+    <div v-if="!isVolumeResource" class="volume-selection">
+      <a-form layout="vertical">
+        <a-form-item :label="$t('label.volume')" required>
+          <a-select
+            v-model:value="selectedVolumeId"
+            :placeholder="$t('label.select.volume')"
+            :loading="volumesLoading"
+            show-search
+            :filter-option="filterOption"
+            @change="onVolumeChange"
+          >
+            <a-select-option
+              v-for="volume in volumes"
+              :key="volume.id"
+              :value="volume.id"
+            >
+              {{ volume.name }} ({{ volume.sizegb }}GB)
+            </a-select-option>
+          </a-select>
+        </a-form-item>
+      </a-form>
+    </div>
+    <a-tabs v-if="currentVolumeResource && currentVolumeResource.id" defaultActiveKey="1" :animated="false">
       <a-tab-pane :tab="$t('label.schedule')" key="1">
         <FormSchedule
           :loading="loading"
-          :resource="resource"
+          :resource="currentVolumeResource"
           :dataSource="dataSource"
           :resourceType="'Volume'"
           @close-action="closeAction"
@@ -30,12 +52,17 @@
       <a-tab-pane :tab="$t('label.action.recurring.snapshot')" key="2">
         <ScheduledSnapshots
           :loading="loading"
-          :resource="resource"
+          :resource="currentVolumeResource"
           :dataSource="dataSource"
           @refresh="handleRefresh"
           @close-action="closeAction"/>
       </a-tab-pane>
     </a-tabs>
+    <div v-if="!currentVolumeResource || !currentVolumeResource.id" class="no-volume-selected">
+      <div class="empty-state">
+        <p>{{ $t('message.select.volume.to.continue') }}</p>
+      </div>
+    </div>
   </div>
 </template>
 
@@ -53,44 +80,150 @@ export default {
   props: {
     resource: {
       type: Object,
-      required: true
+      required: false,
+      default: () => null
     }
   },
+  inject: ['parentFetchData'],
   data () {
     return {
       loading: false,
-      dataSource: []
+      dataSource: [],
+      volumes: [],
+      volumesLoading: false,
+      selectedVolumeId: null,
+      selectedVolume: null
+    }
+  },
+  computed: {
+    resourceType () {
+      if (!this.resource) return 'none'
+      if (this.resource.type === 'ROOT' || this.resource.type === 'DATADISK' ||
+          this.resource.state === 'Ready' || this.resource.state === 'Allocated' ||
+          this.resource.sizegb !== undefined) {
+        return 'volume'
+      }
+      if (this.resource.intervaltype !== undefined || this.resource.schedule !== undefined) {
+        return 'snapshotpolicy'
+      }
+
+      return 'unknown'
+    },
+
+    isVolumeResource () {
+      return this.resourceType === 'volume'
+    },
+
+    currentVolumeResource () {
+      if (this.isVolumeResource) {
+        return this.resource
+      } else {
+        return this.selectedVolume
+      }
     }
   },
   created () {
+    if (this.isVolumeResource) {
       this.fetchData()
+    } else {
+      this.fetchVolumes()
+    }
   },
   methods: {
+    async fetchVolumes () {
+      this.volumesLoading = true
+      try {
+        const response = await getAPI('listVolumes', { listAll: true })
+        const volumes = response.listvolumesresponse.volume || []
+        this.volumes = volumes.filter(volume => {
+          return volume.state === 'Ready' &&
+                (volume.hypervisor !== 'KVM' ||
+                  (['Stopped', 'Destroyed'].includes(volume.vmstate)) ||
+                  (this.$store.getters.features.kvmsnapshotenabled))
+        })
+      } catch (error) {
+        this.$message.error(this.$t('message.error.fetch.volumes'))
+        console.error('Error fetching volumes:', error)
+      } finally {
+        this.volumesLoading = false
+      }
+    },
+    onVolumeChange (volumeId) {
+      const volume = this.volumes.find(v => v.id === volumeId)
+      if (volume) {
+        this.selectedVolume = volume
+        this.selectedVolumeId = volumeId
+        this.dataSource = []
+        this.fetchData()
+      }
+    },
     fetchData () {
-      const params = {}
+      const volumeResource = this.currentVolumeResource
+      if (!volumeResource || !volumeResource.id) {
+        return
+      }
+
+      const params = {
+        volumeid: volumeResource.id,
+        listAll: true
+      }
+
       this.dataSource = []
       this.loading = true
-      params.volumeid = this.resource.id
+
       getAPI('listSnapshotPolicies', params).then(json => {
         this.loading = false
         const listSnapshotPolicies = json.listsnapshotpoliciesresponse.snapshotpolicy
         if (listSnapshotPolicies && listSnapshotPolicies.length > 0) {
           this.dataSource = listSnapshotPolicies
         }
+      }).catch(error => {
+        this.loading = false
+        this.$message.error(this.$t('message.error.fetch.snapshot.policies'))
+        console.error('Error fetching snapshot policies:', error)
       })
     },
     handleRefresh () {
       this.fetchData()
+      this.parentFetchData()
     },
     closeAction () {
+      this.fetchData()
       this.$emit('close-action')
+    },
+    filterOption (input, option) {
+      return option.children[0].children.toLowerCase().indexOf(input.toLowerCase()) >= 0
     }
   }
 }
 </script>
 
 <style lang="less" scoped>
-  .snapshot-layout {
+.snapshot-layout {
-    max-width: 600px;
+  max-width: 800px;
+
+  .volume-selection {
+    margin-bottom: 20px;
+    padding: 16px;
+    border: 1px solid #d9d9d9;
+    border-radius: 6px;
+    background-color: #fafafa;
+    .ant-form-item {
+      margin-bottom: 0;
+      .ant-select {
+        width: 100%;
+        min-width: 400px;
       }
+    }
+  }
+
+  .current-volume-info {
+    margin-bottom: 16px;
+  }
+
+  .no-volume-selected {
+    text-align: center;
+    padding: 40px 20px;
+  }
+}
 </style>