Create API to reassign volume (#6938)

api/src/main/java/com/cloud/storage/VolumeApiService.java

@@ -21,6 +21,7 @@ package com.cloud.storage;
 import java.net.MalformedURLException;
 import java.util.Map;
 
+import org.apache.cloudstack.api.command.user.volume.AssignVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.AttachVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.ChangeOfferingForVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
@@ -119,6 +120,8 @@ public interface VolumeApiService {
      */
     String extractVolume(ExtractVolumeCmd cmd);
 
+    Volume assignVolumeToAccount(AssignVolumeCmd cmd) throws ResourceAllocationException;
+
     boolean isDisplayResourceEnabled(Long id);
 
     void updateDisplay(Volume volume, Boolean displayVolume);

api/src/main/java/org/apache/cloudstack/api/command/user/volume/AssignVolumeCmd.java

@@ -0,0 +1,119 @@
+// 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.volume;
+
+import com.cloud.exception.ResourceAllocationException;
+import com.cloud.user.Account;
+import com.cloud.utils.exception.CloudRuntimeException;
+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.Parameter;
+import org.apache.cloudstack.api.ServerApiException;
+import org.apache.cloudstack.api.command.user.UserCmd;
+import org.apache.cloudstack.api.response.AccountResponse;
+import org.apache.cloudstack.api.response.ProjectResponse;
+import org.apache.cloudstack.api.response.VolumeResponse;
+import org.apache.log4j.Logger;
+
+import com.cloud.storage.Volume;
+
+import java.util.Map;
+
+@APICommand(name = AssignVolumeCmd.CMD_NAME, responseObject = VolumeResponse.class, description = "Changes ownership of a Volume from one account to another.", entityType = {
+        Volume.class}, requestHasSensitiveInfo = false, responseHasSensitiveInfo = false, since = "4.18.0.0")
+public class AssignVolumeCmd extends BaseCmd implements UserCmd {
+    public static final Logger LOGGER = Logger.getLogger(AssignVolumeCmd.class.getName());
+    public static final String CMD_NAME = "assignVolume";
+
+    /////////////////////////////////////////////////////
+    //////////////// API parameters /////////////////////
+    /////////////////////////////////////////////////////
+
+    @Parameter(name = ApiConstants.VOLUME_ID, type = CommandType.UUID, entityType = VolumeResponse.class, required = true, description = "The ID of the volume to be reassigned.")
+    private Long volumeId;
+
+    @Parameter(name = ApiConstants.ACCOUNT_ID, type = CommandType.UUID, entityType = AccountResponse.class,
+            description = "The ID of the account to which the volume will be assigned. Mutually exclusive with parameter 'projectid'.")
+    private Long accountId;
+
+    @Parameter(name = ApiConstants.PROJECT_ID, type = CommandType.UUID, entityType = ProjectResponse.class,
+            description = "The ID of the project to which the volume will be assigned. Mutually exclusive with 'accountid'.")
+    private Long projectid;
+
+    /////////////////////////////////////////////////////
+    /////////////////// Accessors ///////////////////////
+    /////////////////////////////////////////////////////
+
+    public Long getVolumeId() {
+        return volumeId;
+    }
+
+    public Long getAccountId() {
+        return accountId;
+    }
+
+    public Long getProjectid() {
+        return projectid;
+    }
+
+    /////////////////////////////////////////////////////
+    /////////////// API Implementation///////////////////
+    /////////////////////////////////////////////////////
+
+    @Override
+    public void execute() {
+        try {
+            Volume result = _volumeService.assignVolumeToAccount(this);
+            if (result == null) {
+                Map<String,String> fullParams = getFullUrlParams();
+                if (accountId != null) {
+                    throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, String.format("Failed to move volume [%s] to account [%s].", fullParams.get(ApiConstants.VOLUME_ID),
+                            fullParams.get(ApiConstants.ACCOUNT_ID)));
+                }
+                throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, String.format("Failed to move volume [%s] to project [%s].", fullParams.get(ApiConstants.VOLUME_ID),
+                        fullParams.get(ApiConstants.PROJECT_ID)));
+            }
+
+            VolumeResponse response = _responseGenerator.createVolumeResponse(getResponseView(), result);
+            response.setResponseName(getCommandName());
+            setResponseObject(response);
+
+        } catch (CloudRuntimeException | ResourceAllocationException e) {
+            String msg = String.format("Assign volume command for volume [%s] failed due to [%s].", getFullUrlParams().get("volumeid"), e.getMessage());
+            LOGGER.error(msg, e);
+            throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, msg);
+        }
+    }
+
+    @Override
+    public String getCommandName() {
+        return CMD_NAME.toLowerCase() + RESPONSE_SUFFIX;
+    }
+
+    @Override
+    public long getEntityOwnerId() {
+        Volume volume = _responseGenerator.findVolumeById(getVolumeId());
+
+        if (volume != null) {
+            return volume.getAccountId();
+        }
+
+        return Account.ACCOUNT_ID_SYSTEM;
+    }
+}

