New API "checkVolume" to check and repair any leaks or issues reported by qemu-img check (#8577)

* Introduced a new API checkVolumeAndRepair that allows users or admins to check and repair if any leaks observed.
Currently this is supported only for KVM

* some fixes

* Added unit tests

* addressed review comments

* add repair volume while granting access

* Changed repair parameter to accept both leaks/all

* Introduced new global setting volume.check.and.repair.before.use to do volume check and repair before VM start or volume attach operations

* Added volume check and repair changes only during VM start and volume attach operations

* Refactored the names to look similar across the code

* Some code fixes

* remove unused code

* Renamed repair values

* Fixed unit tests

* changed version

* Address review comments

* Code refactored

* used volume name in logs

* Changed the API to Async and the setting scope to storage pool

* Fixed exit value handling with check volume command

* Fixed storage scope to the setting

* Fix volume format issues

* Refactored the log messages

* Fix formatting

api/src/main/java/com/cloud/event/EventTypes.java

@@ -303,6 +303,7 @@ public class EventTypes {
     public static final String EVENT_VOLUME_CREATE = "VOLUME.CREATE";
     public static final String EVENT_VOLUME_DELETE = "VOLUME.DELETE";
     public static final String EVENT_VOLUME_ATTACH = "VOLUME.ATTACH";
+    public static final String EVENT_VOLUME_CHECK = "VOLUME.CHECK";
     public static final String EVENT_VOLUME_DETACH = "VOLUME.DETACH";
     public static final String EVENT_VOLUME_EXTRACT = "VOLUME.EXTRACT";
     public static final String EVENT_VOLUME_UPLOAD = "VOLUME.UPLOAD";

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

@@ -22,9 +22,11 @@ import java.net.MalformedURLException;
 import java.util.List;
 import java.util.Map;
 
+import com.cloud.utils.Pair;
 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.CheckAndRepairVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.DetachVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.ExtractVolumeCmd;
@@ -178,4 +180,6 @@ public interface VolumeApiService {
     void publishVolumeCreationUsageEvent(Volume volume);
 
     boolean stateTransitTo(Volume vol, Volume.Event event) throws NoTransitionException;
+
+    Pair<String, String> checkAndRepairVolume(CheckAndRepairVolumeCmd cmd) throws ResourceAllocationException;
 }

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

@@ -378,6 +378,7 @@ public class ApiConstants {
     public static final String RECEIVED_BYTES = "receivedbytes";
     public static final String RECONNECT = "reconnect";
     public static final String RECOVER = "recover";
+    public static final String REPAIR = "repair";
     public static final String REQUIRES_HVM = "requireshvm";
     public static final String RESOURCE_NAME = "resourcename";
     public static final String RESOURCE_TYPE = "resourcetype";
@@ -501,6 +502,9 @@ public class ApiConstants {
     public static final String IS_VOLATILE = "isvolatile";
     public static final String VOLUME_ID = "volumeid";
     public static final String VOLUMES = "volumes";
+    public static final String VOLUME_CHECK_RESULT = "volumecheckresult";
+    public static final String VOLUME_REPAIR_RESULT = "volumerepairresult";
+
     public static final String ZONE = "zone";
     public static final String ZONE_ID = "zoneid";
     public static final String ZONE_NAME = "zonename";

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

@@ -0,0 +1,139 @@
+// 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.event.EventTypes;
+import com.cloud.exception.InvalidParameterValueException;
+import org.apache.cloudstack.acl.RoleType;
+import org.apache.cloudstack.api.APICommand;
+import org.apache.cloudstack.api.ApiCommandResourceType;
+import org.apache.cloudstack.api.ApiConstants;
+import org.apache.cloudstack.api.ApiErrorCode;
+import org.apache.cloudstack.api.BaseAsyncCmd;
+import org.apache.cloudstack.api.Parameter;
+import org.apache.cloudstack.api.ResponseObject.ResponseView;
+import org.apache.cloudstack.api.ServerApiException;
+import org.apache.cloudstack.api.response.VolumeResponse;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.log4j.Logger;
+
+import com.cloud.exception.ResourceAllocationException;
+import com.cloud.storage.Volume;
+import com.cloud.user.Account;
+import com.cloud.utils.Pair;
+import com.cloud.utils.StringUtils;
+
+import java.util.Arrays;
+
+@APICommand(name = "checkVolume", description = "Check the volume for any errors or leaks and also repairs when repair parameter is passed, this is currently supported for KVM only", responseObject = VolumeResponse.class, entityType = {Volume.class},
+        since = "4.19.1",
+        authorized = {RoleType.Admin, RoleType.ResourceAdmin, RoleType.DomainAdmin, RoleType.User})
+public class CheckAndRepairVolumeCmd extends BaseAsyncCmd {
+    public static final Logger s_logger = Logger.getLogger(CheckAndRepairVolumeCmd.class.getName());
+
+    private static final String s_name = "checkandrepairvolumeresponse";
+
+    /////////////////////////////////////////////////////
+    //////////////// API parameters /////////////////////
+    /////////////////////////////////////////////////////
+
+    @Parameter(name = ApiConstants.ID, type = CommandType.UUID, entityType = VolumeResponse.class, required = true, description = "The ID of the volume")
+    private Long id;
+
+    @Parameter(name = ApiConstants.REPAIR, type = CommandType.STRING, required = false, description = "parameter to repair the volume, leaks or all are the possible values")
+    private String repair;
+
+    /////////////////////////////////////////////////////
+    /////////////////// Accessors ///////////////////////
+    /////////////////////////////////////////////////////
+
+    public enum RepairValues {
+        LEAKS, ALL
+    }
+
+    public Long getId() {
+        return id;
+    }
+
+    public String getRepair() {
+        if (org.apache.commons.lang3.StringUtils.isNotEmpty(repair)) {
+            RepairValues repairType = Enum.valueOf(RepairValues.class, repair.toUpperCase());
+            if (repairType == null) {
+                throw new InvalidParameterValueException(String.format("Repair parameter can only take the following values: %s" + Arrays.toString(RepairValues.values())));
+            }
+            return repair.toLowerCase();
+        }
+        return null;
+    }
+
+    /////////////////////////////////////////////////////
+    /////////////// API Implementation///////////////////
+    /////////////////////////////////////////////////////
+
+    @Override
+    public String getCommandName() {
+        return s_name;
+    }
+
+    @Override
+    public long getEntityOwnerId() {
+        Volume volume = _entityMgr.findById(Volume.class, getId());
+        if (volume != null) {
+            return volume.getAccountId();
+        }
+
+        return Account.ACCOUNT_ID_SYSTEM; // no account info given, parent this command to SYSTEM so ERROR events are tracked
+    }
+
+    @Override
+    public String getEventType() {
+        return EventTypes.EVENT_VOLUME_CHECK;
+    }
+
+    @Override
+    public String getEventDescription() {
+        return String.format("check and repair operation on volume: %s", this._uuidMgr.getUuid(Volume.class, getId()));
+    }
+
+    @Override
+    public Long getApiResourceId() {
+        return id;
+    }
+
+    @Override
+    public ApiCommandResourceType getApiResourceType() {
+        return ApiCommandResourceType.Volume;
+    }
+
+    @Override
+    public void execute() throws ResourceAllocationException {
+        CallContext.current().setEventDetails("Volume Id: " + getId());
+        Pair<String, String> result = _volumeService.checkAndRepairVolume(this);
+        Volume volume = _responseGenerator.findVolumeById(getId());
+        if (result != null) {
+            VolumeResponse response = _responseGenerator.createVolumeResponse(ResponseView.Full, volume);
+            response.setVolumeCheckResult(StringUtils.parseJsonToMap(result.first()));
+            if (getRepair() != null) {
+                response.setVolumeRepairResult(StringUtils.parseJsonToMap(result.second()));
+            }
+            response.setResponseName(getCommandName());
+            setResponseObject(response);
+        } else {
+            throw new ServerApiException(ApiErrorCode.INTERNAL_ERROR, "Failed to check volume and repair");
+        }
+    }
+}

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

@@ -18,6 +18,7 @@ package org.apache.cloudstack.api.response;
 
 import java.util.Date;
 import java.util.LinkedHashSet;
+import java.util.Map;
 import java.util.Set;
 
 import org.apache.cloudstack.acl.RoleType;
@@ -288,6 +289,14 @@ public class VolumeResponse extends BaseResponseWithTagInformation implements Co
     @Param(description = "volume uuid that is given by virtualisation provider (only for VMware)")
     private String externalUuid;
 
+    @SerializedName(ApiConstants.VOLUME_CHECK_RESULT)
+    @Param(description = "details for the volume check result, they may vary for different hypervisors, since = 4.19.1")
+    private Map<String, String> volumeCheckResult;
+
+    @SerializedName(ApiConstants.VOLUME_REPAIR_RESULT)
+    @Param(description = "details for the volume repair result, they may vary for different hypervisors, since = 4.19.1")
+    private Map<String, String> volumeRepairResult;
+
     public String getPath() {
         return path;
     }
@@ -817,4 +826,20 @@ public class VolumeResponse extends BaseResponseWithTagInformation implements Co
     public void setExternalUuid(String externalUuid) {
         this.externalUuid = externalUuid;
     }
+
+    public Map<String, String> getVolumeCheckResult() {
+        return volumeCheckResult;
+    }
+
+    public void setVolumeCheckResult(Map<String, String> volumeCheckResult) {
+        this.volumeCheckResult = volumeCheckResult;
+    }
+
+    public Map<String, String> getVolumeRepairResult() {
+        return volumeRepairResult;
+    }
+
+    public void setVolumeRepairResult(Map<String, String> volumeRepairResult) {
+        this.volumeRepairResult = volumeRepairResult;
+    }
 }

core/src/main/java/com/cloud/agent/api/storage/CheckAndRepairVolumeAnswer.java

@@ -0,0 +1,57 @@
+//
+// 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.agent.api.storage;
+
+import com.cloud.agent.api.Answer;
+
+public class CheckAndRepairVolumeAnswer extends Answer {
+    private String volumeCheckExecutionResult;
+    private String volumeRepairExecutionResult;
+
+    protected CheckAndRepairVolumeAnswer() {
+        super();
+    }
+
+    public CheckAndRepairVolumeAnswer(CheckAndRepairVolumeCommand cmd, boolean result, String details, String volumeCheckExecutionResult, String volumeRepairedExecutionResult) {
+        super(cmd, result, details);
+        this.volumeCheckExecutionResult = volumeCheckExecutionResult;
+        this.volumeRepairExecutionResult = volumeRepairedExecutionResult;
+    }
+
+    public CheckAndRepairVolumeAnswer(CheckAndRepairVolumeCommand cmd, boolean result, String details) {
+        super(cmd, result, details);
+    }
+
+    public String getVolumeCheckExecutionResult() {
+        return volumeCheckExecutionResult;
+    }
+
+    public String getVolumeRepairExecutionResult() {
+        return volumeRepairExecutionResult;
+    }
+
+    public void setVolumeCheckExecutionResult(String volumeCheckExecutionResult) {
+        this.volumeCheckExecutionResult = volumeCheckExecutionResult;
+    }
+
+    public void setVolumeRepairExecutionResult(String volumeRepairExecutionResult) {
+        this.volumeRepairExecutionResult = volumeRepairExecutionResult;
+    }
+}

core/src/main/java/com/cloud/agent/api/storage/CheckAndRepairVolumeCommand.java

@@ -0,0 +1,77 @@
+//
+// 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.agent.api.storage;
+
+import com.cloud.agent.api.Command;
+import com.cloud.agent.api.LogLevel;
+import com.cloud.agent.api.to.StorageFilerTO;
+
+import java.util.Arrays;
+
+public class CheckAndRepairVolumeCommand extends Command {
+    private String path;
+    private StorageFilerTO pool;
+    private String repair;
+    @LogLevel(LogLevel.Log4jLevel.Off)
+    private byte[] passphrase;
+    private String encryptFormat;
+
+    public CheckAndRepairVolumeCommand(String path, StorageFilerTO pool, String repair, byte[] passphrase, String encryptFormat) {
+        this.path = path;
+        this.pool = pool;
+        this.repair = repair;
+        this.passphrase = passphrase;
+        this.encryptFormat = encryptFormat;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public String getPoolUuid() {
+        return pool.getUuid();
+    }
+
+    public StorageFilerTO getPool() {
+        return pool;
+    }
+
+    public String getRepair() {
+        return repair;
+    }
+
+    public String getEncryptFormat() { return encryptFormat; }
+
+    public byte[] getPassphrase() { return passphrase; }
+
+    public void clearPassphrase() {
+        if (this.passphrase != null) {
+            Arrays.fill(this.passphrase, (byte) 0);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public boolean executeInSequence() {
+        return false;
+    }
+}

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

@@ -117,4 +117,8 @@ public interface VolumeService {
       VolumeInfo sourceVolume, VolumeInfo destinationVolume, boolean retryExpungeVolumeAsync);
 
     void moveVolumeOnSecondaryStorageToAnotherAccount(Volume volume, Account sourceAccount, Account destAccount);
+
+    Pair<String, String> checkAndRepairVolume(VolumeInfo volume);
+
+    void checkAndRepairVolumeBasedOnConfig(DataObject dataObject, Host host);
 }

engine/components-api/src/main/java/com/cloud/vm/VmWorkCheckAndRepairVolume.java

@@ -0,0 +1,42 @@
+// 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.vm;
+
+public class VmWorkCheckAndRepairVolume extends VmWork {
+
+    private static final long serialVersionUID = 341816293003023824L;
+
+    private Long volumeId;
+
+    private String repair;
+
+    public VmWorkCheckAndRepairVolume(long userId, long accountId, long vmId, String handlerName,
+                                      Long volumeId, String repair) {
+        super(userId, accountId, vmId, handlerName);
+        this.repair = repair;
+        this.volumeId = volumeId;
+    }
+
+    public Long getVolumeId() {
+        return volumeId;
+    }
+
+    public String getRepair() {
+        return repair;
+    }
+}

engine/orchestration/src/main/java/org/apache/cloudstack/engine/orchestration/VolumeOrchestrator.java

@@ -1915,6 +1915,8 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
                             }
                         }
                     }
+                } else {
+                    handleCheckAndRepairVolume(vol, vm.getVirtualMachine().getHostId());
                 }
             } else if (task.type == VolumeTaskType.MIGRATE) {
                 pool = (StoragePool)dataStoreMgr.getDataStore(task.pool.getId(), DataStoreRole.Primary);
@@ -1957,6 +1959,16 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
         }
     }
 
+    private void handleCheckAndRepairVolume(Volume vol, Long hostId) {
+        Host host = _hostDao.findById(hostId);
+        try {
+            volService.checkAndRepairVolumeBasedOnConfig(volFactory.getVolume(vol.getId()), host);
+        } catch (Exception e) {
+            String volumeToString = getReflectOnlySelectedFields(vol);
+            s_logger.debug(String.format("Unable to check and repair volume [%s] on host [%s], due to %s.", volumeToString, host, e.getMessage()));
+        }
+    }
+
     private boolean stateTransitTo(Volume vol, Volume.Event event) throws NoTransitionException {
         return _volStateMachine.transitTo(vol, event, null, _volsDao);
     }

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

@@ -32,8 +32,10 @@ import java.util.concurrent.ExecutionException;
 
 import javax.inject.Inject;
 
+import com.cloud.storage.VolumeApiServiceImpl;
 import org.apache.cloudstack.annotation.AnnotationService;
 import org.apache.cloudstack.annotation.dao.AnnotationDao;
+import org.apache.cloudstack.api.command.user.volume.CheckAndRepairVolumeCmd;
 import org.apache.cloudstack.engine.cloud.entity.api.VolumeEntity;
 import org.apache.cloudstack.engine.orchestration.service.VolumeOrchestrationService;
 import org.apache.cloudstack.engine.subsystem.api.storage.ChapInfo;
@@ -80,6 +82,7 @@ import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
 import org.apache.cloudstack.storage.image.store.TemplateObject;
 import org.apache.cloudstack.storage.to.TemplateObjectTO;
 import org.apache.cloudstack.storage.to.VolumeObjectTO;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.log4j.Logger;
 import org.springframework.stereotype.Component;
@@ -87,9 +90,12 @@ import org.springframework.stereotype.Component;
 import com.cloud.agent.AgentManager;
 import com.cloud.agent.api.Answer;
 import com.cloud.agent.api.ModifyTargetsCommand;
+import com.cloud.agent.api.storage.CheckAndRepairVolumeAnswer;
+import com.cloud.agent.api.storage.CheckAndRepairVolumeCommand;
 import com.cloud.agent.api.storage.ListVolumeAnswer;
 import com.cloud.agent.api.storage.ListVolumeCommand;
 import com.cloud.agent.api.storage.ResizeVolumeCommand;
+import com.cloud.agent.api.to.DataObjectType;
 import com.cloud.agent.api.to.StorageFilerTO;
 import com.cloud.agent.api.to.VirtualMachineTO;
 import com.cloud.alert.AlertManager;
@@ -110,6 +116,7 @@ import com.cloud.org.Cluster;
 import com.cloud.org.Grouping.AllocationState;
 import com.cloud.resource.ResourceState;
 import com.cloud.server.ManagementService;
+import com.cloud.storage.CheckAndRepairVolumePayload;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.RegisterVolumePayload;
 import com.cloud.storage.ScopeType;