core/src/main/java/org/apache/cloudstack/storage/command/MoveVolumeCommand.java

@@ -0,0 +1,66 @@
+//
+// 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.storage.command;
+
+import com.cloud.agent.api.Command;
+
+public class MoveVolumeCommand extends Command {
+
+     private String srcPath;
+     private String destPath;
+     private String volumeName;
+     private String volumeUuid;
+     private String datastoreUri;
+
+    public MoveVolumeCommand(String volumeUuid, String volumeName, String destPath, String srcPath, String datastoreUri) {
+         this.volumeName = volumeName;
+         this.volumeUuid = volumeUuid;
+         this.srcPath = srcPath;
+         this.destPath = destPath;
+         this.datastoreUri = datastoreUri;
+     }
+
+     public String getSrcPath() {
+         return srcPath;
+     }
+
+     public String getDestPath() {
+         return destPath;
+     }
+
+     public String getVolumeName() {
+         return volumeName;
+     }
+
+     public String getVolumeUuid() {
+         return volumeUuid;
+     }
+
+    public String getDatastoreUri() {
+        return datastoreUri;
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return false;
+    }
+
+ }
+

engine/api/src/main/java/org/apache/cloudstack/engine/subsystem/api/storage/EndPointSelector.java

@@ -46,4 +46,6 @@ public interface EndPointSelector {
     EndPoint select(Scope scope, Long storeId);
 
     EndPoint select(DataStore store, String downloadUrl);
+
+    EndPoint findSsvm(long dcId);
 }

engine/api/src/main/java/org/apache/cloudstack/engine/subsystem/api/storage/VolumeService.java