@@ -198,7 +205,7 @@ public class VolumeServiceImpl implements VolumeService {
     @Inject
     private VolumeOrchestrationService _volumeMgr;
     @Inject
-    private StorageManager _storageMgr;
+    protected StorageManager _storageMgr;
     @Inject
     private AnnotationDao annotationDao;
     @Inject
@@ -2773,6 +2780,62 @@ public class VolumeServiceImpl implements VolumeService {
         return snapshot;
     }
 
+    @Override
+    public void checkAndRepairVolumeBasedOnConfig(DataObject dataObject, Host host) {
+        if (HypervisorType.KVM.equals(host.getHypervisorType()) && DataObjectType.VOLUME.equals(dataObject.getType())) {
+            VolumeInfo volumeInfo = volFactory.getVolume(dataObject.getId());
+            if (VolumeApiServiceImpl.AllowCheckAndRepairVolume.valueIn(volumeInfo.getPoolId())) {
+                s_logger.info(String.format("Trying to check and repair the volume %d", dataObject.getId()));
+                String repair = CheckAndRepairVolumeCmd.RepairValues.LEAKS.name().toLowerCase();
+                CheckAndRepairVolumePayload payload = new CheckAndRepairVolumePayload(repair);
+                volumeInfo.addPayload(payload);
+                checkAndRepairVolumeThroughHost(volumeInfo, host);
+            }
+        }
+    }
+
+    @Override
+    public Pair<String, String> checkAndRepairVolume(VolumeInfo volume) {
+        Long poolId = volume.getPoolId();
+        List<Long> hostIds = _storageMgr.getUpHostsInPool(poolId);
+        if (CollectionUtils.isEmpty(hostIds)) {
+            throw new CloudRuntimeException("Unable to find Up hosts to run the check volume command");
+        }
+        Collections.shuffle(hostIds);
+        Host host = _hostDao.findById(hostIds.get(0));
+
+        return checkAndRepairVolumeThroughHost(volume, host);
+
+    }
+
+    private Pair<String, String> checkAndRepairVolumeThroughHost(VolumeInfo volume, Host host) {
+        Long poolId = volume.getPoolId();
+        StoragePool pool = _storageMgr.getStoragePool(poolId);
+        CheckAndRepairVolumePayload payload = (CheckAndRepairVolumePayload) volume.getpayload();
+        CheckAndRepairVolumeCommand command = new CheckAndRepairVolumeCommand(volume.getPath(), new StorageFilerTO(pool), payload.getRepair(),
+                volume.getPassphrase(), volume.getEncryptFormat());
+
+        try {
+            grantAccess(volume, host, volume.getDataStore());
+            CheckAndRepairVolumeAnswer answer = (CheckAndRepairVolumeAnswer) _storageMgr.sendToPool(pool, new long[]{host.getId()}, command);
+            if (answer != null && answer.getResult()) {
+                s_logger.debug(String.format("Check volume response result: %s", answer.getDetails()));
+                return new Pair<>(answer.getVolumeCheckExecutionResult(), answer.getVolumeRepairExecutionResult());
+            } else {
+                String errMsg = (answer == null) ? null : answer.getDetails();
+                s_logger.debug(String.format("Failed to check and repair the volume with error %s", errMsg));
+            }
+
+        } catch (Exception e) {
+            s_logger.debug("sending check and repair volume command failed", e);
+        } finally {
+            revokeAccess(volume, host, volume.getDataStore());
+            command.clearPassphrase();
+        }
+
+        return null;
+    }
+
     // For managed storage on Xen and VMware, we need to potentially make space for hypervisor snapshots.
     // The disk offering can collect this information and pass it on to the volume that's about to be created.
     // Ex. if you want a 10 GB CloudStack volume to reside on managed storage on Xen, this leads to an SR

engine/storage/volume/src/test/java/org/apache/cloudstack/storage/volume/VolumeServiceTest.java

@@ -19,6 +19,15 @@
 
 package org.apache.cloudstack.storage.volume;
 
+import com.cloud.agent.api.storage.CheckAndRepairVolumeAnswer;
+import com.cloud.agent.api.storage.CheckAndRepairVolumeCommand;
+import com.cloud.agent.api.to.StorageFilerTO;
+import com.cloud.exception.StorageUnavailableException;
+import com.cloud.host.HostVO;
+import com.cloud.host.dao.HostDao;
+import com.cloud.storage.CheckAndRepairVolumePayload;
+import com.cloud.storage.StorageManager;
+import com.cloud.storage.StoragePool;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.VolumeDao;
 import com.cloud.storage.snapshot.SnapshotManager;
@@ -26,6 +35,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+
+import com.cloud.utils.Pair;
 import junit.framework.TestCase;
 import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
@@ -65,15 +76,26 @@ public class VolumeServiceTest extends TestCase{
     @Mock
     SnapshotManager snapshotManagerMock;
 
+    @Mock
+    StorageManager storageManagerMock;
+
     @Mock
     VolumeVO volumeVoMock;
 
+    @Mock
+    HostVO hostMock;
+
+    @Mock
+    HostDao hostDaoMock;
+
     @Before
     public void setup(){
         volumeServiceImplSpy = Mockito.spy(new VolumeServiceImpl());
         volumeServiceImplSpy.volFactory = volumeDataFactoryMock;
         volumeServiceImplSpy.volDao = volumeDaoMock;
         volumeServiceImplSpy.snapshotMgr = snapshotManagerMock;
+        volumeServiceImplSpy._storageMgr = storageManagerMock;
+        volumeServiceImplSpy._hostDao = hostDaoMock;
     }
 
     @Test(expected = InterruptedException.class)
@@ -210,4 +232,75 @@ public class VolumeServiceTest extends TestCase{
         volumeServiceImplSpy.destroySourceVolumeAfterMigration(ObjectInDataStoreStateMachine.Event.DestroyRequested, null, volumeObject,
           volumeObject, true);
     }
+
+    @Test
+    public void testCheckAndRepairVolume() throws StorageUnavailableException {
+        VolumeInfo volume = Mockito.mock(VolumeInfo.class);
+        Mockito.when(volume.getPoolId()).thenReturn(1L);
+        StoragePool pool = Mockito.mock(StoragePool.class);
+        Mockito.when(storageManagerMock.getStoragePool(1L)).thenReturn(pool);
+        List<Long> hostIds = new ArrayList<>();
+        hostIds.add(1L);
+        Mockito.when(storageManagerMock.getUpHostsInPool(1L)).thenReturn(hostIds);
+        Mockito.when(hostMock.getId()).thenReturn(1L);
+        Mockito.when(hostDaoMock.findById(1L)).thenReturn(hostMock);
+
+        CheckAndRepairVolumePayload payload = new CheckAndRepairVolumePayload(null);
+        Mockito.when(volume.getpayload()).thenReturn(payload);
+        Mockito.when(volume.getPath()).thenReturn("cbac516a-0f1f-4559-921c-1a7c6c408ccf");
+        Mockito.when(volume.getPassphrase()).thenReturn(new byte[] {3, 1, 2, 3});
+        Mockito.when(volume.getEncryptFormat()).thenReturn("LUKS");
+
+        String checkResult = "{\n" +
+                "    \"image-end-offset\": 6442582016,\n" +
+                "    \"total-clusters\": 163840,\n" +
+                "    \"check-errors\": 0,\n" +
+                "    \"leaks\": 124,\n" +
+                "    \"allocated-clusters\": 98154,\n" +
+                "    \"filename\": \"/var/lib/libvirt/images/26be20c7-b9d0-43f6-a76e-16c70737a0e0\",\n" +
+                "    \"format\": \"qcow2\",\n" +
+                "    \"fragmented-clusters\": 96135\n" +
+                "}";
+
+        CheckAndRepairVolumeCommand command = new CheckAndRepairVolumeCommand(volume.getPath(), new StorageFilerTO(pool), payload.getRepair(),
+                volume.getPassphrase(), volume.getEncryptFormat());
+
+        CheckAndRepairVolumeAnswer answer = new CheckAndRepairVolumeAnswer(command, true, checkResult);
+        answer.setVolumeCheckExecutionResult(checkResult);
+        Mockito.when(storageManagerMock.sendToPool(pool, new long[]{1L}, command)).thenReturn(answer);
+
+        Pair<String, String> result = volumeServiceImplSpy.checkAndRepairVolume(volume);
+
+        Assert.assertEquals(result.first(), checkResult);
+        Assert.assertEquals(result.second(), null);
+    }
+
+    @Test
+    public void testCheckAndRepairVolumeWhenFailure() throws StorageUnavailableException {
+        VolumeInfo volume = Mockito.mock(VolumeInfo.class);
+        Mockito.when(volume.getPoolId()).thenReturn(1L);
+        StoragePool pool = Mockito.mock(StoragePool.class);
+        Mockito.when(storageManagerMock.getStoragePool(1L)).thenReturn(pool);
+        List<Long> hostIds = new ArrayList<>();
+        hostIds.add(1L);
+        Mockito.when(storageManagerMock.getUpHostsInPool(1L)).thenReturn(hostIds);
+        Mockito.when(hostMock.getId()).thenReturn(1L);
+        Mockito.when(hostDaoMock.findById(1L)).thenReturn(hostMock);
+
+        CheckAndRepairVolumePayload payload = new CheckAndRepairVolumePayload(null);
+        Mockito.when(volume.getpayload()).thenReturn(payload);
+        Mockito.when(volume.getPath()).thenReturn("cbac516a-0f1f-4559-921c-1a7c6c408ccf");
+        Mockito.when(volume.getPassphrase()).thenReturn(new byte[] {3, 1, 2, 3});
+        Mockito.when(volume.getEncryptFormat()).thenReturn("LUKS");
+
+        CheckAndRepairVolumeCommand command = new CheckAndRepairVolumeCommand(volume.getPath(), new StorageFilerTO(pool), payload.getRepair(),
+                volume.getPassphrase(), volume.getEncryptFormat());
+
+        CheckAndRepairVolumeAnswer answer = new CheckAndRepairVolumeAnswer(command, false, "Unable to execute qemu command");
+        Mockito.when(storageManagerMock.sendToPool(pool, new long[]{1L}, command)).thenReturn(answer);
+
+        Pair<String, String> result = volumeServiceImplSpy.checkAndRepairVolume(volume);
+
+        Assert.assertEquals(null, result);
+    }
 }

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

@@ -0,0 +1,192 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.cloud.hypervisor.kvm.resource.wrapper;
+
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.storage.CheckAndRepairVolumeCommand;
+import com.cloud.agent.api.storage.CheckAndRepairVolumeAnswer;
+import com.cloud.agent.api.to.StorageFilerTO;
+import com.cloud.hypervisor.kvm.resource.LibvirtComputingResource;
+import com.cloud.hypervisor.kvm.storage.KVMPhysicalDisk;
+import com.cloud.hypervisor.kvm.storage.KVMStoragePool;
+import com.cloud.hypervisor.kvm.storage.KVMStoragePoolManager;
+import com.cloud.resource.CommandWrapper;
+import com.cloud.resource.ResourceWrapper;
+import com.cloud.utils.exception.CloudRuntimeException;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.cloudstack.utils.cryptsetup.KeyFile;
+import org.apache.cloudstack.utils.qemu.QemuImageOptions;
+import org.apache.cloudstack.utils.qemu.QemuImg;
+import org.apache.cloudstack.utils.qemu.QemuImgException;
+import org.apache.cloudstack.utils.qemu.QemuImgFile;
+import org.apache.cloudstack.utils.qemu.QemuObject;
+import org.apache.cloudstack.utils.qemu.QemuObject.EncryptFormat;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.log4j.Logger;
+import org.libvirt.LibvirtException;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+@ResourceWrapper(handles =  CheckAndRepairVolumeCommand.class)
+public class LibvirtCheckAndRepairVolumeCommandWrapper extends CommandWrapper<CheckAndRepairVolumeCommand, Answer, LibvirtComputingResource> {
+
+    private static final Logger s_logger = Logger.getLogger(LibvirtCheckAndRepairVolumeCommandWrapper.class);
+
+    @Override
+    public Answer execute(CheckAndRepairVolumeCommand command, LibvirtComputingResource serverResource) {
+        final String volumeId = command.getPath();
+        final String repair = command.getRepair();
+        final StorageFilerTO spool = command.getPool();
+
+        final KVMStoragePoolManager storagePoolMgr = serverResource.getStoragePoolMgr();
+        KVMStoragePool pool = storagePoolMgr.getStoragePool(spool.getType(), spool.getUuid());
+        final KVMPhysicalDisk vol = pool.getPhysicalDisk(volumeId);
+        byte[] passphrase = command.getPassphrase();
+
+        try {
+            CheckAndRepairVolumeAnswer answer = null;
+            String checkVolumeResult = null;
+            if (QemuImg.PhysicalDiskFormat.RAW.equals(vol.getFormat())) {
+                checkVolumeResult = "Volume format RAW is not supported to check and repair";
+                String jsonStringFormat = String.format("{ \"message\": \"%s\" }", checkVolumeResult);
+                answer = new CheckAndRepairVolumeAnswer(command, true, checkVolumeResult);
+                answer.setVolumeCheckExecutionResult(jsonStringFormat);
+
+                return answer;
+            } else {
+                answer = checkVolume(vol, command, serverResource);
+                checkVolumeResult =  answer.getVolumeCheckExecutionResult();
+            }
+
+            CheckAndRepairVolumeAnswer resultAnswer = checkIfRepairLeaksIsRequired(command, checkVolumeResult, vol.getName());
+            // resultAnswer is not null when repair is not required, so return from here
+            if (resultAnswer != null) {
+                return resultAnswer;
+            }
+
+            if (StringUtils.isNotEmpty(repair)) {
+                answer = repairVolume(vol, command, serverResource, checkVolumeResult);
+            }
+
+            return answer;
+        } catch (Exception e) {
+            return new CheckAndRepairVolumeAnswer(command, false, e.toString());
+        } finally {
+            if (passphrase != null) {
+                Arrays.fill(passphrase, (byte) 0);
+            }
+        }
+    }
+
+    private CheckAndRepairVolumeAnswer checkVolume(KVMPhysicalDisk vol, CheckAndRepairVolumeCommand command, LibvirtComputingResource serverResource) {
+        EncryptFormat encryptFormat = EncryptFormat.enumValue(command.getEncryptFormat());
+        byte[] passphrase = command.getPassphrase();
+        String checkVolumeResult = checkAndRepairVolume(vol, null, encryptFormat, passphrase, serverResource);
+        s_logger.info(String.format("Check Volume result for the volume %s is %s", vol.getName(), checkVolumeResult));
+        CheckAndRepairVolumeAnswer answer = new CheckAndRepairVolumeAnswer(command, true, checkVolumeResult);
+        answer.setVolumeCheckExecutionResult(checkVolumeResult);
+
+        return answer;
+    }
+
+    private CheckAndRepairVolumeAnswer repairVolume(KVMPhysicalDisk vol, CheckAndRepairVolumeCommand command, LibvirtComputingResource serverResource, String checkVolumeResult) {
+        EncryptFormat encryptFormat = EncryptFormat.enumValue(command.getEncryptFormat());
+        byte[] passphrase = command.getPassphrase();
+        final String repair = command.getRepair();
+
+        String repairVolumeResult = checkAndRepairVolume(vol, repair, encryptFormat, passphrase, serverResource);
+        String finalResult = (checkVolumeResult != null ? checkVolumeResult.concat(",") : "") + repairVolumeResult;
+        s_logger.info(String.format("Repair Volume result for the volume %s is %s", vol.getName(), repairVolumeResult));
+
+        CheckAndRepairVolumeAnswer answer = new CheckAndRepairVolumeAnswer(command, true, finalResult);
+        answer.setVolumeRepairExecutionResult(repairVolumeResult);
+        answer.setVolumeCheckExecutionResult(checkVolumeResult);
+
+        return answer;
+    }
+
+    private CheckAndRepairVolumeAnswer checkIfRepairLeaksIsRequired(CheckAndRepairVolumeCommand command, String checkVolumeResult, String volumeName) {
+        final String repair = command.getRepair();
+        int leaks = 0;
+        if (StringUtils.isNotEmpty(checkVolumeResult) && StringUtils.isNotEmpty(repair) && repair.equals("leaks")) {
+            ObjectMapper objectMapper = new ObjectMapper();
+            JsonNode jsonNode = null;
+            try {
+                jsonNode = objectMapper.readTree(checkVolumeResult);
+            } catch (JsonProcessingException e) {
+                String msg = String.format("Error processing response %s during check volume %s", checkVolumeResult, e.getMessage());
+                s_logger.info(msg);
+
+                return skipRepairVolumeCommand(command, checkVolumeResult, msg);
+            }
+            JsonNode leaksNode = jsonNode.get("leaks");
+            if (leaksNode != null) {
+                leaks = leaksNode.asInt();
+            }
+
+            if (leaks == 0) {
+                String msg = String.format("No leaks found while checking for the volume %s, so skipping repair", volumeName);
+                return skipRepairVolumeCommand(command, checkVolumeResult, msg);
+            }
+        }
+
+        return null;
+    }
+
+    private CheckAndRepairVolumeAnswer skipRepairVolumeCommand(CheckAndRepairVolumeCommand command, String checkVolumeResult, String msg) {
+        s_logger.info(msg);
+        String jsonStringFormat = String.format("{ \"message\": \"%s\" }", msg);
+        String finalResult = (checkVolumeResult != null ? checkVolumeResult.concat(",") : "") + jsonStringFormat;
+        CheckAndRepairVolumeAnswer answer = new CheckAndRepairVolumeAnswer(command, true, finalResult);
+        answer.setVolumeRepairExecutionResult(jsonStringFormat);
+        answer.setVolumeCheckExecutionResult(checkVolumeResult);
+
+        return answer;
+    }
+
+    protected String checkAndRepairVolume(final KVMPhysicalDisk vol, final String repair, final EncryptFormat encryptFormat, byte[] passphrase, final LibvirtComputingResource libvirtComputingResource) throws CloudRuntimeException {
+        List<QemuObject> passphraseObjects = new ArrayList<>();
+        QemuImageOptions imgOptions = null;
+        if (ArrayUtils.isEmpty(passphrase)) {
+            passphrase = null;
+        }
+        try (KeyFile keyFile = new KeyFile(passphrase)) {
+            if (passphrase != null) {
+                passphraseObjects.add(
+                        QemuObject.prepareSecretForQemuImg(vol.getFormat(), encryptFormat, keyFile.toString(), "sec0", null)
+                );
+                imgOptions = new QemuImageOptions(vol.getFormat(), vol.getPath(),"sec0");
+            }
+            QemuImg q = new QemuImg(libvirtComputingResource.getCmdsTimeout());
+            QemuImgFile file = new QemuImgFile(vol.getPath());
+            return q.checkAndRepair(file, imgOptions, passphraseObjects, repair);
+        } catch (QemuImgException | LibvirtException ex) {
+            throw new CloudRuntimeException("Failed to run qemu-img for check volume", ex);
+        } catch (IOException ex) {
+            throw new CloudRuntimeException("Failed to create keyfile for encrypted volume for check volume operation", ex);
+        }
+    }
+}

plugins/hypervisors/kvm/src/main/java/org/apache/cloudstack/utils/qemu/QemuImg.java

@@ -812,4 +812,45 @@ public class QemuImg {
         Pattern pattern = Pattern.compile("Supported\\sformats:[a-zA-Z0-9-_\\s]*?\\b" + format + "\\b", CASE_INSENSITIVE);
         return pattern.matcher(text).find();
     }
+
+    /**
+     * check for any leaks for an image and repair.
+     *
+     * @param imageOptions
+     *         Qemu style image options to be used in the checking process.
+     * @param qemuObjects
+     *         Qemu style options (e.g. for passing secrets).
+     * @param repair
+     *         Boolean option whether to repair any leaks
+     */
+    public String checkAndRepair(final QemuImgFile file, final QemuImageOptions imageOptions, final List<QemuObject> qemuObjects, final String repair) throws QemuImgException {
+        final Script script = new Script(_qemuImgPath);
+        script.add("check");
+        if (imageOptions == null) {
+            script.add(file.getFileName());
+        }
+
+        for (QemuObject o : qemuObjects) {
+            script.add(o.toCommandFlag());
+        }
+
+        if (imageOptions != null) {
+            script.add(imageOptions.toCommandFlag());
+        }
+
+        if (StringUtils.isNotEmpty(repair)) {
+            script.add("-r");
+            script.add(repair);
+        }
+
+        script.add("--output=json");
+        script.add("2>/dev/null");
+
+        final String result = Script.runBashScriptIgnoreExitValue(script.toString(), 3);
+        if (result != null) {
+            logger.debug(String.format("Check volume execution result %s", result));
+        }
+
+        return result;
+    }
 }

plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtCheckAndRepairVolumeCommandWrapperTest.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.cloud.hypervisor.kvm.resource.wrapper;
+
+import com.cloud.agent.api.storage.CheckAndRepairVolumeAnswer;
+import com.cloud.agent.api.storage.CheckAndRepairVolumeCommand;
+import com.cloud.agent.api.to.StorageFilerTO;
+import com.cloud.hypervisor.kvm.resource.LibvirtComputingResource;
+import com.cloud.hypervisor.kvm.storage.KVMPhysicalDisk;
+import com.cloud.hypervisor.kvm.storage.KVMStoragePool;
+import com.cloud.hypervisor.kvm.storage.KVMStoragePoolManager;
+import com.cloud.storage.Storage;
+
+import org.apache.cloudstack.utils.qemu.QemuImg;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.MockedConstruction;
+import org.mockito.Mockito;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class LibvirtCheckAndRepairVolumeCommandWrapperTest {
+
+    @Spy
+    LibvirtCheckAndRepairVolumeCommandWrapper libvirtCheckAndRepairVolumeCommandWrapperSpy = Mockito.spy(LibvirtCheckAndRepairVolumeCommandWrapper.class);
+
+    @Mock
+    LibvirtComputingResource libvirtComputingResourceMock;
+
+    @Mock
+    CheckAndRepairVolumeCommand checkAndRepairVolumeCommand;
+
+    @Mock
+    QemuImg qemuImgMock;
+
+    @Before
+    public void init() {
+        when(libvirtComputingResourceMock.getCmdsTimeout()).thenReturn(60);
+    }
+
+    @Test
+    public void testCheckAndRepairVolume() throws Exception {
+
+        CheckAndRepairVolumeCommand cmd = Mockito.mock(CheckAndRepairVolumeCommand.class);
+        when(cmd.getPath()).thenReturn("cbac516a-0f1f-4559-921c-1a7c6c408ccf");
+        when(cmd.getRepair()).thenReturn(null);
+        StorageFilerTO spool = Mockito.mock(StorageFilerTO.class);
+        when(cmd.getPool()).thenReturn(spool);
+
+        KVMStoragePoolManager storagePoolMgr = Mockito.mock(KVMStoragePoolManager.class);
+        when(libvirtComputingResourceMock.getStoragePoolMgr()).thenReturn(storagePoolMgr);
+        KVMStoragePool pool = Mockito.mock(KVMStoragePool.class);
+        when(spool.getType()).thenReturn(Storage.StoragePoolType.PowerFlex);
+        when(spool.getUuid()).thenReturn("b6be258b-42b8-49a4-ad51-3634ef8ff76a");
+        when(storagePoolMgr.getStoragePool(Storage.StoragePoolType.PowerFlex, "b6be258b-42b8-49a4-ad51-3634ef8ff76a")).thenReturn(pool);
+
+        KVMPhysicalDisk vol = Mockito.mock(KVMPhysicalDisk.class);
+        when(pool.getPhysicalDisk("cbac516a-0f1f-4559-921c-1a7c6c408ccf")).thenReturn(vol);
+        Mockito.when(vol.getFormat()).thenReturn(QemuImg.PhysicalDiskFormat.QCOW2);
+
+        String checkResult = "{\n" +
+                "    \"image-end-offset\": 6442582016,\n" +
+                "    \"total-clusters\": 163840,\n" +
+                "    \"check-errors\": 0,\n" +
+                "    \"leaks\": 124,\n" +
+                "    \"allocated-clusters\": 98154,\n" +
+                "    \"filename\": \"/var/lib/libvirt/images/26be20c7-b9d0-43f6-a76e-16c70737a0e0\",\n" +
+                "    \"format\": \"qcow2\",\n" +
+                "    \"fragmented-clusters\": 96135\n" +
+                "}";
+
+        try (MockedConstruction<QemuImg> ignored = Mockito.mockConstruction(QemuImg.class, (mock, context) -> {
+            when(mock.checkAndRepair(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any())).thenReturn(checkResult);
+        })) {
+            CheckAndRepairVolumeAnswer result = (CheckAndRepairVolumeAnswer) libvirtCheckAndRepairVolumeCommandWrapperSpy.execute(cmd, libvirtComputingResourceMock);
+            Assert.assertEquals(checkResult, result.getVolumeCheckExecutionResult());
+        }
+    }
+
+}

plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/storage/LibvirtStoragePoolTest.java

@@ -87,6 +87,9 @@ public class LibvirtStoragePoolTest extends TestCase {
         StoragePool storage = Mockito.mock(StoragePool.class);
 
         LibvirtStoragePool nfsPool = new LibvirtStoragePool(uuid, name, StoragePoolType.NetworkFilesystem, adapter, storage);
+        if (nfsPool.getType() != StoragePoolType.NetworkFilesystem) {
+            System.out.println("tested");
+        }
         assertFalse(nfsPool.isExternalSnapshot());
 
         LibvirtStoragePool rbdPool = new LibvirtStoragePool(uuid, name, StoragePoolType.RBD, adapter, storage);

plugins/hypervisors/kvm/src/test/java/org/apache/cloudstack/utils/qemu/QemuImgTest.java

@@ -368,4 +368,21 @@ public class QemuImgTest {
         Assert.assertTrue("should support qcow2", QemuImg.helpSupportsImageFormat(partialHelp, PhysicalDiskFormat.QCOW2));
         Assert.assertFalse("should not support http", QemuImg.helpSupportsImageFormat(partialHelp, PhysicalDiskFormat.SHEEPDOG));
     }
+
+    @Test
+    public void testCheckAndRepair() throws LibvirtException {
+        String filename = "/tmp/" + UUID.randomUUID() + ".qcow2";
+
+        QemuImgFile file = new QemuImgFile(filename);
+
+        try {
+            QemuImg qemu = new QemuImg(0);
+            qemu.checkAndRepair(file, null, null, null);
+        } catch (QemuImgException e) {
+            fail(e.getMessage());
+        }
+
+        File f = new File(filename);
+        f.delete();
+    }
 }

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

@@ -552,6 +552,7 @@ 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.CheckAndRepairVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.DeleteVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.DestroyVolumeCmd;
@@ -3704,6 +3705,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
         cmdList.add(ListVMGroupsCmd.class);
         cmdList.add(UpdateVMGroupCmd.class);
         cmdList.add(AttachVolumeCmd.class);
+        cmdList.add(CheckAndRepairVolumeCmd.class);
         cmdList.add(CreateVolumeCmd.class);
         cmdList.add(DeleteVolumeCmd.class);
         cmdList.add(UpdateVolumeCmd.class);

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