@@ -30,6 +30,8 @@ import com.cloud.exception.StorageAccessException;
 import com.cloud.host.Host;
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.offering.DiskOffering;
+import com.cloud.storage.Volume;
+import com.cloud.user.Account;
 import com.cloud.utils.Pair;
 
 public interface VolumeService {
@@ -108,4 +110,6 @@ public interface VolumeService {
      */
     boolean copyPoliciesBetweenVolumesAndDestroySourceVolumeAfterMigration(ObjectInDataStoreStateMachine.Event destinationEvent, Answer destinationEventAnswer,
       VolumeInfo sourceVolume, VolumeInfo destinationVolume, boolean retryExpungeVolumeAsync);
+
+    void moveVolumeOnSecondaryStorageToAnotherAccount(Volume volume, Account sourceAccount, Account destAccount);
 }

engine/schema/src/main/resources/META-INF/db/schema-41720to41800.sql

@@ -1063,3 +1063,8 @@ CREATE TABLE IF NOT EXISTS `cloud`.`console_session` (
     CONSTRAINT `fk_consolesession__host_id` FOREIGN KEY(`host_id`) REFERENCES `cloud`.`host`(`id`),
     CONSTRAINT `uc_consolesession__uuid` UNIQUE (`uuid`)
 );
+
+-- Add assignVolume API permission to default resource admin and domain admin
+INSERT INTO `cloud`.`role_permissions` (`uuid`, `role_id`, `rule`, `permission`) VALUES (UUID(), 2, 'assignVolume', 'ALLOW');
+INSERT INTO `cloud`.`role_permissions` (`uuid`, `role_id`, `rule`, `permission`) VALUES (UUID(), 3, 'assignVolume', 'ALLOW');
+

engine/storage/src/main/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java

@@ -330,9 +330,15 @@ public class DefaultEndPointSelector implements EndPointSelector {
         if (storeScope.getScopeType() == ScopeType.ZONE) {
             dcId = storeScope.getScopeId();
         }
-        // find ssvm that can be used to download data to store. For zone-wide
+
-        // image store, use SSVM for that zone. For region-wide store,
+        return findSsvm(dcId);
-        // we can arbitrarily pick one ssvm to do that task
+    }
+
+    /**
+     * Finds an SSVM that can be used to execute a command.
+     * For zone-wide image store, use SSVM for that zone. For region-wide store, we can arbitrarily pick one SSVM to do the task.
+     * */
+    public EndPoint findSsvm(long dcId) {
         List<HostVO> ssAHosts = listUpAndConnectingSecondaryStorageVmHost(dcId);
         if (ssAHosts == null || ssAHosts.isEmpty()) {
             return null;

engine/storage/volume/src/main/java/org/apache/cloudstack/storage/volume/VolumeServiceImpl.java

@@ -18,6 +18,9 @@
  */
 package org.apache.cloudstack.storage.volume;
 
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
@@ -25,12 +28,14 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.ExecutionException;
 
 import javax.inject.Inject;
 
 import org.apache.cloudstack.secret.dao.PassphraseDao;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.dao.VMTemplateDao;
+import com.cloud.storage.resource.StorageProcessor;
 import org.apache.cloudstack.annotation.AnnotationService;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
 import org.apache.cloudstack.engine.cloud.entity.api.VolumeEntity;
@@ -65,6 +70,7 @@ import org.apache.cloudstack.storage.RemoteHostEndPoint;
 import org.apache.cloudstack.storage.command.CommandResult;
 import org.apache.cloudstack.storage.command.CopyCmdAnswer;
 import org.apache.cloudstack.storage.command.DeleteCommand;
+import org.apache.cloudstack.storage.command.MoveVolumeCommand;
 import org.apache.cloudstack.storage.datastore.PrimaryDataStoreProviderManager;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.SnapshotDataStoreDao;
@@ -125,7 +131,9 @@ import com.cloud.storage.dao.VolumeDao;
 import com.cloud.storage.dao.VolumeDetailsDao;
 import com.cloud.storage.snapshot.SnapshotApiService;
 import com.cloud.storage.snapshot.SnapshotManager;
+import com.cloud.storage.template.TemplateConstants;
 import com.cloud.storage.template.TemplateProp;
+import com.cloud.user.Account;
 import com.cloud.user.AccountManager;
 import com.cloud.user.ResourceLimitService;
 import com.cloud.utils.NumbersUtil;
@@ -136,9 +144,6 @@ import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.vm.VirtualMachine;
 import org.apache.commons.lang3.StringUtils;
 
-import static com.cloud.storage.resource.StorageProcessor.REQUEST_TEMPLATE_RELOAD;
-import java.util.concurrent.ExecutionException;
-
 @Component
 public class VolumeServiceImpl implements VolumeService {
     private static final Logger s_logger = Logger.getLogger(VolumeServiceImpl.class);
@@ -806,7 +811,7 @@ public class VolumeServiceImpl implements VolumeService {
             // hack for Vmware: host is down, previously download template to the host needs to be re-downloaded, so we need to reset
             // template_spool_ref entry here to NOT_DOWNLOADED and Allocated state
             Answer ans = result.getAnswer();
-            if (ans != null && ans instanceof CopyCmdAnswer && ans.getDetails().contains(REQUEST_TEMPLATE_RELOAD)) {
+            if (ans instanceof CopyCmdAnswer && ans.getDetails().contains(StorageProcessor.REQUEST_TEMPLATE_RELOAD)) {
                 if (tmplOnPrimary != null) {
                     s_logger.info("Reset template_spool_ref entry so that vmware template can be reloaded in next try");
                     VMTemplateStoragePoolVO templatePoolRef = _tmpltPoolDao.findByPoolTemplate(tmplOnPrimary.getDataStore().getId(), tmplOnPrimary.getId(), deployAsIsConfiguration);
@@ -2751,4 +2756,50 @@ public class VolumeServiceImpl implements VolumeService {
             volDao.remove(vol.getId());
         }
     }
+
+    @Override
+    public void moveVolumeOnSecondaryStorageToAnotherAccount(Volume volume, Account sourceAccount, Account destAccount) {
+        VolumeDataStoreVO volumeStore = _volumeStoreDao.findByVolume(volume.getId());
+
+        if (volumeStore == null) {
+            s_logger.debug(String.format("Volume [%s] is not present in the secondary storage. Therefore we do not need to move it in the secondary storage.", volume));
+            return;
+        }
+        s_logger.debug(String.format("Volume [%s] is present in secondary storage. It will be necessary to move it from the source account's [%s] folder to the destination "
+                        + "account's [%s] folder.",
+                volume.getUuid(), sourceAccount, destAccount));
+
+        VolumeInfo volumeInfo = volFactory.getVolume(volume.getId(), DataStoreRole.Image);
+        String datastoreUri = volumeInfo.getDataStore().getUri();
+        Path srcPath = Paths.get(volumeInfo.getPath());
+        String destPath = buildVolumePath(destAccount.getAccountId(), volume.getId());
+
+        EndPoint ssvm = _epSelector.findSsvm(volume.getDataCenterId());
+
+        MoveVolumeCommand cmd = new MoveVolumeCommand(volume.getUuid(), volume.getName(), destPath, srcPath.getParent().toString(), datastoreUri);
+
+        Answer answer = ssvm.sendMessage(cmd);
+
+        if (!answer.getResult()) {
+            String msg = String.format("Unable to move volume [%s] from [%s] (source account's [%s] folder) to [%s] (destination account's [%s] folder) in the secondary storage, due "
+                            + "to [%s].",
+                    volume.getUuid(), srcPath.getParent(), sourceAccount, destPath, destAccount, answer.getDetails());
+            s_logger.error(msg);
+            throw new CloudRuntimeException(msg);
+        }
+
+        s_logger.debug(String.format("Volume [%s] was moved from [%s] (source account's [%s] folder) to [%s] (destination account's [%s] folder) in the secondary storage.",
+                volume.getUuid(), srcPath.getParent(), sourceAccount, destPath, destAccount));
+
+        volumeStore.setInstallPath(String.format("%s/%s", destPath, srcPath.getFileName().toString()));
+        if (!_volumeStoreDao.update(volumeStore.getId(), volumeStore)) {
+            String msg = String.format("Unable to update volume [%s] install path in the DB.", volumeStore.getVolumeId());
+            s_logger.error(msg);
+            throw new CloudRuntimeException(msg);
+        }
+    }
+
+    protected String buildVolumePath(long accountId, long volumeId) {
+        return String.format("%s/%s/%s", TemplateConstants.DEFAULT_VOLUME_ROOT_DIR, accountId, volumeId);
+    }
 }

server/src/main/java/com/cloud/server/ManagementServerImpl.java

@@ -724,6 +724,7 @@ import org.apache.cloudstack.api.command.user.vmsnapshot.DeleteVMSnapshotCmd;
 import org.apache.cloudstack.api.command.user.vmsnapshot.ListVMSnapshotCmd;
 import org.apache.cloudstack.api.command.user.vmsnapshot.RevertToVMSnapshotCmd;
 import org.apache.cloudstack.api.command.user.volume.AddResourceDetailCmd;
+import org.apache.cloudstack.api.command.user.volume.AssignVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.AttachVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.ChangeOfferingForVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
@@ -3717,6 +3718,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
         cmdList.add(ListResourceIconCmd.class);
         cmdList.add(PatchSystemVMCmd.class);
         cmdList.add(ListGuestVlansCmd.class);
+        cmdList.add(AssignVolumeCmd.class);
 
         // Out-of-band management APIs for admins
         cmdList.add(EnableOutOfBandManagementForHostCmd.class);

server/src/main/java/com/cloud/storage/VolumeApiServiceImpl.java

@@ -33,6 +33,9 @@ import java.util.concurrent.ExecutionException;
 
 import javax.inject.Inject;
 
+import com.cloud.projects.Project;
+import com.cloud.projects.ProjectManager;
+import org.apache.cloudstack.api.command.user.volume.AssignVolumeCmd;
 import org.apache.cloudstack.api.ApiErrorCode;
 import org.apache.cloudstack.api.ServerApiException;
 import org.apache.cloudstack.api.ApiConstants.IoDriverPolicy;
@@ -93,6 +96,7 @@ import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
 import org.apache.cloudstack.storage.image.datastore.ImageStoreEntity;
+import org.apache.cloudstack.utils.bytescale.ByteScaleUtils;
 import org.apache.cloudstack.utils.identity.ManagementServerNode;
 import org.apache.cloudstack.utils.imagestore.ImageStoreUtil;
 import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
@@ -183,6 +187,7 @@ import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.Transaction;
 import com.cloud.utils.db.TransactionCallback;
+import com.cloud.utils.db.TransactionCallbackNoReturn;
 import com.cloud.utils.db.TransactionCallbackWithException;
 import com.cloud.utils.db.TransactionStatus;
 import com.cloud.utils.db.UUIDManager;
@@ -318,6 +323,9 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
     @Inject
     protected SnapshotHelper snapshotHelper;
 
+    @Inject
+    protected ProjectManager projectManager;
+
     protected Gson _gson;
 
     private static final List<HypervisorType> SupportedHypervisorsForVolResize = Arrays.asList(HypervisorType.KVM, HypervisorType.XenServer,
@@ -3745,6 +3753,126 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         return orchestrateExtractVolume(volume.getId(), zoneId);
     }
 
+    @Override
+    @ActionEvent(eventType = EventTypes.EVENT_VOLUME_CREATE, eventDescription = "Assigning volume to new account", async = false)
+    public Volume assignVolumeToAccount(AssignVolumeCmd command) throws ResourceAllocationException {
+        Account caller = CallContext.current().getCallingAccount();
+        VolumeVO volume = _volsDao.findById(command.getVolumeId());
+        Map<String, String> fullUrlParams = command.getFullUrlParams();
+
+        validateVolume(fullUrlParams.get("volumeid"), volume);
+
+        Account oldAccount = _accountMgr.getActiveAccountById(volume.getAccountId());
+        Account newAccount = getAccountOrProject(fullUrlParams.get("projectid"), command.getAccountId(), command.getProjectid(), caller);
+
+        validateAccounts(fullUrlParams.get("accountid"), volume, oldAccount, newAccount);
+
+        _accountMgr.checkAccess(caller, null, true, oldAccount);
+        _accountMgr.checkAccess(caller, null, true, newAccount);
+
+        _resourceLimitMgr.checkResourceLimit(newAccount, ResourceType.volume, ByteScaleUtils.bytesToGibibytes(volume.getSize()));
+        _resourceLimitMgr.checkResourceLimit(newAccount, ResourceType.primary_storage, volume.getSize());
+
+        Transaction.execute(new TransactionCallbackNoReturn() {
+            @Override
+            public void doInTransactionWithoutResult(TransactionStatus status) {
+                updateVolumeAccount(oldAccount, volume, newAccount);
+            }
+        });
+
+        return volume;
+    }
+
+    protected void updateVolumeAccount(Account oldAccount, VolumeVO volume, Account newAccount) {
+        UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VOLUME_DELETE, volume.getAccountId(), volume.getDataCenterId(), volume.getId(), volume.getName(),
+                Volume.class.getName(), volume.getUuid(), volume.isDisplayVolume());
+        _resourceLimitMgr.decrementResourceCount(oldAccount.getAccountId(), ResourceType.volume, ByteScaleUtils.bytesToGibibytes(volume.getSize()));
+        _resourceLimitMgr.decrementResourceCount(oldAccount.getAccountId(), ResourceType.primary_storage, volume.getSize());
+
+        volume.setAccountId(newAccount.getAccountId());
+        volume.setDomainId(newAccount.getDomainId());
+        _volsDao.persist(volume);
+
+        _resourceLimitMgr.incrementResourceCount(newAccount.getAccountId(), ResourceType.volume, ByteScaleUtils.bytesToGibibytes(volume.getSize()));
+        _resourceLimitMgr.incrementResourceCount(newAccount.getAccountId(), ResourceType.primary_storage, volume.getSize());
+
+        UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VOLUME_CREATE, volume.getAccountId(), volume.getDataCenterId(), volume.getId(), volume.getName(),
+                volume.getDiskOfferingId(), volume.getTemplateId(), volume.getSize(), Volume.class.getName(),
+                volume.getUuid(), volume.isDisplayVolume());
+
+        volService.moveVolumeOnSecondaryStorageToAnotherAccount(volume, oldAccount, newAccount);
+    }
+
+    /**
+     * Validates if the accounts are null, if the new account state is correct, and if the two accounts are the same.
+     * Throws {@link InvalidParameterValueException}.
+     * */
+    protected void validateAccounts(String newAccountUuid, VolumeVO volume, Account oldAccount, Account newAccount) {
+        if (oldAccount == null) {
+            throw new InvalidParameterValueException(String.format("The current account of the volume [%s] is invalid.",
+                    ReflectionToStringBuilderUtils.reflectOnlySelectedFields(volume, "name", "uuid")));
+        }
+
+        if (newAccount == null) {
+            throw new InvalidParameterValueException(String.format("UUID of the destination account is invalid. No account was found with UUID [%s].", newAccountUuid));
+        }
+
+        if (newAccount.getState() == Account.State.DISABLED || newAccount.getState() == Account.State.LOCKED) {
+            throw new InvalidParameterValueException(String.format("Unable to assign volume to destination account [%s], as it is in [%s] state.", newAccount,
+                    newAccount.getState().toString()));
+        }
+
+        if (oldAccount.getAccountId() == newAccount.getAccountId()) {
+            throw new InvalidParameterValueException(String.format("The new account and the old account are the same [%s].", oldAccount));
+        }
+    }
+
+    /**
+     * Validates if the volume can be reassigned to another account.
+     * Throws {@link InvalidParameterValueException} if volume is null.
+     * Throws {@link PermissionDeniedException} if volume is attached to a VM or if it has snapshots.
+     * */
+    protected void validateVolume(String volumeUuid, VolumeVO volume) {
+        if (volume == null) {
+            throw new InvalidParameterValueException(String.format("No volume was found with UUID [%s].", volumeUuid));
+        }
+
+        String volumeToString = ReflectionToStringBuilderUtils.reflectOnlySelectedFields(volume, "name", "uuid");
+
+        if (volume.getInstanceId() != null) {
+            VMInstanceVO vmInstanceVo = _vmInstanceDao.findById(volume.getInstanceId());
+            String msg = String.format("Volume [%s] is attached to [%s], so it cannot be moved to a different account.", volumeToString, vmInstanceVo);
+            s_logger.error(msg);
+            throw new PermissionDeniedException(msg);
+        }
+
+        List<SnapshotVO> snapshots = _snapshotDao.listByStatusNotIn(volume.getId(), Snapshot.State.Destroyed, Snapshot.State.Error);
+        if (CollectionUtils.isNotEmpty(snapshots)) {
+            throw new PermissionDeniedException(String.format("Volume [%s] has snapshots. Remove the volume's snapshots before assigning it to another account.", volumeToString));
+        }
+    }
+
+    protected Account getAccountOrProject(String projectUuid, Long accountId, Long projectId, Account caller) {
+        if (projectId != null && accountId != null) {
+            throw new InvalidParameterValueException("Both 'accountid' and 'projectid' were informed. You must inform only one of them.");
+        }
+
+        if (projectId != null) {
+            Project project = projectManager.getProject(projectId);
+            if (project == null) {
+                throw new InvalidParameterValueException(String.format("Unable to find project [%s]", projectUuid));
+            }
+
+            if (!projectManager.canAccessProjectAccount(caller, project.getProjectAccountId())) {
+                throw new PermissionDeniedException(String.format("Account [%s] does not have access to project [%s].", caller, projectUuid));
+            }
+
+            return _accountMgr.getAccount(project.getProjectAccountId());
+        }
+
+        return  _accountMgr.getActiveAccountById(accountId);
+    }
+
     private Optional<String> setExtractVolumeSearchCriteria(SearchCriteria<VolumeDataStoreVO> sc, VolumeVO volume) {
         final long volumeId = volume.getId();
         sc.addAnd("state", SearchCriteria.Op.EQ, ObjectInDataStoreStateMachine.State.Ready.toString());

server/src/test/java/com/cloud/storage/VolumeApiServiceImplTest.java

@@ -17,6 +17,10 @@
 package com.cloud.storage;
 
 import static org.junit.Assert.assertEquals;
+import com.cloud.exception.PermissionDeniedException;
+import com.cloud.projects.Project;
+import com.cloud.projects.ProjectManager;
+import com.cloud.storage.dao.SnapshotDao;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyObject;
@@ -60,6 +64,8 @@ import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
+import org.apache.cloudstack.utils.bytescale.ByteScaleUtils;
+import org.apache.commons.collections.CollectionUtils;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -191,12 +197,28 @@ public class VolumeApiServiceImplTest {
     private VolumeDataStoreVO volumeDataStoreVoMock;
     @Mock
     private AsyncCallFuture<VolumeApiResult> asyncCallFutureVolumeapiResultMock;
+    @Mock
+    private ArrayList<SnapshotVO> snapshotVOArrayListMock;
+
+    @Mock
+    private SnapshotDao snapshotDaoMock;
+
+    @Mock
+    private Project projectMock;
+
+    @Mock
+    private ProjectManager projectManagerMock;
 
     private long accountMockId = 456l;
     private long volumeMockId = 12313l;
     private long vmInstanceMockId = 1123l;
     private long volumeSizeMock = 456789921939l;
 
+    private String projectMockUuid = "projectUuid";
+    private long projecMockId = 13801801923810L;
+
+    private long projectMockAccountId = 132329390L;
+
     private long diskOfferingMockId = 100203L;
 
     private long offeringMockId = 31902L;
@@ -208,6 +230,9 @@ public class VolumeApiServiceImplTest {
         Mockito.lenient().doReturn(accountMockId).when(accountMock).getId();
         Mockito.doReturn(volumeSizeMock).when(volumeVoMock).getSize();
         Mockito.doReturn(volumeSizeMock).when(newDiskOfferingMock).getDiskSize();
+        Mockito.doReturn(projectMockUuid).when(projectMock).getUuid();
+        Mockito.doReturn(projecMockId).when(projectMock).getId();
+        Mockito.doReturn(projectMockAccountId).when(projectMock).getProjectAccountId();
 
         Mockito.doReturn(Mockito.mock(VolumeApiResult.class)).when(asyncCallFutureVolumeapiResultMock).get();
 
@@ -1317,6 +1342,138 @@ public class VolumeApiServiceImplTest {
         Assert.assertEquals(expectedResult, result);
     }
 
+    @Test (expected = InvalidParameterValueException.class)
+    public void checkIfVolumeCanBeReassignedTestNullVolume() {
+        volumeApiServiceImpl.validateVolume(volumeVoMock.getUuid(), null);
+    }
+
+    @Test (expected = PermissionDeniedException.class)
+    public void checkIfVolumeCanBeReassignedTestAttachedVolume() {
+        Mockito.doReturn(vmInstanceMockId).when(volumeVoMock).getInstanceId();
+
+        volumeApiServiceImpl.validateVolume(volumeVoMock.getUuid(), volumeVoMock);
+    }
+
+    @Test (expected = PermissionDeniedException.class)
+    @PrepareForTest (CollectionUtils.class)
+    public void checkIfVolumeCanBeReassignedTestVolumeWithSnapshots() {
+        Mockito.doReturn(null).when(volumeVoMock).getInstanceId();
+        Mockito.doReturn(snapshotVOArrayListMock).when(snapshotDaoMock).listByStatusNotIn(Mockito.anyLong(), Mockito.any(), Mockito.any());
+
+        PowerMockito.mockStatic(CollectionUtils.class);
+        PowerMockito.when(CollectionUtils.isNotEmpty(snapshotVOArrayListMock)).thenReturn(true);
+
+        volumeApiServiceImpl.validateVolume(volumeVoMock.getUuid(), volumeVoMock);
+    }
+
+    @Test
+    @PrepareForTest (CollectionUtils.class)
+    public void checkIfVolumeCanBeReassignedTestValidVolume() {
+        Mockito.doReturn(null).when(volumeVoMock).getInstanceId();
+        Mockito.doReturn(snapshotVOArrayListMock).when(snapshotDaoMock).listByStatusNotIn(Mockito.anyLong(), Mockito.any(), Mockito.any());
+
+        PowerMockito.mockStatic(CollectionUtils.class);
+        PowerMockito.when(CollectionUtils.isNotEmpty(snapshotVOArrayListMock)).thenReturn(false);
+
+        volumeApiServiceImpl.validateVolume(volumeVoMock.getUuid(), volumeVoMock);
+    }
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void validateAccountsTestNullOldAccount() {
+        volumeApiServiceImpl.validateAccounts(accountMock.getUuid(), volumeVoMock, null, accountMock);
+    }
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void validateAccountsTestNullNewAccount() {
+        volumeApiServiceImpl.validateAccounts(accountMock.getUuid(), volumeVoMock, accountMock, null);
+    }
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void validateAccountsTestDisabledNewAccount() {
+        Mockito.doReturn(Account.State.DISABLED).when(accountMock).getState();
+        volumeApiServiceImpl.validateAccounts(accountMock.getUuid(), volumeVoMock, null, accountMock);
+    }
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void validateAccountsTestLockedNewAccount() {
+        Mockito.doReturn(Account.State.LOCKED).when(accountMock).getState();
+        volumeApiServiceImpl.validateAccounts(accountMock.getUuid(), volumeVoMock, null, accountMock);
+    }
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void validateAccountsTestSameAccounts() {
+        volumeApiServiceImpl.validateAccounts(accountMock.getUuid(), volumeVoMock, accountMock, accountMock);
+    }
+
+    @Test
+    public void validateAccountsTestValidAccounts() {
+        Account newAccount = new AccountVO(accountMockId+1);
+        volumeApiServiceImpl.validateAccounts(accountMock.getUuid(), volumeVoMock, accountMock, newAccount);
+    }
+
+    @Test
+    @PrepareForTest(UsageEventUtils.class)
+    public void updateVolumeAccountTest() {
+        PowerMockito.mockStatic(UsageEventUtils.class);
+        Account newAccountMock = new AccountVO(accountMockId+1);
+
+        Mockito.doReturn(volumeVoMock).when(volumeDaoMock).persist(volumeVoMock);
+
+        volumeApiServiceImpl.updateVolumeAccount(accountMock, volumeVoMock, newAccountMock);
+
+        PowerMockito.verifyStatic(UsageEventUtils.class);
+        UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VOLUME_DELETE, volumeVoMock.getAccountId(), volumeVoMock.getDataCenterId(), volumeVoMock.getId(),
+                volumeVoMock.getName(), Volume.class.getName(), volumeVoMock.getUuid(), volumeVoMock.isDisplayVolume());
+
+        Mockito.verify(resourceLimitServiceMock).decrementResourceCount(accountMock.getAccountId(), ResourceType.volume, ByteScaleUtils.bytesToGibibytes(volumeVoMock.getSize()));
+        Mockito.verify(resourceLimitServiceMock).decrementResourceCount(accountMock.getAccountId(), ResourceType.primary_storage, volumeVoMock.getSize());
+
+        Mockito.verify(volumeVoMock).setAccountId(newAccountMock.getAccountId());
+        Mockito.verify(volumeVoMock).setDomainId(newAccountMock.getDomainId());
+
+        Mockito.verify(volumeDaoMock).persist(volumeVoMock);
+
+        Mockito.verify(resourceLimitServiceMock).incrementResourceCount(newAccountMock.getAccountId(), ResourceType.volume, ByteScaleUtils.bytesToGibibytes(volumeVoMock.getSize()));
+        Mockito.verify(resourceLimitServiceMock).incrementResourceCount(newAccountMock.getAccountId(), ResourceType.primary_storage, volumeVoMock.getSize());
+
+        PowerMockito.verifyStatic(UsageEventUtils.class);
+        UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VOLUME_DELETE, volumeVoMock.getAccountId(), volumeVoMock.getDataCenterId(), volumeVoMock.getId(),
+                volumeVoMock.getName(), Volume.class.getName(), volumeVoMock.getUuid(), volumeVoMock.isDisplayVolume());
+
+        Mockito.verify(volumeServiceMock).moveVolumeOnSecondaryStorageToAnotherAccount(volumeVoMock, accountMock, newAccountMock);
+    }
+
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void getAccountOrProjectTestAccountAndProjectInformed() {
+        volumeApiServiceImpl.getAccountOrProject(projectMock.getUuid(), accountMock.getId(), projectMock.getId(), accountMock);
+    }
+
+    @Test (expected = InvalidParameterValueException.class)
+    public void getAccountOrProjectTestUnableToFindProject() {
+        Mockito.doReturn(null).when(projectManagerMock).getProject(projecMockId);
+        volumeApiServiceImpl.getAccountOrProject(projectMock.getUuid(), null, projectMock.getId(), accountMock);
+    }
+
+    @Test (expected = PermissionDeniedException.class)
+    public void getAccountOrProjectTestCallerDoesNotHaveAccessToProject() {
+        Mockito.doReturn(projectMock).when(projectManagerMock).getProject(projecMockId);
+        Mockito.doReturn(false).when(projectManagerMock).canAccessProjectAccount(accountMock, projectMockAccountId);
+        volumeApiServiceImpl.getAccountOrProject(projectMock.getUuid(), null, projectMock.getId(), accountMock);
+    }
+
+    @Test
+    public void getAccountOrProjectTestValidProject() {
+        Mockito.doReturn(projectMock).when(projectManagerMock).getProject(projecMockId);
+        Mockito.doReturn(true).when(projectManagerMock).canAccessProjectAccount(accountMock, projectMockAccountId);
+        volumeApiServiceImpl.getAccountOrProject(projectMock.getUuid(), null, projectMock.getId(), accountMock);
+    }
+
+    @Test
+    public void getAccountOrProjectTestValidAccount() {
+        volumeApiServiceImpl.getAccountOrProject(projectMock.getUuid(), accountMock.getId(),null, accountMock);
+    }
+
     @Test
     @PrepareForTest(UsageEventUtils.class)
     public void publishVolumeCreationUsageEventTestNullDiskOfferingId() {

services/secondary-storage/server/src/main/java/org/apache/cloudstack/storage/resource/NfsSecondaryStorageResource.java

@@ -41,6 +41,7 @@ import java.net.URI;
 import java.net.UnknownHostException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -58,6 +59,7 @@ import org.apache.cloudstack.storage.command.CopyCommand;
 import org.apache.cloudstack.storage.command.DeleteCommand;
 import org.apache.cloudstack.storage.command.DownloadCommand;
 import org.apache.cloudstack.storage.command.DownloadProgressCommand;
+import org.apache.cloudstack.storage.command.MoveVolumeCommand;
 import org.apache.cloudstack.storage.command.TemplateOrVolumePostUploadCommand;
 import org.apache.cloudstack.storage.command.UploadStatusAnswer;
 import org.apache.cloudstack.storage.command.UploadStatusAnswer.UploadStatus;
@@ -73,6 +75,7 @@ import org.apache.cloudstack.storage.to.SnapshotObjectTO;
 import org.apache.cloudstack.storage.to.TemplateObjectTO;
 import org.apache.cloudstack.storage.to.VolumeObjectTO;
 import org.apache.cloudstack.utils.imagestore.ImageStoreUtil;
+import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils;
 import org.apache.cloudstack.utils.security.DigestHelper;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.io.FileUtils;
@@ -311,6 +314,8 @@ public class NfsSecondaryStorageResource extends ServerResourceBase implements S
             return execute((GetDatadisksCommand)cmd);
         } else if (cmd instanceof CreateDatadiskTemplateCommand) {
             return execute((CreateDatadiskTemplateCommand)cmd);
+        } else if (cmd instanceof MoveVolumeCommand) {
+            return execute((MoveVolumeCommand)cmd);
         } else {
             return Answer.createUnsupportedCommandAnswer(cmd);
         }
@@ -544,6 +549,36 @@ public class NfsSecondaryStorageResource extends ServerResourceBase implements S
         return new CreateDatadiskTemplateAnswer(diskTemplate);
     }
 
+    public Answer execute(MoveVolumeCommand cmd) {
+        String volumeToString = ReflectionToStringBuilderUtils.reflectOnlySelectedFields(cmd, "volumeUuid", "volumeName");
+
+        String rootDir = getRootDir(cmd.getDatastoreUri(), _nfsVersion);
+
+        if (!rootDir.endsWith("/")) {
+            rootDir += "/";
+        }
+
+        Path srcPath = Paths.get(rootDir + cmd.getSrcPath());
+        Path destPath = Paths.get(rootDir + cmd.getDestPath());
+
+        try {
+            s_logger.debug(String.format("Trying to create missing directories (if any) to move volume [%s].", volumeToString));
+            Files.createDirectories(destPath.getParent());
+            s_logger.debug(String.format("Trying to move volume [%s] to [%s].", volumeToString, destPath));
+            Files.move(srcPath, destPath);
+
+            String msg = String.format("Moved volume [%s] from [%s] to [%s].", volumeToString, srcPath, destPath);
+            s_logger.debug(msg);
+
+            return new Answer(cmd, true, msg);
+
+        } catch (IOException ioException) {
+            s_logger.error(String.format("Failed to move volume [%s] from [%s] to [%s] due to [%s].", volumeToString, srcPath, destPath, ioException.getMessage()),
+                    ioException);
+            return new Answer(cmd, ioException);
+        }
+    }
+
     /*
      *  return Pair of <Template relative path, Template name>
      *  Template url may or may not end with .ova extension

utils/src/main/java/org/apache/cloudstack/utils/bytescale/ByteScaleUtils.java

@@ -55,4 +55,14 @@ public class ByteScaleUtils {
     public static long bytesToMebibytes(long bytes) {
         return bytes / MiB;
     }
+
+    /**
+     * Converts bytes to gibibytes.
+     *
+     * @param b The value in bytes to convert to gibibytes.
+     * @return The parameter divided by 1024 * 1024 * 1024 (1 GiB).
+     */
+    public static long bytesToGibibytes(long b) {
+        return b / GiB;
+    }
 }

utils/src/test/java/org/apache/cloudstack/utils/bytescale/ByteScaleUtilsTest.java

@@ -43,6 +43,13 @@ public class ByteScaleUtilsTest {
         Assert.assertEquals(mib, ByteScaleUtils.bytesToMebibytes(bytes));
     }
 
+    @Test
+    public void validateBytesToGib(){
+        long gib = 3000L;
+        long b = 1024L * 1024L * 1024L * gib;
+        Assert.assertEquals(gib, ByteScaleUtils.bytesToGibibytes(b));
+    }
+
     @Test
     public void validateMebibytesToBytesIfIntTimesIntThenMustExtrapolateIntMaxValue() {
         int mib = 3000;