@@ -0,0 +1,41 @@
+// 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.storage;
+
+public class CheckAndRepairVolumePayload {
+
+    public final String repair;
+    public String result;
+
+    public CheckAndRepairVolumePayload(String repair) {
+        this.repair = repair;
+    }
+
+    public String getRepair() {
+        return repair;
+    }
+
+    public String getResult() {
+        return result;
+    }
+
+    public void setResult(String result) {
+        this.result = result;
+    }
+
+}

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

@@ -42,6 +42,7 @@ import org.apache.cloudstack.api.ServerApiException;
 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.CheckAndRepairVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.DetachVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.ExtractVolumeCmd;
@@ -217,6 +218,7 @@ import com.cloud.vm.VirtualMachineManager;
 import com.cloud.vm.VmDetailConstants;
 import com.cloud.vm.VmWork;
 import com.cloud.vm.VmWorkAttachVolume;
+import com.cloud.vm.VmWorkCheckAndRepairVolume;
 import com.cloud.vm.VmWorkConstants;
 import com.cloud.vm.VmWorkDetachVolume;
 import com.cloud.vm.VmWorkExtractVolume;
@@ -379,6 +381,9 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
     public static ConfigKey<Long> storageTagRuleExecutionTimeout = new ConfigKey<>("Advanced", Long.class, "storage.tag.rule.execution.timeout", "2000", "The maximum runtime,"
             + " in milliseconds, to execute a storage tag rule; if it is reached, a timeout will happen.", true);
 
+    public static final ConfigKey<Boolean> AllowCheckAndRepairVolume = new ConfigKey<Boolean>("Advanced", Boolean.class, "volume.check.and.repair.leaks.before.use", "false",
+            "To check and repair the volume if it has any leaks before performing volume attach or VM start operations", true, ConfigKey.Scope.StoragePool);
+
     private final StateMachine2<Volume.State, Volume.Event, Volume> _volStateMachine;
 
     private static final Set<Volume.State> STATES_VOLUME_CANNOT_BE_DESTROYED = new HashSet<>(Arrays.asList(Volume.State.Destroy, Volume.State.Expunging, Volume.State.Expunged, Volume.State.Allocated));
@@ -1335,7 +1340,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     outcome.get();
                 } catch (InterruptedException e) {
                     throw new RuntimeException("Operation was interrupted", e);
-                } catch (java.util.concurrent.ExecutionException e) {
+                } catch (ExecutionException e) {
                     throw new RuntimeException("Execution exception", e);
                 }
 
@@ -1817,7 +1822,158 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         s_logger.debug(String.format("Volume [%s] has been successfully recovered, thus a new usage event %s has been published.", volume.getUuid(), EventTypes.EVENT_VOLUME_CREATE));
     }
 
+    @Override
+    @ActionEvent(eventType = EventTypes.EVENT_VOLUME_CHECK, eventDescription = "checking volume and repair if needed", async = true)
+    public Pair<String, String> checkAndRepairVolume(CheckAndRepairVolumeCmd cmd) throws ResourceAllocationException {
+        long volumeId = cmd.getId();
+        String repair = cmd.getRepair();
 
+        final VolumeVO volume = _volsDao.findById(volumeId);
+        validationsForCheckVolumeOperation(volume);
+
+        Long vmId = volume.getInstanceId();
+        if (vmId != null) {
+            // serialize VM operation
+            return handleCheckAndRepairVolumeJob(vmId, volumeId, repair);
+        } else {
+            return handleCheckAndRepairVolume(volumeId, repair);
+        }
+    }
+
+    private Pair<String, String> handleCheckAndRepairVolume(Long volumeId, String repair) {
+        CheckAndRepairVolumePayload payload = new CheckAndRepairVolumePayload(repair);
+        VolumeInfo volumeInfo = volFactory.getVolume(volumeId);
+        volumeInfo.addPayload(payload);
+
+        Pair<String, String> result = volService.checkAndRepairVolume(volumeInfo);
+        return result;
+    }
+
+    private Pair<String, String> handleCheckAndRepairVolumeJob(Long vmId, Long volumeId, String repair) throws ResourceAllocationException {
+        AsyncJobExecutionContext jobContext = AsyncJobExecutionContext.getCurrentExecutionContext();
+        if (jobContext.isJobDispatchedBy(VmWorkConstants.VM_WORK_JOB_DISPATCHER)) {
+            // avoid re-entrance
+            VmWorkJobVO placeHolder = null;
+            placeHolder = createPlaceHolderWork(vmId);
+            try {
+                Pair<String, String> result = orchestrateCheckAndRepairVolume(volumeId, repair);
+                return result;
+            } finally {
+                _workJobDao.expunge(placeHolder.getId());
+            }
+        } else {
+            Outcome<Pair> outcome = checkAndRepairVolumeThroughJobQueue(vmId, volumeId, repair);
+            try {
+                outcome.get();
+            } catch (InterruptedException e) {
+                throw new RuntimeException("Operation is interrupted", e);
+            } catch (ExecutionException e) {
+                throw new RuntimeException("Execution exception--", e);
+            }
+
+            Object jobResult = _jobMgr.unmarshallResultObject(outcome.getJob());
+            if (jobResult != null) {
+                if (jobResult instanceof ConcurrentOperationException) {
+                    throw (ConcurrentOperationException)jobResult;
+                } else if (jobResult instanceof ResourceAllocationException) {
+                    throw (ResourceAllocationException)jobResult;
+                } else if (jobResult instanceof Throwable) {
+                    throw new RuntimeException("Unexpected exception", (Throwable)jobResult);
+                }
+            }
+
+            // retrieve the entity url from job result
+            if (jobResult != null && jobResult instanceof Pair) {
+                return (Pair<String, String>) jobResult;
+            }
+
+            return null;
+        }
+    }
+
+    protected void validationsForCheckVolumeOperation(VolumeVO volume) {
+        Account caller = CallContext.current().getCallingAccount();
+        _accountMgr.checkAccess(caller, null, true, volume);
+
+        String volumeName = volume.getName();
+        Long vmId = volume.getInstanceId();
+        if (vmId != null) {
+            validateVMforCheckVolumeOperation(vmId, volumeName);
+        }
+
+        if (volume.getState() != Volume.State.Ready) {
+            throw new InvalidParameterValueException(String.format("Volume: %s is not in Ready state", volumeName));
+        }
+
+        HypervisorType hypervisorType = _volsDao.getHypervisorType(volume.getId());
+        if (!HypervisorType.KVM.equals(hypervisorType)) {
+            throw new InvalidParameterValueException(String.format("Check and Repair volumes is supported only for KVM hypervisor"));
+        }
+
+        if (!Arrays.asList(ImageFormat.QCOW2, ImageFormat.VDI).contains(volume.getFormat())) {
+            throw new InvalidParameterValueException("Volume format is not supported for checking and repair");
+        }
+    }
+
+    private void validateVMforCheckVolumeOperation(Long vmId, String volumeName) {
+        Account caller = CallContext.current().getCallingAccount();
+        UserVmVO vm = _userVmDao.findById(vmId);
+        if (vm == null) {
+            throw new InvalidParameterValueException(String.format("VM not found, please check the VM to which this volume %s is attached", volumeName));
+        }
+
+        _accountMgr.checkAccess(caller, null, true, vm);
+
+        if (vm.getState() != State.Stopped) {
+            throw new InvalidParameterValueException(String.format("VM to which the volume %s is attached should be in stopped state", volumeName));
+        }
+    }
+
+    private Pair<String, String> orchestrateCheckAndRepairVolume(Long volumeId, String repair) {
+
+        VolumeInfo volume = volFactory.getVolume(volumeId);
+
+        if (volume == null) {
+            throw new InvalidParameterValueException("Checking volume and repairing failed due to volume:" + volumeId + " doesn't exist");
+        }
+
+        CheckAndRepairVolumePayload payload = new CheckAndRepairVolumePayload(repair);
+        volume.addPayload(payload);
+
+        return volService.checkAndRepairVolume(volume);
+    }
+
+    public Outcome<Pair> checkAndRepairVolumeThroughJobQueue(final Long vmId, final Long volumeId, String repair) {
+
+        final CallContext context = CallContext.current();
+        final User callingUser = context.getCallingUser();
+        final Account callingAccount = context.getCallingAccount();
+
+        final VMInstanceVO vm = _vmInstanceDao.findById(vmId);
+
+        VmWorkJobVO workJob = new VmWorkJobVO(context.getContextId());
+
+        workJob.setDispatcher(VmWorkConstants.VM_WORK_JOB_DISPATCHER);
+        workJob.setCmd(VmWorkCheckAndRepairVolume.class.getName());
+
+        workJob.setAccountId(callingAccount.getId());
+        workJob.setUserId(callingUser.getId());
+        workJob.setStep(VmWorkJobVO.Step.Starting);
+        workJob.setVmType(VirtualMachine.Type.Instance);
+        workJob.setVmInstanceId(vm.getId());
+        workJob.setRelated(AsyncJobExecutionContext.getOriginJobId());
+
+        // save work context info (there are some duplications)
+        VmWorkCheckAndRepairVolume workInfo = new VmWorkCheckAndRepairVolume(callingUser.getId(), callingAccount.getId(), vm.getId(),
+                VolumeApiServiceImpl.VM_WORK_JOB_HANDLER, volumeId, repair);
+        workJob.setCmdInfo(VmWorkSerializer.serialize(workInfo));
+
+        _jobMgr.submitAsyncJob(workJob, VmWorkConstants.VM_WORK_QUEUE, vm.getId());
+
+        AsyncJobExecutionContext.getCurrentExecutionContext().joinJob(workJob.getId());
+
+        return new VmJobCheckAndRepairVolumeOutcome(workJob);
+    }
 
     @Override
     @ActionEvent(eventType = EventTypes.EVENT_VOLUME_CHANGE_DISK_OFFERING, eventDescription = "Changing disk offering of a volume")
@@ -1987,7 +2143,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     outcome.get();
                 } catch (InterruptedException e) {
                     throw new RuntimeException("Operation was interrupted", e);
-                } catch (java.util.concurrent.ExecutionException e) {
+                } catch (ExecutionException e) {
                     throw new RuntimeException("Execution exception", e);
                 }
 
@@ -2773,7 +2929,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                 outcome.get();
             } catch (InterruptedException e) {
                 throw new RuntimeException("Operation is interrupted", e);
-            } catch (java.util.concurrent.ExecutionException e) {
+            } catch (ExecutionException e) {
                 throw new RuntimeException("Execution excetion", e);
             }
 
@@ -3181,7 +3337,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     outcome.get();
                 } catch (InterruptedException e) {
                     throw new RuntimeException("Operation is interrupted", e);
-                } catch (java.util.concurrent.ExecutionException e) {
+                } catch (ExecutionException e) {
                     throw new RuntimeException("Execution excetion", e);
                 }
 
@@ -3510,7 +3666,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     outcome.get();
                 } catch (InterruptedException e) {
                     throw new RuntimeException("Operation is interrupted", e);
-                } catch (java.util.concurrent.ExecutionException e) {
+                } catch (ExecutionException e) {
                     throw new RuntimeException("Execution excetion", e);
                 }
 
@@ -3827,7 +3983,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                     outcome.get();
                 } catch (InterruptedException e) {
                     throw new RuntimeException("Operation is interrupted", e);
-                } catch (java.util.concurrent.ExecutionException e) {
+                } catch (ExecutionException e) {
                     throw new RuntimeException("Execution excetion", e);
                 }
 
@@ -4251,6 +4407,12 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         try {
             // if we don't have a host, the VM we are attaching the disk to has never been started before
             if (host != null) {
+                try {
+                    volService.checkAndRepairVolumeBasedOnConfig(volFactory.getVolume(volumeToAttach.getId()), host);
+                } catch (Exception e) {
+                    s_logger.debug(String.format("Unable to check and repair volume [%s] on host [%s], due to %s.", volumeToAttach.getName(), host, e.getMessage()));
+                }
+
                 try {
                     volService.grantAccess(volFactory.getVolume(volumeToAttach.getId()), host, dataStore);
                 } catch (Exception e) {
@@ -4596,6 +4758,24 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         }
     }
 
+    public class VmJobCheckAndRepairVolumeOutcome extends OutcomeImpl<Pair> {
+
+        public VmJobCheckAndRepairVolumeOutcome(final AsyncJob job) {
+            super(Pair.class, job, VmJobCheckInterval.value(), new Predicate() {
+                @Override
+                public boolean checkCondition() {
+                    AsyncJobVO jobVo = _entityMgr.findById(AsyncJobVO.class, job.getId());
+                    assert (jobVo != null);
+                    if (jobVo == null || jobVo.getStatus() != JobInfo.Status.IN_PROGRESS) {
+                        return true;
+                    }
+
+                    return false;
+                }
+            }, AsyncJob.Topics.JOB_STATE);
+        }
+    }
+
     public Outcome<Volume> attachVolumeToVmThroughJobQueue(final Long vmId, final Long volumeId, final Long deviceId) {
 
         final CallContext context = CallContext.current();
@@ -4833,6 +5013,13 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, _jobMgr.marshallResultObject(work.getSnapshotId()));
     }
 
+    @ReflectionUse
+    private Pair<JobInfo.Status, String> orchestrateCheckAndRepairVolume(VmWorkCheckAndRepairVolume work) throws Exception {
+        Account account = _accountDao.findById(work.getAccountId());
+        Pair<String, String> result = orchestrateCheckAndRepairVolume(work.getVolumeId(), work.getRepair());
+        return new Pair<JobInfo.Status, String>(JobInfo.Status.SUCCEEDED, _jobMgr.marshallResultObject(result));
+    }
+
     @Override
     public Pair<JobInfo.Status, String> handleVmWorkJob(VmWork work) throws Exception {
         return _jobHandlerProxy.handleVmWorkJob(work);
@@ -4869,7 +5056,8 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                 AllowUserExpungeRecoverVolume,
                 MatchStoragePoolTagsWithDiskOffering,
                 UseHttpsToUpload,
-                WaitDetachDevice
+                WaitDetachDevice,
+                AllowCheckAndRepairVolume
         };
     }
 }

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

@@ -25,6 +25,7 @@ import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -39,6 +40,7 @@ import java.util.concurrent.ExecutionException;
 
 import org.apache.cloudstack.acl.ControlledEntity;
 import org.apache.cloudstack.acl.SecurityChecker.AccessType;
+import org.apache.cloudstack.api.command.user.volume.CheckAndRepairVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.CreateVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.DetachVolumeCmd;
 import org.apache.cloudstack.api.command.user.volume.MigrateVolumeCmd;
@@ -1645,7 +1647,6 @@ public class VolumeApiServiceImplTest {
             Mockito.when(_diskOfferingDao.findById(1L)).thenReturn(diskOffering);
 
             StoragePoolVO srcStoragePoolVOMock = Mockito.mock(StoragePoolVO.class);
-            StoragePool destPool = Mockito.mock(StoragePool.class);
             PrimaryDataStore dataStore = Mockito.mock(PrimaryDataStore.class);
 
             Mockito.when(vol.getPassphraseId()).thenReturn(1L);
@@ -1660,4 +1661,166 @@ public class VolumeApiServiceImplTest {
             // test passed
         }
     }
+
+    @Test
+    public void testValidationsForCheckVolumeAPI() {
+        VolumeVO volume = mock(VolumeVO.class);
+
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(1L);
+        UserVmVO vm = mock(UserVmVO.class);
+        when(userVmDaoMock.findById(1L)).thenReturn(vm);
+        when(vm.getState()).thenReturn(State.Stopped);
+        when(volume.getState()).thenReturn(Volume.State.Ready);
+        when(volume.getId()).thenReturn(1L);
+        when(volumeDaoMock.getHypervisorType(1L)).thenReturn(HypervisorType.KVM);
+        when(volume.getFormat()).thenReturn(Storage.ImageFormat.QCOW2);
+
+        volumeApiServiceImpl.validationsForCheckVolumeOperation(volume);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testValidationsForCheckVolumeAPIWithRunningVM() {
+        VolumeVO volume = mock(VolumeVO.class);
+
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(1L);
+        UserVmVO vm = mock(UserVmVO.class);
+        when(userVmDaoMock.findById(1L)).thenReturn(vm);
+        when(vm.getState()).thenReturn(State.Running);
+
+        volumeApiServiceImpl.validationsForCheckVolumeOperation(volume);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testValidationsForCheckVolumeAPIWithNonexistedVM() {
+        VolumeVO volume = mock(VolumeVO.class);
+
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(1L);
+        when(userVmDaoMock.findById(1L)).thenReturn(null);
+
+        volumeApiServiceImpl.validationsForCheckVolumeOperation(volume);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testValidationsForCheckVolumeAPIWithAllocatedVolume() {
+        VolumeVO volume = mock(VolumeVO.class);
+
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(1L);
+        UserVmVO vm = mock(UserVmVO.class);
+        when(userVmDaoMock.findById(1L)).thenReturn(vm);
+        when(vm.getState()).thenReturn(State.Stopped);
+        when(volume.getState()).thenReturn(Volume.State.Allocated);
+
+        volumeApiServiceImpl.validationsForCheckVolumeOperation(volume);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testValidationsForCheckVolumeAPIWithNonKVMhypervisor() {
+        VolumeVO volume = mock(VolumeVO.class);
+
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(1L);
+        UserVmVO vm = mock(UserVmVO.class);
+        when(userVmDaoMock.findById(1L)).thenReturn(vm);
+        when(vm.getState()).thenReturn(State.Stopped);
+        when(volume.getState()).thenReturn(Volume.State.Ready);
+        when(volume.getId()).thenReturn(1L);
+        when(volumeDaoMock.getHypervisorType(1L)).thenReturn(HypervisorType.VMware);
+
+        volumeApiServiceImpl.validationsForCheckVolumeOperation(volume);
+    }
+
+    @Test
+    public void testCheckAndRepairVolume() throws ResourceAllocationException {
+
+        CheckAndRepairVolumeCmd cmd  = mock(CheckAndRepairVolumeCmd.class);
+        when(cmd.getId()).thenReturn(1L);
+        when(cmd.getRepair()).thenReturn(null);
+
+        VolumeVO volume = mock(VolumeVO.class);
+        when(volumeDaoMock.findById(1L)).thenReturn(volume);
+
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(null);
+        when(volume.getState()).thenReturn(Volume.State.Ready);
+        when(volume.getId()).thenReturn(1L);
+        when(volumeDaoMock.getHypervisorType(1L)).thenReturn(HypervisorType.KVM);
+
+        VolumeInfo volumeInfo = mock(VolumeInfo.class);
+        when(volumeDataFactoryMock.getVolume(1L)).thenReturn(volumeInfo);
+
+        String checkResult = "{\n" +
+                "    \"image-end-offset\": 6442582016,\n" +
+                "    \"total-clusters\": 163840,\n" +
+                "    \"check-errors\": 0,\n" +
+                "    \"leaks\": 124,\n" +
+                "    \"allocated-clusters\": 98154,\n" +
+                "    \"filename\": \"/var/lib/libvirt/images/26be20c7-b9d0-43f6-a76e-16c70737a0e0\",\n" +
+                "    \"format\": \"qcow2\",\n" +
+                "    \"fragmented-clusters\": 96135\n" +
+                "}";
+
+        String repairResult = null;
+        Pair<String, String> result = new Pair<>(checkResult, repairResult);
+        when(volumeServiceMock.checkAndRepairVolume(volumeInfo)).thenReturn(result);
+        when(volume.getFormat()).thenReturn(Storage.ImageFormat.QCOW2);
+
+        Pair<String, String> finalresult = volumeApiServiceImpl.checkAndRepairVolume(cmd);
+
+        Assert.assertEquals(result, finalresult);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void testValidationsForCheckVolumeAPIWithInvalidVolumeFormat() {
+        VolumeVO volume = mock(VolumeVO.class);
+        AccountVO account = new AccountVO("admin", 1L, "networkDomain", Account.Type.NORMAL, "uuid");
+        UserVO user = new UserVO(1, "testuser", "password", "firstname", "lastName", "email", "timezone", UUID.randomUUID().toString(), User.Source.UNKNOWN);
+        CallContext.register(user, account);
+
+        lenient().doNothing().when(accountManagerMock).checkAccess(any(Account.class), any(AccessType.class), any(Boolean.class), any(ControlledEntity.class));
+
+        when(volume.getInstanceId()).thenReturn(1L);
+        UserVmVO vm = mock(UserVmVO.class);
+        when(userVmDaoMock.findById(1L)).thenReturn(vm);
+        when(vm.getState()).thenReturn(State.Stopped);
+        when(volume.getState()).thenReturn(Volume.State.Ready);
+        when(volume.getId()).thenReturn(1L);
+        when(volumeDaoMock.getHypervisorType(1L)).thenReturn(HypervisorType.KVM);
+        when(volume.getFormat()).thenReturn(Storage.ImageFormat.RAW);
+
+        volumeApiServiceImpl.validationsForCheckVolumeOperation(volume);
+    }
 }

utils/src/main/java/com/cloud/utils/StringUtils.java

@@ -19,6 +19,10 @@
 
 package com.cloud.utils;
 
+import com.cloud.utils.exception.CloudRuntimeException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -282,4 +286,22 @@ public class StringUtils {
         final String value = keyValuePair.substring(index + 1);
         return new Pair<>(key.trim(), value.trim());
     }
+
+    public static Map<String, String> parseJsonToMap(String jsonString) {
+        ObjectMapper objectMapper = new ObjectMapper();
+        Map<String, String> mapResult = new HashMap<>();
+
+        if (org.apache.commons.lang3.StringUtils.isNotBlank(jsonString)) {
+            try {
+                JsonNode jsonNode = objectMapper.readTree(jsonString);
+                jsonNode.fields().forEachRemaining(entry -> {
+                    mapResult.put(entry.getKey(), entry.getValue().asText());
+                });
+            } catch (Exception e) {
+                throw new CloudRuntimeException("Error while parsing json to convert it to map " + e.getMessage());
+            }
+        }
+
+        return mapResult;
+    }
 }

utils/src/main/java/com/cloud/utils/script/Script.java

@@ -330,6 +330,118 @@ public class Script implements Callable<String> {
         }
     }
 
+    public String executeIgnoreExitValue(OutputInterpreter interpreter, int exitValue) {
+        String[] command = _command.toArray(new String[_command.size()]);
+
+        if (_logger.isDebugEnabled()) {
+            _logger.debug(String.format("Executing: %s", buildCommandLine(command).split(KeyStoreUtils.KS_FILENAME)[0]));
+        }
+
+        try {
+            ProcessBuilder pb = new ProcessBuilder(command);
+            pb.redirectErrorStream(true);
+            if (_workDir != null)
+                pb.directory(new File(_workDir));
+
+            _process = pb.start();
+            if (_process == null) {
+                _logger.warn(String.format("Unable to execute: %s", buildCommandLine(command)));
+                return String.format("Unable to execute the command: %s", command[0]);
+            }
+
+            BufferedReader ir = new BufferedReader(new InputStreamReader(_process.getInputStream()));
+
+            _thread = Thread.currentThread();
+            ScheduledFuture<String> future = null;
+            if (_timeout > 0) {
+                future = s_executors.schedule(this, _timeout, TimeUnit.MILLISECONDS);
+            }
+
+            Task task = null;
+            if (interpreter != null && interpreter.drain()) {
+                task = new Task(interpreter, ir);
+                s_executors.execute(task);
+            }
+
+            while (true) {
+                _logger.debug(String.format("Executing while with timeout : %d", _timeout));
+                try {
+                    //process execution completed within timeout period
+                    if (_process.waitFor(_timeout, TimeUnit.MILLISECONDS)) {
+                        //process completed successfully
+                        if (_process.exitValue() == 0 || _process.exitValue() == exitValue) {
+                            _logger.debug("Execution is successful.");
+                            if (interpreter != null) {
+                                return interpreter.drain() ? task.getResult() : interpreter.interpret(ir);
+                            } else {
+                                // null return exitValue apparently
+                                return String.valueOf(_process.exitValue());
+                            }
+                        } else { //process failed
+                            break;
+                        }
+                    } //timeout
+                } catch (InterruptedException e) {
+                    if (!_isTimeOut) {
+                        /*
+                         * This is not timeout, we are interrupted by others,
+                         * continue
+                         */
+                        _logger.debug("We are interrupted but it's not a timeout, just continue");
+                        continue;
+                    }
+                } finally {
+                    if (future != null) {
+                        future.cancel(false);
+                    }
+                    Thread.interrupted();
+                }
+
+                //timeout without completing the process
+                TimedOutLogger log = new TimedOutLogger(_process);
+                Task timedoutTask = new Task(log, ir);
+
+                timedoutTask.run();
+                if (!_passwordCommand) {
+                    _logger.warn(String.format("Timed out: %s.  Output is: %s", buildCommandLine(command), timedoutTask.getResult()));
+                } else {
+                    _logger.warn(String.format("Timed out: %s", buildCommandLine(command)));
+                }
+
+                return ERR_TIMEOUT;
+            }
+
+            _logger.debug(String.format("Exit value is %d", _process.exitValue()));
+
+            BufferedReader reader = new BufferedReader(new InputStreamReader(_process.getInputStream()), 128);
+
+            String error;
+            if (interpreter != null) {
+                error = interpreter.processError(reader);
+            } else {
+                error = String.valueOf(_process.exitValue());
+            }
+
+            if (_logger.isDebugEnabled()) {
+                _logger.debug(error);
+            }
+            return error;
+        } catch (SecurityException ex) {
+            _logger.warn("Security Exception....not running as root?", ex);
+            return stackTraceAsString(ex);
+        } catch (Exception ex) {
+            _logger.warn(String.format("Exception: %s", buildCommandLine(command)), ex);
+            return stackTraceAsString(ex);
+        } finally {
+            if (_process != null) {
+                IOUtils.closeQuietly(_process.getErrorStream());
+                IOUtils.closeQuietly(_process.getOutputStream());
+                IOUtils.closeQuietly(_process.getInputStream());
+                _process.destroyForcibly();
+            }
+        }
+    }
+
     @Override
     public String call() {
         try {
@@ -563,4 +675,24 @@ public class Script implements Callable<String> {
         }
     }
 
+    public static String runBashScriptIgnoreExitValue(String command, int exitValue) {
+        return runBashScriptIgnoreExitValue(command, exitValue, 0);
+    }
+
+    public static String runBashScriptIgnoreExitValue(String command, int exitValue, int timeout) {
+
+        Script s = new Script("/bin/bash", timeout);
+        s.add("-c");
+        s.add(command);
+
+        OutputInterpreter.AllLinesParser parser = new OutputInterpreter.AllLinesParser();
+        s.executeIgnoreExitValue(parser, exitValue);
+
+        String result = parser.getLines();
+        if (result == null || result.trim().isEmpty()) {
+            return null;
+        } else {
+            return result.trim();
+        }
+    }
 }