CLOUDSTACK-6191 Add support for specifying volume provisioning

type (thin, sparse, fat) in disk/compute offerings.

Submitted-by: Yoshikazu Nojima <mail@ynojima.net>
Reviewed-by: Marcus Sorensen, Mike Tutowski

api/src/com/cloud/offering/DiskOffering.java

@@ -22,6 +22,8 @@ import org.apache.cloudstack.acl.InfrastructureEntity;
 import org.apache.cloudstack.api.Identity;
 import org.apache.cloudstack.api.InternalIdentity;
 
+import com.cloud.storage.Storage.ProvisioningType;
+
 /**
  * Represents a disk offering that specifies what the end user needs in
  * the disk offering.
@@ -61,6 +63,8 @@ public interface DiskOffering extends InfrastructureEntity, Identity, InternalId
 
     String getDisplayText();
 
+    public ProvisioningType getProvisioningType();
+
     public String getTags();
 
     public String[] getTagsArray();

api/src/com/cloud/storage/Storage.java

@@ -16,6 +16,8 @@
 // under the License.
 package com.cloud.storage;
 
+import org.apache.commons.lang.NotImplementedException;
+
 import java.util.ArrayList;
 import java.util.List;
 
@@ -32,27 +34,27 @@ public class Storage {
         VDI(true, true, false, "vdi"),
         TAR(false, false, false, "tar");
 
-        private final boolean thinProvisioned;
+        private final boolean supportThinProvisioning;
         private final boolean supportSparse;
         private final boolean supportSnapshot;
         private final String fileExtension;
 
-        private ImageFormat(boolean thinProvisioned, boolean supportSparse, boolean supportSnapshot) {
+        private ImageFormat(boolean supportThinProvisioning, boolean supportSparse, boolean supportSnapshot) {
-            this.thinProvisioned = thinProvisioned;
+            this.supportThinProvisioning = supportThinProvisioning;
             this.supportSparse = supportSparse;
             this.supportSnapshot = supportSnapshot;
             fileExtension = null;
         }
 
-        private ImageFormat(boolean thinProvisioned, boolean supportSparse, boolean supportSnapshot, String fileExtension) {
+        private ImageFormat(boolean supportThinProvisioning, boolean supportSparse, boolean supportSnapshot, String fileExtension) {
-            this.thinProvisioned = thinProvisioned;
+            this.supportThinProvisioning = supportThinProvisioning;
             this.supportSparse = supportSparse;
             this.supportSnapshot = supportSnapshot;
             this.fileExtension = fileExtension;
         }
 
-        public boolean isThinProvisioned() {
+        public boolean supportThinProvisioning() {
-            return thinProvisioned;
+            return supportThinProvisioning;
         }
 
         public boolean supportsSparse() {
@@ -72,6 +74,35 @@ public class Storage {
 
     }
 
+    public static enum ProvisioningType {
+        THIN("thin"),
+        SPARSE("sparse"),
+        FAT("fat");
+
+        private final String provisionType;
+
+        private ProvisioningType(String provisionType){
+            this.provisionType = provisionType;
+        }
+
+        public String toString(){
+            return this.provisionType;
+        }
+
+        public static ProvisioningType getProvisioningType(String provisioningType){
+
+            if(provisioningType.equals(THIN.provisionType)){
+                return ProvisioningType.THIN;
+            } else if(provisioningType.equals(SPARSE.provisionType)){
+                return ProvisioningType.SPARSE;
+            } else if (provisioningType.equals(FAT.provisionType)){
+                return ProvisioningType.FAT;
+            } else{
+                    throw new NotImplementedException();
+            }
+        }
+    }
+
     public static enum FileSystem {
         Unknown, ext3, ntfs, fat, fat32, ext2, ext4, cdfs, hpfs, ufs, hfs, hfsp
     }

api/src/com/cloud/storage/Volume.java

@@ -187,6 +187,8 @@ public interface Volume extends ControlledEntity, Identity, InternalIdentity, Ba
 
     Storage.ImageFormat getFormat();
 
+    Storage.ProvisioningType getProvisioningType();
+
     Long getVmSnapshotChainSize();
 
     Integer getHypervisorSnapshotReserve();

api/src/com/cloud/vm/DiskProfile.java

@@ -18,6 +18,7 @@ package com.cloud.vm;
 
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.offering.DiskOffering;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.Volume;
 
 /**
@@ -35,6 +36,7 @@ public class DiskProfile {
     private Long templateId;
     private long volumeId;
     private String path;
+    private ProvisioningType provisioningType;
     private Long bytesReadRate;
     private Long bytesWriteRate;
     private Long iopsReadRate;
@@ -165,6 +167,14 @@ public class DiskProfile {
         return this.path;
     }
 
+    public void setProvisioningType(ProvisioningType provisioningType){
+        this.provisioningType = provisioningType;
+    }
+
+    public ProvisioningType getProvisioningType(){
+        return this.provisioningType;
+    }
+
     public void setSize(long size) {
         this.size = size;
     }

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

@@ -195,6 +195,7 @@ public class ApiConstants {
     public static final String PRIVATE_END_PORT = "privateendport";
     public static final String PRIVATE_ZONE = "privatezone";
     public static final String PROTOCOL = "protocol";
+    public static final String PROVISIONINGTYPE = "provisioningtype";
     public static final String PUBLIC_INTERFACE = "publicinterface";
     public static final String PUBLIC_IP_ID = "publicipid";
     public static final String PUBLIC_IP = "publicip";

api/src/org/apache/cloudstack/api/command/admin/offering/CreateDiskOfferingCmd.java

@@ -27,6 +27,7 @@ import org.apache.cloudstack.api.ServerApiException;
 import org.apache.cloudstack.api.response.DiskOfferingResponse;
 import org.apache.cloudstack.api.response.DomainResponse;
 
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.offering.DiskOffering;
 import com.cloud.offering.ServiceOffering;
 import com.cloud.user.Account;
@@ -66,6 +67,11 @@ public class CreateDiskOfferingCmd extends BaseCmd {
     @Parameter(name = ApiConstants.STORAGE_TYPE, type = CommandType.STRING, description = "the storage type of the disk offering. Values are local and shared.")
     private String storageType = ServiceOffering.StorageType.shared.toString();
 
+    @Parameter(name = ApiConstants.PROVISIONINGTYPE,
+            type = CommandType.STRING,
+            description = "provisioning type used to create volumes. Valid values are thin, sparse, fat.")
+    private String provisioningType = ProvisioningType.THIN.toString();
+
     @Parameter(name = ApiConstants.DISPLAY_OFFERING,
             type = CommandType.BOOLEAN,
             description = "an optional field, whether to display the offering to the end user or not.")
@@ -158,6 +164,10 @@ public class CreateDiskOfferingCmd extends BaseCmd {
         return storageType;
     }
 
+    public String getProvisioningType(){
+        return provisioningType;
+    }
+
     public Boolean getDisplayOffering() {
         return displayOffering;
     }

api/src/org/apache/cloudstack/api/command/admin/offering/CreateServiceOfferingCmd.java

@@ -21,6 +21,7 @@ import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
 
+import com.cloud.storage.Storage;
 import org.apache.cloudstack.api.APICommand;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.api.ApiErrorCode;
@@ -53,6 +54,9 @@ public class CreateServiceOfferingCmd extends BaseCmd {
     @Parameter(name = ApiConstants.DISPLAY_TEXT, type = CommandType.STRING, required = true, description = "the display text of the service offering")
     private String displayText;
 
+    @Parameter(name = ApiConstants.PROVISIONINGTYPE, type = CommandType.STRING, description = "provisioning type used to create volumes. Valid values are thin, sparse, fat.")
+    private String provisioningType = Storage.ProvisioningType.THIN.toString();
+
     @Parameter(name = ApiConstants.MEMORY, type = CommandType.INTEGER, required = false, description = "the total memory of the service offering in MB")
     private Integer memory;
 
@@ -150,6 +154,10 @@ public class CreateServiceOfferingCmd extends BaseCmd {
         return displayText;
     }
 
+    public String getProvisioningType(){
+        return provisioningType;
+    }
+
     public Integer getMemory() {
         return memory;
     }

api/src/org/apache/cloudstack/api/response/DiskOfferingResponse.java

@@ -85,6 +85,9 @@ public class DiskOfferingResponse extends BaseResponse {
     @Param(description = "the storage type for this disk offering")
     private String storageType;
 
+    @SerializedName("provisioningtype") @Param(description="provisioning type used to create volumes. Valid values are thin, sparse, fat.", since = "4.4.0")
+    private String provisioningType;
+
     @SerializedName("diskBytesReadRate")
     @Param(description = "bytes read rate of the disk offering")
     private Long bytesReadRate;
@@ -238,6 +241,14 @@ public class DiskOfferingResponse extends BaseResponse {
         this.storageType = storageType;
     }
 
+    public String getProvisioningType(){
+        return provisioningType;
+    }
+
+    public void setProvisioningType(String provisioningType){
+        this.provisioningType = provisioningType;
+    }
+
     public void setBytesReadRate(Long bytesReadRate) {
         this.bytesReadRate = bytesReadRate;
     }

api/src/org/apache/cloudstack/api/response/ServiceOfferingResponse.java

@@ -62,6 +62,9 @@ public class ServiceOfferingResponse extends BaseResponse {
     @Param(description = "the storage type for this service offering")
     private String storageType;
 
+    @SerializedName("provisioningtype") @Param(description="provisioning type used to create volumes. Valid values are thin, sparse, fat.", since = "4.4.0")
+    private String provisioningType;
+
     @SerializedName("offerha")
     @Param(description = "the ha support in the service offering")
     private Boolean offerHa;
@@ -241,6 +244,14 @@ public class ServiceOfferingResponse extends BaseResponse {
         this.storageType = storageType;
     }
 
+    public String getProvisioningType(){
+        return provisioningType;
+    }
+
+    public void setProvisioningType(String provisioningType){
+        this.provisioningType = provisioningType;
+    }
+
     public Boolean getOfferHa() {
         return offerHa;
     }

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

@@ -72,6 +72,10 @@ public class VolumeResponse extends BaseResponse implements ControlledViewEntity
     @Param(description = "state of the virtual machine")
     private String virtualMachineState;
 
+    @SerializedName(ApiConstants.PROVISIONINGTYPE)
+    @Param(description = "provisioning type used to create volumes.")
+    private String provisioningType;
+
     @SerializedName(ApiConstants.SIZE)
     @Param(description = "size of the disk volume")
     private Long size;
@@ -278,6 +282,10 @@ public class VolumeResponse extends BaseResponse implements ControlledViewEntity
         this.virtualMachineState = virtualMachineState;
     }
 
+    public void setProvisioningType(String provisioningType){
+        this.provisioningType = provisioningType;
+    }
+
     public void setSize(Long size) {
         this.size = size;
     }

client/WEB-INF/classes/resources/messages.properties

@@ -529,6 +529,7 @@ label.disk.bytes.write.rate=Disk Write Rate (BPS)
 label.disk.iops.read.rate=Disk Read Rate (IOPS)
 label.disk.iops.write.rate=Disk Write Rate (IOPS)
 label.disk.offering=Disk Offering
+label.disk.provisioningtype=Provisioning Type
 label.disk.read.bytes=Disk Read (Bytes)
 label.disk.read.io=Disk Read (IO)
 label.disk.size.gb=Disk Size (in GB)

core/src/org/apache/cloudstack/storage/to/VolumeObjectTO.java

@@ -38,6 +38,7 @@ public class VolumeObjectTO implements DataTO {
     private long accountId;
     private String chainInfo;
     private Storage.ImageFormat format;
+    private Storage.ProvisioningType provisioningType;
     private long id;
 
     private Long deviceId;
@@ -69,6 +70,7 @@ public class VolumeObjectTO implements DataTO {
         name = volume.getName();
         setId(volume.getId());
         format = volume.getFormat();
+        provisioningType = volume.getProvisioningType();
         bytesReadRate = volume.getBytesReadRate();
         bytesWriteRate = volume.getBytesWriteRate();
         iopsReadRate = volume.getIopsReadRate();
@@ -187,6 +189,14 @@ public class VolumeObjectTO implements DataTO {
         this.format = format;
     }
 
+    public Storage.ProvisioningType getProvisioningType(){
+        return provisioningType;
+    }
+
+    public void setProvisioningType(Storage.ProvisioningType provisioningType){
+        this.provisioningType = provisioningType;
+    }
+
     @Override
     public String toString() {
         return new StringBuilder("volumeTO[uuid=").append(uuid).append("|path=").append(path).append("|datastore=").append(dataStore).append("]").toString();

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

@@ -209,8 +209,17 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
     }
 
     public VolumeVO allocateDuplicateVolumeVO(Volume oldVol, Long templateId) {
-        VolumeVO newVol = new VolumeVO(oldVol.getVolumeType(), oldVol.getName(), oldVol.getDataCenterId(), oldVol.getDomainId(), oldVol.getAccountId(), oldVol.getDiskOfferingId(),
+        VolumeVO newVol = new VolumeVO(oldVol.getVolumeType(),
-                oldVol.getSize(), oldVol.getMinIops(), oldVol.getMaxIops(), oldVol.get_iScsiName());
+            oldVol.getName(),
+            oldVol.getDataCenterId(),
+            oldVol.getDomainId(),
+            oldVol.getAccountId(),
+            oldVol.getDiskOfferingId(),
+            oldVol.getProvisioningType(),
+            oldVol.getSize(),
+            oldVol.getMinIops(),
+            oldVol.getMaxIops(),
+            oldVol.get_iScsiName());
         if (templateId != null) {
             newVol.setTemplateId(templateId);
         } else {
@@ -584,7 +593,17 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
         minIops = minIops != null ? minIops : offering.getMinIops();
         maxIops = maxIops != null ? maxIops : offering.getMaxIops();
 
-        VolumeVO vol = new VolumeVO(type, name, vm.getDataCenterId(), owner.getDomainId(), owner.getId(), offering.getId(), size, minIops, maxIops, null);
+        VolumeVO vol = new VolumeVO(type,
+            name,
+            vm.getDataCenterId(),
+            owner.getDomainId(),
+            owner.getId(),
+            offering.getId(),
+            offering.getProvisioningType(),
+            size,
+            minIops,
+            maxIops,
+            null);
         if (vm != null) {
             vol.setInstanceId(vm.getId());
         }
@@ -635,8 +654,17 @@ public class VolumeOrchestrator extends ManagerBase implements VolumeOrchestrati
         minIops = minIops != null ? minIops : offering.getMinIops();
         maxIops = maxIops != null ? maxIops : offering.getMaxIops();
 
-        VolumeVO vol = new VolumeVO(type, name, vm.getDataCenterId(), owner.getDomainId(), owner.getId(), offering.getId(), size, minIops, maxIops, null);
+        VolumeVO vol = new VolumeVO(type,
-
+            name,
+            vm.getDataCenterId(),
+            owner.getDomainId(),
+            owner.getId(),
+            offering.getId(),
+            offering.getProvisioningType(),
+            size,
+            minIops,
+            maxIops,
+            null);
         vol.setFormat(getSupportedImageFormatForCluster(template.getHypervisorType()));
         if (vm != null) {
             vol.setInstanceId(vm.getId());

engine/orchestration/test/com/cloud/vm/VirtualMachineManagerImplTest.java

@@ -83,16 +83,17 @@ import com.cloud.hypervisor.HypervisorGuru;
 import com.cloud.hypervisor.HypervisorGuruManager;
 import com.cloud.offering.ServiceOffering;
 import com.cloud.service.ServiceOfferingVO;
-import com.cloud.storage.DiskOfferingVO;
-import com.cloud.storage.StoragePool;
-import com.cloud.storage.StoragePoolHostVO;
-import com.cloud.storage.VMTemplateVO;
-import com.cloud.storage.Volume;
-import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.storage.dao.StoragePoolHostDao;
 import com.cloud.storage.dao.VMTemplateDao;
 import com.cloud.storage.dao.VolumeDao;
+import com.cloud.storage.Storage.ProvisioningType;
+import com.cloud.storage.DiskOfferingVO;
+import com.cloud.storage.StoragePool;
+import com.cloud.storage.StoragePoolHostVO;
+import com.cloud.storage.Volume;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.VMTemplateVO;
 import com.cloud.user.Account;
 import com.cloud.user.AccountVO;
 import com.cloud.user.UserVO;
@@ -310,7 +311,7 @@ public class VirtualMachineManagerImplTest {
         boolean useLocalStorage = false;
 
         ServiceOfferingVO serviceOffering =
-                new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, ha, displayText, useLocalStorage, false, null, false, null, false);
+                new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, ha, displayText, ProvisioningType.THIN, useLocalStorage, false, null, false, null, false);
         return serviceOffering;
     }
 

engine/schema/src/com/cloud/service/ServiceOfferingVO.java

@@ -27,6 +27,7 @@ import javax.persistence.Transient;
 
 import com.cloud.offering.ServiceOffering;
 import com.cloud.storage.DiskOfferingVO;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.vm.VirtualMachine;
 
 @Entity
@@ -89,8 +90,8 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
     }
 
     public ServiceOfferingVO(String name, Integer cpu, Integer ramSize, Integer speed, Integer rateMbps, Integer multicastRateMbps, boolean offerHA, String displayText,
-            boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse, VirtualMachine.Type vmType, boolean defaultUse) {
+            ProvisioningType provisioningType, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse, VirtualMachine.Type vmType, boolean defaultUse) {
-        super(name, displayText, false, tags, recreatable, useLocalStorage, systemUse, true);
+        super(name, displayText, provisioningType, false, tags, recreatable, useLocalStorage, systemUse, true);
         this.cpu = cpu;
         this.ramSize = ramSize;
         this.speed = speed;
@@ -104,9 +105,8 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
     }
 
     public ServiceOfferingVO(String name, Integer cpu, Integer ramSize, Integer speed, Integer rateMbps, Integer multicastRateMbps, boolean offerHA, boolean limitCpuUse,
-            boolean volatileVm, String displayText, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse, VirtualMachine.Type vmType,
+            boolean volatileVm, String displayText, ProvisioningType provisioningType, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse, VirtualMachine.Type vmType, Long domainId) {
-            Long domainId) {
+        super(name, displayText, provisioningType, false, tags, recreatable, useLocalStorage, systemUse, true, domainId);
-        super(name, displayText, false, tags, recreatable, useLocalStorage, systemUse, true, domainId);
         this.cpu = cpu;
         this.ramSize = ramSize;
         this.speed = speed;
@@ -119,7 +119,7 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
     }
 
     public ServiceOfferingVO(String name, Integer cpu, Integer ramSize, Integer speed, Integer rateMbps, Integer multicastRateMbps, boolean offerHA,
-            boolean limitResourceUse, boolean volatileVm, String displayText, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse,
+            boolean limitResourceUse, boolean volatileVm, String displayText, ProvisioningType provisioningType, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse,
             VirtualMachine.Type vmType, Long domainId, String hostTag) {
         this(name,
             cpu,
@@ -131,6 +131,7 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
             limitResourceUse,
             volatileVm,
             displayText,
+            provisioningType,
             useLocalStorage,
             recreatable,
             tags,
@@ -141,7 +142,7 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
     }
 
     public ServiceOfferingVO(String name, Integer cpu, Integer ramSize, Integer speed, Integer rateMbps, Integer multicastRateMbps, boolean offerHA,
-            boolean limitResourceUse, boolean volatileVm, String displayText, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse,
+            boolean limitResourceUse, boolean volatileVm, String displayText, ProvisioningType provisioningType, boolean useLocalStorage, boolean recreatable, String tags, boolean systemUse,
             VirtualMachine.Type vmType, Long domainId, String hostTag, String deploymentPlanner) {
         this(name,
             cpu,
@@ -153,6 +154,7 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
             limitResourceUse,
             volatileVm,
             displayText,
+            provisioningType,
             useLocalStorage,
             recreatable,
             tags,
@@ -167,6 +169,7 @@ public class ServiceOfferingVO extends DiskOfferingVO implements ServiceOffering
         super(offering.getId(),
             offering.getName(),
             offering.getDisplayText(),
+            offering.getProvisioningType(),
             false,
             offering.getTags(),
             offering.isRecreatable(),

engine/schema/src/com/cloud/storage/DiskOfferingVO.java

@@ -124,6 +124,9 @@ public class DiskOfferingVO implements DiskOffering {
     @Enumerated(value = EnumType.STRING)
     private DiskCacheMode cacheMode;
 
+    @Column(name="provisioning_type")
+    Storage.ProvisioningType provisioningType;
+
     @Column(name="display_offering")
     boolean displayOffering = true;
 
@@ -138,11 +141,12 @@ public class DiskOfferingVO implements DiskOffering {
         uuid = UUID.randomUUID().toString();
     }
 
-    public DiskOfferingVO(Long domainId, String name, String displayText, long diskSize, String tags, boolean isCustomized,
+    public DiskOfferingVO(Long domainId, String name, String displayText, Storage.ProvisioningType provisioningType, long diskSize, String tags, boolean isCustomized,
             Boolean isCustomizedIops, Long minIops, Long maxIops, DiskCacheMode cacheMode) {
         this.domainId = domainId;
         this.name = name;
         this.displayText = displayText;
+        this.provisioningType = provisioningType;
         this.diskSize = diskSize;
         this.tags = tags;
         recreatable = false;
@@ -156,11 +160,12 @@ public class DiskOfferingVO implements DiskOffering {
         this.cacheMode = cacheMode;
     }
 
-    public DiskOfferingVO(Long domainId, String name, String displayText, long diskSize, String tags, boolean isCustomized,
+    public DiskOfferingVO(Long domainId, String name, String displayText, Storage.ProvisioningType provisioningType, long diskSize, String tags, boolean isCustomized,
             Boolean isCustomizedIops, Long minIops, Long maxIops) {
         this.domainId = domainId;
         this.name = name;
         this.displayText = displayText;
+        this.provisioningType = provisioningType;
         this.diskSize = diskSize;
         this.tags = tags;
         recreatable = false;
@@ -174,12 +179,13 @@ public class DiskOfferingVO implements DiskOffering {
         state = State.Active;
     }
 
-    public DiskOfferingVO(String name, String displayText, boolean mirrored, String tags, boolean recreatable, boolean useLocalStorage, boolean systemUse,
+    public DiskOfferingVO(String name, String displayText, Storage.ProvisioningType provisioningType, boolean mirrored, String tags, boolean recreatable,
-            boolean customized) {
+            boolean useLocalStorage, boolean systemUse, boolean customized) {
         domainId = null;
         type = Type.Service;
         this.name = name;
         this.displayText = displayText;
+        this.provisioningType = provisioningType;
         this.tags = tags;
         this.recreatable = recreatable;
         this.useLocalStorage = useLocalStorage;
@@ -191,11 +197,12 @@ public class DiskOfferingVO implements DiskOffering {
 
     // domain specific offerings constructor (null domainId implies public
     // offering)
-    public DiskOfferingVO(String name, String displayText, boolean mirrored, String tags, boolean recreatable, boolean useLocalStorage, boolean systemUse,
+    public DiskOfferingVO(String name, String displayText, Storage.ProvisioningType provisioningType, boolean mirrored, String tags, boolean recreatable,
-            boolean customized, Long domainId) {
+            boolean useLocalStorage, boolean systemUse, boolean customized, Long domainId) {
         type = Type.Service;
         this.name = name;
         this.displayText = displayText;
+        this.provisioningType = provisioningType;
         this.tags = tags;
         this.recreatable = recreatable;
         this.useLocalStorage = useLocalStorage;
@@ -206,12 +213,13 @@ public class DiskOfferingVO implements DiskOffering {
         state = State.Active;
     }
 
-    public DiskOfferingVO(long id, String name, String displayText, boolean mirrored, String tags, boolean recreatable,
+    public DiskOfferingVO(long id, String name, String displayText, Storage.ProvisioningType provisioningType, boolean mirrored, String tags, boolean recreatable,
             boolean useLocalStorage, boolean systemUse, boolean customized, boolean customizedIops, Long domainId) {
         this.id = id;
         type = Type.Service;
         this.name = name;
         this.displayText = displayText;
+        this.provisioningType = provisioningType;
         this.tags = tags;
         this.recreatable = recreatable;
         this.useLocalStorage = useLocalStorage;
@@ -341,6 +349,11 @@ public class DiskOfferingVO implements DiskOffering {
         this.displayText = displayText;
     }
 
+    @Override
+    public Storage.ProvisioningType getProvisioningType(){
+        return provisioningType;
+    }
+
     @Override
     public long getDiskSize() {
         return diskSize;

engine/schema/src/com/cloud/storage/VolumeVO.java

@@ -32,6 +32,7 @@ import javax.persistence.Temporal;
 import javax.persistence.TemporalType;
 import javax.persistence.Transient;
 
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.utils.NumbersUtil;
 import com.cloud.utils.db.GenericDao;
@@ -143,6 +144,9 @@ public class VolumeVO implements Volume {
     @Column(name = "format")
     private Storage.ImageFormat format;
 
+    @Column(name = "provisioning_type")
+    private Storage.ProvisioningType provisioningType;
+
     @Column(name = "display_volume", updatable = true, nullable = false)
     protected boolean displayVolume = true;
 
@@ -163,12 +167,15 @@ public class VolumeVO implements Volume {
     Integer hypervisorSnapshotReserve;
 
     // Real Constructor
-    public VolumeVO(Type type, String name, long dcId, long domainId, long accountId, long diskOfferingId, long size, Long minIops, Long maxIops, String iScsiName) {
+    public VolumeVO(Type type, String name, long dcId, long domainId,
-        volumeType = type;
+            long accountId, long diskOfferingId, Storage.ProvisioningType provisioningType, long size,
+            Long minIops, Long maxIops, String iScsiName) {
+        this.volumeType = type;
         this.name = name;
         dataCenterId = dcId;
         this.accountId = accountId;
         this.domainId = domainId;
+        this.provisioningType = provisioningType;
         this.size = size;
         this.minIops = minIops;
         this.maxIops = maxIops;
@@ -178,14 +185,17 @@ public class VolumeVO implements Volume {
         uuid = UUID.randomUUID().toString();
     }
 
-    public VolumeVO(String name, long dcId, Long podId, long accountId, long domainId, Long instanceId, String folder, String path, long size, Long minIops,
+    public VolumeVO(String name, long dcId, Long podId, long accountId,
-            Long maxIops, String iScsiName, Volume.Type vType) {
+            long domainId, Long instanceId, String folder, String path, Storage.ProvisioningType provisioningType,
+            long size, Long minIops, Long maxIops, String iScsiName,
+            Volume.Type vType) {
         this.name = name;
         this.accountId = accountId;
         this.domainId = domainId;
         this.instanceId = instanceId;
         this.folder = folder;
         this.path = path;
+        this.provisioningType = provisioningType;
         this.size = size;
         this.minIops = minIops;
         this.maxIops = maxIops;
@@ -198,13 +208,16 @@ public class VolumeVO implements Volume {
         uuid = UUID.randomUUID().toString();
     }
 
-    public VolumeVO(String name, long dcId, long podId, long accountId, long domainId, Long instanceId, String folder, String path, long size, Volume.Type vType) {
+    public VolumeVO(String name, long dcId, long podId, long accountId,
+            long domainId, Long instanceId, String folder, String path, Storage.ProvisioningType provisioningType,
+            long size, Volume.Type vType) {
         this.name = name;
         this.accountId = accountId;
         this.domainId = domainId;
         this.instanceId = instanceId;
         this.folder = folder;
         this.path = path;
+        this.provisioningType = provisioningType;
         this.size = size;
         minIops = null;
         maxIops = null;
@@ -227,6 +240,7 @@ public class VolumeVO implements Volume {
             that.getInstanceId(),
             that.getFolder(),
             that.getPath(),
+            that.getProvisioningType(),
             that.getSize(),
             that.getMinIops(),
             that.getMaxIops(),
@@ -245,6 +259,7 @@ public class VolumeVO implements Volume {
         templateId = that.getTemplateId();
         deviceId = that.getDeviceId();
         format = that.getFormat();
+        provisioningType = that.getProvisioningType();
         uuid = UUID.randomUUID().toString();
     }
 
@@ -565,6 +580,15 @@ public class VolumeVO implements Volume {
         this.format = format;
     }
 
+    @Override
+    public ProvisioningType getProvisioningType(){
+        return provisioningType;
+    }
+
+    public void setProvisioningType(ProvisioningType provisioningType){
+        this.provisioningType = provisioningType;
+    }
+
     public void setVmSnapshotChainSize(Long vmSnapshotChainSize){
         this.vmSnapshotChainSize = vmSnapshotChainSize;
     }

engine/storage/volume/src/org/apache/cloudstack/storage/volume/VolumeObject.java

@@ -42,6 +42,7 @@ import com.cloud.offering.DiskOffering.DiskCacheMode;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.Storage.ImageFormat;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.DiskOfferingDao;
@@ -655,6 +656,11 @@ public class VolumeObject implements VolumeInfo {
         return volumeVO.getFormat();
     }
 
+    @Override
+    public ProvisioningType getProvisioningType(){
+        return this.volumeVO.getProvisioningType();
+    }
+
     @Override
     public boolean delete() {
         if (dataStore != null) {

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/resource/LibvirtComputingResource.java

@@ -1735,13 +1735,14 @@ public class LibvirtComputingResource extends ServerResourceBase implements Serv
                     vol = templateToPrimaryDownload(cmd.getTemplateUrl(), primaryPool, dskch.getPath());
                 } else {
                     BaseVol = primaryPool.getPhysicalDisk(cmd.getTemplateUrl());
-                    vol = _storagePoolMgr.createDiskFromTemplate(BaseVol, dskch.getPath(), primaryPool, 0);
+                    vol = _storagePoolMgr.createDiskFromTemplate(BaseVol,
+                            dskch.getPath(), dskch.getProvisioningType(), primaryPool, 0);
                 }
                 if (vol == null) {
                     return new Answer(cmd, false, " Can't create storage volume on storage pool");
                 }
             } else {
-                vol = primaryPool.createPhysicalDisk(dskch.getPath(), dskch.getSize());
+                vol = primaryPool.createPhysicalDisk(dskch.getPath(), dskch.getProvisioningType(), dskch.getSize());
             }
             VolumeTO volume =
                     new VolumeTO(cmd.getVolumeId(), dskch.getType(), pool.getType(), pool.getUuid(), pool.getPath(), vol.getName(), vol.getName(), disksize, null);

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/resource/LibvirtStorageVolumeDef.java

@@ -16,6 +16,9 @@
 // under the License.
 package com.cloud.hypervisor.kvm.resource;
 
+import org.apache.cloudstack.utils.qemu.QemuImg;
+import org.apache.commons.lang.NotImplementedException;
+
 public class LibvirtStorageVolumeDef {
     public enum volFormat {
         RAW("raw"), QCOW2("qcow2"), DIR("dir"), TAR("tar");
@@ -45,6 +48,21 @@ public class LibvirtStorageVolumeDef {
             }
             return null;
         }
+
+        public static volFormat getFormat(QemuImg.PhysicalDiskFormat format){
+            switch (format){
+                case RAW:
+                    return RAW;
+                case QCOW2:
+                    return QCOW2;
+                case DIR:
+                    return DIR;
+                case TAR:
+                    return TAR;
+                default:
+                    throw new NotImplementedException();
+            }
+        }
     }
 
     private String _volName;

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/IscsiAdmStorageAdaptor.java

@@ -20,11 +20,13 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import com.cloud.storage.Storage;
 import org.apache.log4j.Logger;
 
 import org.apache.cloudstack.utils.qemu.QemuImg.PhysicalDiskFormat;
 
 import com.cloud.agent.api.to.DiskTO;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.utils.StringUtils;
 import com.cloud.utils.exception.CloudRuntimeException;
@@ -63,7 +65,7 @@ public class IscsiAdmStorageAdaptor implements StorageAdaptor {
     // called from LibvirtComputingResource.execute(CreateCommand)
     // does not apply for iScsiAdmStorageAdaptor
     @Override
-    public KVMPhysicalDisk createPhysicalDisk(String volumeUuid, KVMStoragePool pool, PhysicalDiskFormat format, long size) {
+    public KVMPhysicalDisk createPhysicalDisk(String volumeUuid, KVMStoragePool pool, PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
         throw new UnsupportedOperationException("Creating a physical disk is not supported.");
     }
 
@@ -336,7 +338,9 @@ public class IscsiAdmStorageAdaptor implements StorageAdaptor {
     }
 
     @Override
-    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, PhysicalDiskFormat format, long size, KVMStoragePool destPool, int timeout) {
+    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, PhysicalDiskFormat format,
+            ProvisioningType provisioningType, long size,
+            KVMStoragePool destPool, int timeout) {
         throw new UnsupportedOperationException("Creating a disk from a template is not yet supported for this configuration.");
     }
 

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/IscsiAdmStoragePool.java

@@ -19,6 +19,7 @@ package com.cloud.hypervisor.kvm.storage;
 import java.util.List;
 import java.util.Map;
 
+import com.cloud.storage.Storage;
 import org.apache.cloudstack.utils.qemu.QemuImg.PhysicalDiskFormat;
 
 import com.cloud.storage.Storage.StoragePoolType;
@@ -86,7 +87,7 @@ public class IscsiAdmStoragePool implements KVMStoragePool {
     // from LibvirtComputingResource.createDiskFromTemplate(KVMPhysicalDisk, String, PhysicalDiskFormat, long, KVMStoragePool)
     // does not apply for iScsiAdmStoragePool
     @Override
-    public KVMPhysicalDisk createPhysicalDisk(String name, PhysicalDiskFormat format, long size) {
+    public KVMPhysicalDisk createPhysicalDisk(String name, PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
         throw new UnsupportedOperationException("Creating a physical disk is not supported.");
     }
 
@@ -94,7 +95,7 @@ public class IscsiAdmStoragePool implements KVMStoragePool {
     // from KVMStorageProcessor.createVolume(CreateObjectCommand)
     // does not apply for iScsiAdmStoragePool
     @Override
-    public KVMPhysicalDisk createPhysicalDisk(String name, long size) {
+    public KVMPhysicalDisk createPhysicalDisk(String name, Storage.ProvisioningType provisioningType, long size) {
         throw new UnsupportedOperationException("Creating a physical disk is not supported.");
     }
 

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStoragePool.java

@@ -19,14 +19,15 @@ package com.cloud.hypervisor.kvm.storage;
 import java.util.List;
 import java.util.Map;
 
+import com.cloud.storage.Storage;
 import org.apache.cloudstack.utils.qemu.QemuImg.PhysicalDiskFormat;
 
 import com.cloud.storage.Storage.StoragePoolType;
 
 public interface KVMStoragePool {
-    public KVMPhysicalDisk createPhysicalDisk(String volumeUuid, PhysicalDiskFormat format, long size);
+    public KVMPhysicalDisk createPhysicalDisk(String volumeUuid, PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size);
 
-    public KVMPhysicalDisk createPhysicalDisk(String volumeUuid, long size);
+    public KVMPhysicalDisk createPhysicalDisk(String volumeUuid, Storage.ProvisioningType provisioningType, long size);
 
     public boolean connectPhysicalDisk(String volumeUuid, Map<String, String> details);
 

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStoragePoolManager.java

@@ -36,6 +36,7 @@ import com.cloud.agent.api.to.VirtualMachineTO;
 import com.cloud.hypervisor.kvm.resource.KVMHABase;
 import com.cloud.hypervisor.kvm.resource.KVMHABase.PoolType;
 import com.cloud.hypervisor.kvm.resource.KVMHAMonitor;
+import com.cloud.storage.Storage;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.storage.StorageLayer;
 import com.cloud.storage.Volume;
@@ -301,22 +302,32 @@ public class KVMStoragePoolManager {
         return true;
     }
 
-    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, KVMStoragePool destPool, int timeout) {
+    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, Storage.ProvisioningType provisioningType,
-        return createDiskFromTemplate(template, name, destPool, template.getSize(), timeout);
+                                                    KVMStoragePool destPool, int timeout) {
+        return createDiskFromTemplate(template, name, provisioningType, destPool, template.getSize(), timeout);
     }
 
-    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, KVMStoragePool destPool, long size, int timeout) {
+    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, Storage.ProvisioningType provisioningType,
+                                                    KVMStoragePool destPool, long size, int timeout) {
         StorageAdaptor adaptor = getStorageAdaptor(destPool.getType());
 
         // LibvirtStorageAdaptor-specific statement
         if (destPool.getType() == StoragePoolType.RBD) {
-            return adaptor.createDiskFromTemplate(template, name, PhysicalDiskFormat.RAW, size, destPool, timeout);
+            return adaptor.createDiskFromTemplate(template, name,
+                    PhysicalDiskFormat.RAW, provisioningType,
+                    size, destPool, timeout);
         } else if (destPool.getType() == StoragePoolType.CLVM) {
-            return adaptor.createDiskFromTemplate(template, name, PhysicalDiskFormat.RAW, size, destPool, timeout);
+            return adaptor.createDiskFromTemplate(template, name,
+                    PhysicalDiskFormat.RAW, provisioningType,
+                    size, destPool, timeout);
         } else if (template.getFormat() == PhysicalDiskFormat.DIR) {
-            return adaptor.createDiskFromTemplate(template, name, PhysicalDiskFormat.DIR, size, destPool, timeout);
+            return adaptor.createDiskFromTemplate(template, name,
+                    PhysicalDiskFormat.DIR, provisioningType,
+                    size, destPool, timeout);
         } else {
-            return adaptor.createDiskFromTemplate(template, name, PhysicalDiskFormat.QCOW2, size, destPool, timeout);
+            return adaptor.createDiskFromTemplate(template, name,
+                    PhysicalDiskFormat.QCOW2, provisioningType,
+                    size, destPool, timeout);
         }
     }
 

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java

@@ -335,7 +335,8 @@ public class KVMStorageProcessor implements StorageProcessor {
                     templatePath = templatePath.substring(templatePath.lastIndexOf(File.separator) + 1);
                 }
                 BaseVol = storagePoolMgr.getPhysicalDisk(primaryStore.getPoolType(), primaryStore.getUuid(), templatePath);
-                vol = storagePoolMgr.createDiskFromTemplate(BaseVol, volume.getUuid(), BaseVol.getPool(), volume.getSize(), cmd.getWaitInMillSeconds());
+                vol = storagePoolMgr.createDiskFromTemplate(BaseVol, volume.getUuid(), volume.getProvisioningType(),
+                        BaseVol.getPool(), volume.getSize(), cmd.getWaitInMillSeconds());
             }
             if (vol == null) {
                 return new CopyCmdAnswer(" Can't create storage volume on storage pool");
@@ -1074,7 +1075,7 @@ public class KVMStorageProcessor implements StorageProcessor {
             primaryPool = storagePoolMgr.getStoragePool(primaryStore.getPoolType(), primaryStore.getUuid());
             disksize = volume.getSize();
 
-            vol = primaryPool.createPhysicalDisk(volume.getUuid(), disksize);
+            vol = primaryPool.createPhysicalDisk(volume.getUuid(), volume.getProvisioningType(), disksize);
 
             VolumeObjectTO newVol = new VolumeObjectTO();
             newVol.setPath(vol.getName());

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/LibvirtStorageAdaptor.java

@@ -21,6 +21,8 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.HashMap;
+
 
 import org.apache.commons.codec.binary.Base64;
 import org.apache.log4j.Logger;
@@ -55,6 +57,7 @@ import com.cloud.hypervisor.kvm.resource.LibvirtStoragePoolXMLParser;
 import com.cloud.hypervisor.kvm.resource.LibvirtStorageVolumeDef;
 import com.cloud.hypervisor.kvm.resource.LibvirtStorageVolumeDef.volFormat;
 import com.cloud.hypervisor.kvm.resource.LibvirtStorageVolumeXMLParser;
+import com.cloud.storage.Storage;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.storage.StorageLayer;
 import com.cloud.utils.exception.CloudRuntimeException;
@@ -614,16 +617,103 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
     }
 
     @Override
-    public KVMPhysicalDisk createPhysicalDisk(String name, KVMStoragePool pool, PhysicalDiskFormat format, long size) {
+    public KVMPhysicalDisk createPhysicalDisk(String name, KVMStoragePool pool,
+            PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
+
+        switch (pool.getType()){
+            case RBD:
+                return createPhysicalDiskOnRBD(name, pool, format, provisioningType, size);
+            case NetworkFilesystem:
+            case Filesystem:
+                switch (format){
+                    case QCOW2:
+                        return createPhysicalDiskByQemuImg(name, pool, format, provisioningType, size);
+                    case RAW:
+                        return createPhysicalDiskByQemuImg(name, pool, format, provisioningType, size);
+                    case DIR:
+                        return createPhysicalDiskByLibVirt(name, pool, format, provisioningType, size);
+                    case TAR:
+                        return createPhysicalDiskByLibVirt(name, pool, format, provisioningType, size);
+                    default:
+                        throw new CloudRuntimeException("Unexpected disk format is specified.");
+                }
+            default:
+                return createPhysicalDiskByLibVirt(name, pool, format, provisioningType, size);
+        }
+    }
+
+    private KVMPhysicalDisk createPhysicalDiskByLibVirt(String name, KVMStoragePool pool,
+                                                        PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
         LibvirtStoragePool libvirtPool = (LibvirtStoragePool) pool;
         StoragePool virtPool = libvirtPool.getPool();
-        LibvirtStorageVolumeDef.volFormat libvirtformat = null;
+        LibvirtStorageVolumeDef.volFormat libvirtformat = LibvirtStorageVolumeDef.volFormat.getFormat(format);
 
         String volPath = null;
         String volName = null;
         long volAllocation = 0;
         long volCapacity = 0;
 
+        LibvirtStorageVolumeDef volDef = new LibvirtStorageVolumeDef(name,
+                size, libvirtformat, null, null);
+        s_logger.debug(volDef.toString());
+        try {
+            StorageVol vol = virtPool.storageVolCreateXML(volDef.toString(), 0);
+            volPath = vol.getPath();
+            volName = vol.getName();
+            volAllocation = vol.getInfo().allocation;
+            volCapacity = vol.getInfo().capacity;
+        } catch (LibvirtException e) {
+            throw new CloudRuntimeException(e.toString());
+        }
+
+        KVMPhysicalDisk disk = new KVMPhysicalDisk(volPath, volName, pool);
+        disk.setFormat(format);
+        disk.setSize(volAllocation);
+        disk.setVirtualSize(volCapacity);
+        return disk;
+    }
+
+
+    private KVMPhysicalDisk createPhysicalDiskByQemuImg(String name, KVMStoragePool pool,
+                                                    PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
+        String volPath = pool.getLocalPath() + "/" + name;
+        String volName = name;
+        long volAllocation = 0;
+        long volCapacity = 0;
+
+        final int timeout = 0;
+
+        QemuImgFile destFile = new QemuImgFile(volPath);
+        destFile.setFormat(format);
+        destFile.setSize(size);
+        QemuImg qemu = new QemuImg(timeout);
+        Map<String, String> options = new HashMap<String, String>();
+        if (pool.getType() == StoragePoolType.NetworkFilesystem){
+            options.put("preallocation", QemuImg.PreallocationType.getPreallocationType(provisioningType).toString());
+        }
+
+        try{
+            qemu.create(destFile, options);
+            Map<String, String> info = qemu.info(destFile);
+            volAllocation = Long.parseLong(info.get(new String("virtual-size")));
+            volCapacity = Long.parseLong(info.get(new String("actual-size")));
+        } catch (QemuImgException e) {
+            s_logger.error("Failed to create " + volPath +
+                    " due to a failed executing of qemu-img: " + e.getMessage());
+        }
+
+        KVMPhysicalDisk disk = new KVMPhysicalDisk(volPath, volName, pool);
+        disk.setFormat(format);
+        disk.setSize(volAllocation);
+        disk.setVirtualSize(volCapacity);
+        return disk;
+    }
+
+    private KVMPhysicalDisk createPhysicalDiskOnRBD(String name, KVMStoragePool pool,
+                                               PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
+        LibvirtStoragePool libvirtPool = (LibvirtStoragePool) pool;
+        String volPath = null;
+
         /**
          * To have RBD function properly we want RBD images of format 2
          * libvirt currently defaults to format 1
@@ -631,7 +721,6 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
          * For that reason we use the native RBD bindings to create the
          * RBD image until libvirt creates RBD format 2 by default
          */
-        if (pool.getType() == StoragePoolType.RBD) {
         format = PhysicalDiskFormat.RAW;
 
         try {
@@ -646,7 +735,7 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
 
             IoCTX io = r.ioCtxCreate(pool.getSourceDir());
             Rbd rbd = new Rbd(io);
-                rbd.create(name, size, rbdFeatures, rbdOrder);
+            rbd.create(name, size, this.rbdFeatures, this.rbdOrder);
 
             r.ioCtxDestroy(io);
         } catch (RadosException e) {
@@ -656,38 +745,10 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
         }
 
         volPath = pool.getSourceDir() + "/" + name;
-            volName = name;
+        KVMPhysicalDisk disk = new KVMPhysicalDisk(volPath, name, pool);
-            volCapacity = size;
-            volAllocation = size;
-        } else {
-
-            if (format == PhysicalDiskFormat.QCOW2) {
-                libvirtformat = LibvirtStorageVolumeDef.volFormat.QCOW2;
-            } else if (format == PhysicalDiskFormat.RAW) {
-                libvirtformat = LibvirtStorageVolumeDef.volFormat.RAW;
-            } else if (format == PhysicalDiskFormat.DIR) {
-                libvirtformat = LibvirtStorageVolumeDef.volFormat.DIR;
-            } else if (format == PhysicalDiskFormat.TAR) {
-                libvirtformat = LibvirtStorageVolumeDef.volFormat.TAR;
-            }
-
-            LibvirtStorageVolumeDef volDef = new LibvirtStorageVolumeDef(name, size, libvirtformat, null, null);
-            s_logger.debug(volDef.toString());
-            try {
-                StorageVol vol = virtPool.storageVolCreateXML(volDef.toString(), 0);
-                volPath = vol.getPath();
-                volName = vol.getName();
-                volAllocation = vol.getInfo().allocation;
-                volCapacity = vol.getInfo().capacity;
-            } catch (LibvirtException e) {
-                throw new CloudRuntimeException(e.toString());
-            }
-        }
-
-        KVMPhysicalDisk disk = new KVMPhysicalDisk(volPath, volName, pool);
         disk.setFormat(format);
-        disk.setSize(volAllocation);
+        disk.setSize(size);
-        disk.setVirtualSize(volCapacity);
+        disk.setVirtualSize(size);
         return disk;
     }
 
@@ -806,38 +867,44 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
      * If it has been created on Primary Storage, it will be copied on the Primary Storage
      */
     @Override
-    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, PhysicalDiskFormat format, long size, KVMStoragePool destPool, int timeout) {
+    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template,
+            String name, PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size, KVMStoragePool destPool, int timeout) {
 
-        String newUuid = name;
-        KVMStoragePool srcPool = template.getPool();
         KVMPhysicalDisk disk = null;
 
-        /*
+        if (destPool.getType() == StoragePoolType.RBD) {
-            With RBD you can't run qemu-img convert with an existing RBD image as destination
+            disk = createDiskFromTemplateOnRBD(template, name, format, provisioningType, size, destPool, timeout);
-            qemu-img will exit with the error that the destination already exists.
+        } else {
-            So for RBD we don't create the image, but let qemu-img do that for us.
-
-            We then create a KVMPhysicalDisk object that we can return
-         */
             try {
-            if (destPool.getType() != StoragePoolType.RBD) {
+                String newUuid = name;
-                disk = destPool.createPhysicalDisk(newUuid, format, template.getVirtualSize());
+                disk = destPool.createPhysicalDisk(newUuid, format, provisioningType, template.getVirtualSize());
                 if (template.getFormat() == PhysicalDiskFormat.TAR) {
-                    Script.runSimpleBashScript("tar -x -f " + template.getPath() + " -C " + disk.getPath(), timeout);
+                    Script.runSimpleBashScript("tar -x -f " + template.getPath() + " -C " + disk.getPath(), timeout); // TO BE FIXED to aware provisioningType
                 } else if (template.getFormat() == PhysicalDiskFormat.DIR) {
                     Script.runSimpleBashScript("mkdir -p " + disk.getPath());
                     Script.runSimpleBashScript("chmod 755 " + disk.getPath());
-                    Script.runSimpleBashScript("cp -p -r " + template.getPath() + "/* " + disk.getPath(), timeout);
+                    Script.runSimpleBashScript("cp -p -r " + template.getPath() + "/* " + disk.getPath(), timeout); // TO BE FIXED to aware provisioningType
                 } else if (format == PhysicalDiskFormat.QCOW2) {
-                    QemuImgFile backingFile = new QemuImgFile(template.getPath(), template.getFormat());
+                    QemuImg qemu = new QemuImg(timeout);
-                    QemuImgFile destFile = new QemuImgFile(disk.getPath());
+                    QemuImgFile destFile = new QemuImgFile(disk.getPath(), format);
                     if (size > template.getVirtualSize()) {
                         destFile.setSize(size);
                     } else {
                         destFile.setSize(template.getVirtualSize());
                     }
-                    QemuImg qemu = new QemuImg(timeout);
+                    Map<String, String> options = new HashMap<String, String>();
-                    qemu.create(destFile, backingFile);
+                    options.put("preallocation", QemuImg.PreallocationType.getPreallocationType(provisioningType).toString());
+                    switch(provisioningType){
+                        case THIN:
+                            QemuImgFile backingFile = new QemuImgFile(template.getPath(), template.getFormat());
+                            qemu.create(destFile, backingFile, options);
+                            break;
+                        case SPARSE:
+                        case FAT:
+                            QemuImgFile srcFile = new QemuImgFile(template.getPath(), template.getFormat());
+                            qemu.convert(srcFile, destFile, options);
+                            break;
+                    }
                 } else if (format == PhysicalDiskFormat.RAW) {
                     QemuImgFile sourceFile = new QemuImgFile(template.getPath(), template.getFormat());
                     QemuImgFile destFile = new QemuImgFile(disk.getPath(), PhysicalDiskFormat.RAW);
@@ -847,9 +914,37 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
                         destFile.setSize(template.getVirtualSize());
                     }
                     QemuImg qemu = new QemuImg(timeout);
-                    qemu.convert(sourceFile, destFile);
+                    Map<String, String> options = new HashMap<String, String>();
+                    qemu.convert(sourceFile, destFile, options);
                 }
-            } else {
+            } catch (QemuImgException e) {
+                s_logger.error("Failed to create " + disk.getPath() +
+                        " due to a failed executing of qemu-img: " + e.getMessage());
+            }
+        }
+
+        if (disk == null) {
+            throw new CloudRuntimeException("Failed to create disk from template " + template.getName());
+        }
+
+        return disk;
+    }
+
+    private KVMPhysicalDisk createDiskFromTemplateOnRBD(KVMPhysicalDisk template,
+            String name, PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size, KVMStoragePool destPool, int timeout){
+
+        /*
+            With RBD you can't run qemu-img convert with an existing RBD image as destination
+            qemu-img will exit with the error that the destination already exists.
+            So for RBD we don't create the image, but let qemu-img do that for us.
+
+            We then create a KVMPhysicalDisk object that we can return
+        */
+
+        KVMStoragePool srcPool = template.getPool();
+        KVMPhysicalDisk disk = null;
+        String newUuid = name;
+
         format = PhysicalDiskFormat.RAW;
         disk = new KVMPhysicalDisk(destPool.getSourceDir() + "/" + newUuid, newUuid, destPool);
         disk.setFormat(format);
@@ -862,21 +957,25 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
             disk.setVirtualSize(disk.getSize());
         }
 
+
         QemuImg qemu = new QemuImg(timeout);
         QemuImgFile srcFile;
-                QemuImgFile destFile =
+        QemuImgFile destFile = new QemuImgFile(KVMPhysicalDisk.RBDStringBuilder(destPool.getSourceHost(),
-                        new QemuImgFile(KVMPhysicalDisk.RBDStringBuilder(destPool.getSourceHost(), destPool.getSourcePort(), destPool.getAuthUserName(),
+                destPool.getSourcePort(),
-                                destPool.getAuthSecret(), disk.getPath()));
+                destPool.getAuthUserName(),
+                destPool.getAuthSecret(),
+                disk.getPath()));
         destFile.setFormat(format);
-                if (size > template.getVirtualSize()) {
+
-                    destFile.setSize(size);
-                } else {
-                    destFile.setSize(template.getVirtualSize());
-                }
 
         if (srcPool.getType() != StoragePoolType.RBD) {
             srcFile = new QemuImgFile(template.getPath(), template.getFormat());
+            try{
                 qemu.convert(srcFile, destFile);
+            } catch (QemuImgException e) {
+                s_logger.error("Failed to create " + disk.getPath() +
+                        " due to a failed executing of qemu-img: " + e.getMessage());
+            }
         } else {
 
             /**
@@ -917,11 +1016,12 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
                         s_logger.debug("Finished copying " + srcImage.getName() +  " to " + destImage.getName() + " in Ceph pool " + srcPool.getSourceDir());
                         rbd.close(destImage);
                     } else {
-                                s_logger.debug("The source image " + srcPool.getSourceDir() + "/" + template.getName() +
+                        s_logger.debug("The source image " + srcPool.getSourceDir() + "/" + template.getName()
-                                        " is RBD format 2. We will perform a RBD clone using snapshot " + rbdTemplateSnapName);
+                                + " is RBD format 2. We will perform a RBD clone using snapshot "
+                                + this.rbdTemplateSnapName);
                                 /* The source image is format 2, we can do a RBD snapshot+clone (layering) */
-                                rbd.clone(template.getName(), rbdTemplateSnapName, io, disk.getName(), rbdFeatures, rbdOrder);
+                        rbd.clone(template.getName(), this.rbdTemplateSnapName, io, disk.getName(), this.rbdFeatures, this.rbdOrder);
-                                s_logger.debug("Succesfully cloned " + template.getName() + "@" + rbdTemplateSnapName + " to " + disk.getName());
+                        s_logger.debug("Succesfully cloned " + template.getName() + "@" + this.rbdTemplateSnapName + " to " + disk.getName());
                     }
 
                     rbd.close(srcImage);
@@ -957,8 +1057,8 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
                     RbdImage srcImage = sRbd.open(template.getName());
                     RbdImage destImage = dRbd.open(disk.getName());
 
-                            s_logger.debug("Copying " + template.getName() + " from Ceph cluster " + rSrc.confGet("mon_host") + " to " + disk.getName() + " on cluster " +
+                    s_logger.debug("Copying " + template.getName() + " from Ceph cluster " + rSrc.confGet("mon_host") + " to " + disk.getName()
-                                    rDest.confGet("mon_host"));
+                            + " on cluster " + rDest.confGet("mon_host"));
                     sRbd.copy(srcImage, destImage);
 
                     sRbd.close(srcImage);
@@ -975,15 +1075,6 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
                 disk = null;
             }
         }
-            }
-        } catch (QemuImgException e) {
-            s_logger.error("Failed to create " + disk.getPath() + " due to a failed executing of qemu-img: " + e.getMessage());
-        }
-
-        if (disk == null) {
-            throw new CloudRuntimeException("Failed to create disk from template " + template.getName());
-        }
-
         return disk;
     }
 
@@ -1037,13 +1128,13 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
         s_logger.debug("copyPhysicalDisk: disk size:" + disk.getSize() + ", virtualsize:" + disk.getVirtualSize()+" format:"+disk.getFormat());
         if (destPool.getType() != StoragePoolType.RBD) {
             if (disk.getFormat() == PhysicalDiskFormat.TAR) {
-                newDisk = destPool.createPhysicalDisk(name, PhysicalDiskFormat.DIR, disk.getVirtualSize());
+                newDisk = destPool.createPhysicalDisk(name, PhysicalDiskFormat.DIR, Storage.ProvisioningType.THIN, disk.getVirtualSize());
             } else {
                 /* If the source device is on a RBD storage pool force the new disk to the same format (RAW) */
                 if (srcPool.getType() != StoragePoolType.RBD) {
-                    newDisk = destPool.createPhysicalDisk(name, disk.getVirtualSize());
+                    newDisk = destPool.createPhysicalDisk(name, Storage.ProvisioningType.THIN, disk.getVirtualSize());
                 } else {
-                    newDisk = destPool.createPhysicalDisk(name, sourceFormat, disk.getVirtualSize());
+                    newDisk = destPool.createPhysicalDisk(name, sourceFormat, Storage.ProvisioningType.THIN, disk.getVirtualSize());
                 }
             }
         } else {
@@ -1071,7 +1162,7 @@ public class LibvirtStorageAdaptor implements StorageAdaptor {
                 srcFile = new QemuImgFile(sourcePath, sourceFormat);
                 try {
                     Map<String, String> info = qemu.info(srcFile);
-                    String backingFile = info.get(new String("backing_file"));
+                    String backingFile = info.get(new String("backing-file"));
                     // qcow2 templates can just be copied into place
                     if (sourceFormat.equals(destFormat) && backingFile == null && sourcePath.endsWith(".qcow2")) {
                         String result = Script.runSimpleBashScript("cp -f " + sourcePath + " " + destPath, timeout);

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/LibvirtStoragePool.java

@@ -25,6 +25,7 @@ import org.libvirt.StoragePool;
 
 import org.apache.cloudstack.utils.qemu.QemuImg.PhysicalDiskFormat;
 
+import com.cloud.storage.Storage;
 import com.cloud.storage.Storage.StoragePoolType;
 import com.cloud.utils.exception.CloudRuntimeException;
 
@@ -113,13 +114,16 @@ public class LibvirtStoragePool implements KVMStoragePool {
     }
 
     @Override
-    public KVMPhysicalDisk createPhysicalDisk(String name, PhysicalDiskFormat format, long size) {
+    public KVMPhysicalDisk createPhysicalDisk(String name,
-        return this._storageAdaptor.createPhysicalDisk(name, this, format, size);
+            PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size) {
+        return this._storageAdaptor
+                .createPhysicalDisk(name, this, format, provisioningType, size);
     }
 
     @Override
-    public KVMPhysicalDisk createPhysicalDisk(String name, long size) {
+    public KVMPhysicalDisk createPhysicalDisk(String name, Storage.ProvisioningType provisioningType, long size) {
-        return this._storageAdaptor.createPhysicalDisk(name, this, this.getDefaultFormat(), size);
+        return this._storageAdaptor.createPhysicalDisk(name, this,
+                this.getDefaultFormat(), provisioningType, size);
     }
 
     @Override

plugins/hypervisors/kvm/src/com/cloud/hypervisor/kvm/storage/StorageAdaptor.java

@@ -21,6 +21,7 @@ import java.util.Map;
 
 import org.apache.cloudstack.utils.qemu.QemuImg.PhysicalDiskFormat;
 
+import com.cloud.storage.Storage;
 import com.cloud.storage.Storage.StoragePoolType;
 
 public interface StorageAdaptor {
@@ -35,7 +36,8 @@ public interface StorageAdaptor {
 
     public boolean deleteStoragePool(String uuid);
 
-    public KVMPhysicalDisk createPhysicalDisk(String name, KVMStoragePool pool, PhysicalDiskFormat format, long size);
+    public KVMPhysicalDisk createPhysicalDisk(String name, KVMStoragePool pool,
+            PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size);
 
     // given disk path (per database) and pool, prepare disk on host
     public boolean connectPhysicalDisk(String volumePath, KVMStoragePool pool, Map<String, String> details);
@@ -49,7 +51,9 @@ public interface StorageAdaptor {
 
     public boolean deletePhysicalDisk(String uuid, KVMStoragePool pool);
 
-    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template, String name, PhysicalDiskFormat format, long size, KVMStoragePool destPool, int timeout);
+    public KVMPhysicalDisk createDiskFromTemplate(KVMPhysicalDisk template,
+            String name, PhysicalDiskFormat format, Storage.ProvisioningType provisioningType, long size,
+            KVMStoragePool destPool, int timeout);
 
     public KVMPhysicalDisk createTemplateFromDisk(KVMPhysicalDisk disk, String name, PhysicalDiskFormat format, long size, KVMStoragePool destPool);
 

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

@@ -16,11 +16,16 @@
 // under the License.
 package org.apache.cloudstack.utils.qemu;
 
-import java.util.HashMap;
 import java.util.Map;
+import com.cloud.storage.Storage;
 
-import com.cloud.utils.script.OutputInterpreter;
 import com.cloud.utils.script.Script;
+import com.cloud.utils.script.OutputInterpreter;
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import org.apache.commons.lang.NotImplementedException;
+
+import java.lang.reflect.Type;
 
 public class QemuImg {
 
@@ -43,6 +48,35 @@ public class QemuImg {
         }
     }
 
+    public static enum PreallocationType {
+        Off("off"),
+        Metadata("metadata"),
+        Full("full");
+
+        private final String preallocationType;
+
+        private PreallocationType(String preallocationType){
+            this.preallocationType = preallocationType;
+        }
+
+        public String toString(){
+            return this.preallocationType;
+        }
+
+        public static PreallocationType getPreallocationType(Storage.ProvisioningType provisioningType){
+            switch (provisioningType){
+                case THIN:
+                    return PreallocationType.Off;
+                case SPARSE:
+                    return PreallocationType.Metadata;
+                case FAT:
+                    return PreallocationType.Full;
+                default:
+                    throw new NotImplementedException();
+            }
+        }
+    }
+
     public QemuImg(int timeout) {
         this.timeout = timeout;
     }
@@ -251,9 +285,9 @@ public class QemuImg {
      * Qemu-img returns human readable output, but this method does it's best
      * to turn that into machine readeable data.
      *
-     * Spaces in keys are replaced by underscores (_).
+     * Spaces in keys are replaced by hyphen-minus (-).
-     * Sizes (virtual_size and disk_size) are returned in bytes
+     * Sizes (virtual-size and disk-size) are returned in bytes
-     * Paths (image and backing_file) are the absolute path to the file
+     * Paths (image and backing-file) are the absolute path to the file
      *
      * @param file
      *            A QemuImgFile object containing the file to get the information from
@@ -262,6 +296,8 @@ public class QemuImg {
     public Map<String, String> info(QemuImgFile file) throws QemuImgException {
         Script s = new Script(_qemuImgPath);
         s.add("info");
+        s.add("--output");
+        s.add("json");
         s.add(file.getFileName());
         OutputInterpreter.AllLinesParser parser = new OutputInterpreter.AllLinesParser();
         String result = s.execute(parser);
@@ -269,24 +305,9 @@ public class QemuImg {
             throw new QemuImgException(result);
         }
 
-        HashMap<String, String> info = new HashMap<String, String>();
+        Type stringStringMap = new TypeToken<Map<String, String>>(){}.getType();
-        String[] outputBuffer = parser.getLines().trim().split("\n");
+        Gson gson = new Gson();
-        for (int i = 0; i < outputBuffer.length; i++) {
+        return gson.fromJson(parser.getLines(), stringStringMap);
-            String[] lineBuffer = outputBuffer[i].split(":", 2);
-            if (lineBuffer.length == 2) {
-                String key = lineBuffer[0].trim().replace(" ", "_");
-                String value = null;
-
-                if (key.equals("virtual_size")) {
-                    value = lineBuffer[1].trim().replaceAll("^.*\\(([0-9]+).*$", "$1");
-                } else {
-                    value = lineBuffer[1].trim();
-                }
-
-                info.put(key, value);
-            }
-        }
-        return info;
     }
 
     /* List, apply, create or delete snapshots in image */

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

@@ -17,9 +17,11 @@
 package org.apache.cloudstack.utils.qemu;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 
 import java.io.File;
+import com.cloud.utils.script.Script;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
@@ -29,6 +31,7 @@ import org.junit.Test;
 
 import org.apache.cloudstack.utils.qemu.QemuImg.PhysicalDiskFormat;
 
+
 @Ignore
 public class QemuImgTest {
 
@@ -48,7 +51,7 @@ public class QemuImgTest {
             fail("We didn't get any information back from qemu-img");
         }
 
-        Long infoSize = Long.parseLong(info.get(new String("virtual_size")));
+        Long infoSize = Long.parseLong(info.get(new String("virtual-size")));
         assertEquals(Long.valueOf(size), Long.valueOf(infoSize));
 
         String infoPath = info.get(new String("image"));
@@ -75,13 +78,13 @@ public class QemuImgTest {
         qemu.create(file, options);
         Map<String, String> info = qemu.info(file);
 
-        Long infoSize = Long.parseLong(info.get(new String("virtual_size")));
+        Long infoSize = Long.parseLong(info.get(new String("virtual-size")));
         assertEquals(Long.valueOf(size), Long.valueOf(infoSize));
 
         String infoPath = info.get(new String("image"));
         assertEquals(filename, infoPath);
 
-        String infoClusterSize = info.get(new String("cluster_size"));
+        String infoClusterSize = info.get(new String("cluster-size"));
         assertEquals(clusterSize, infoClusterSize);
 
         File f = new File(filename);
@@ -89,6 +92,31 @@ public class QemuImgTest {
 
     }
 
+    @Test
+    public void testCreateSparseVolume() throws QemuImgException {
+        String filename = "/tmp/" + UUID.randomUUID() + ".qcow2";
+
+        /* 10TB virtual_size */
+        long size = 10995116277760l;
+        QemuImgFile file = new QemuImgFile(filename, size, PhysicalDiskFormat.QCOW2);
+        String preallocation = "metadata";
+        Map<String, String> options = new HashMap<String, String>();
+
+        options.put("preallocation", preallocation);
+
+        QemuImg qemu = new QemuImg(0);
+        qemu.create(file, options);
+
+        String allocatedSize = Script.runSimpleBashScript(String.format("ls -alhs %s | awk '{print $1}'", file));
+        String declaredSize  = Script.runSimpleBashScript(String.format("ls -alhs %s | awk '{print $6}'", file));
+
+        assertFalse(allocatedSize.equals(declaredSize));
+
+        File f = new File(filename);
+        f.delete();
+
+    }
+
     @Test
     public void testCreateAndResize() throws QemuImgException {
         String filename = "/tmp/" + UUID.randomUUID() + ".qcow2";
@@ -107,7 +135,7 @@ public class QemuImgTest {
                 fail("We didn't get any information back from qemu-img");
             }
 
-            Long infoSize = Long.parseLong(info.get(new String("virtual_size")));
+            Long infoSize = Long.parseLong(info.get(new String("virtual-size")));
             assertEquals(Long.valueOf(endSize), Long.valueOf(infoSize));
         } catch (QemuImgException e) {
             fail(e.getMessage());
@@ -136,7 +164,7 @@ public class QemuImgTest {
                 fail("We didn't get any information back from qemu-img");
             }
 
-            Long infoSize = Long.parseLong(info.get(new String("virtual_size")));
+            Long infoSize = Long.parseLong(info.get(new String("virtual-size")));
             assertEquals(Long.valueOf(startSize + increment), Long.valueOf(infoSize));
         } catch (QemuImgException e) {
             fail(e.getMessage());
@@ -164,7 +192,7 @@ public class QemuImgTest {
                 fail("We didn't get any information back from qemu-img");
             }
 
-            Long infoSize = Long.parseLong(info.get(new String("virtual_size")));
+            Long infoSize = Long.parseLong(info.get(new String("virtual-size")));
             assertEquals(Long.valueOf(startSize + increment), Long.valueOf(infoSize));
         } catch (QemuImgException e) {
             fail(e.getMessage());
@@ -227,7 +255,7 @@ public class QemuImgTest {
             fail("We didn't get any information back from qemu-img");
         }
 
-        String backingFile = info.get(new String("backing_file"));
+        String backingFile = info.get(new String("backing-file"));
         if (backingFile == null) {
             fail("The second file does not have a property backing_file! Create failed?");
         }
@@ -275,10 +303,10 @@ public class QemuImgTest {
 
         Map<String, String> info = qemu.info(destFile);
 
-        PhysicalDiskFormat infoFormat = PhysicalDiskFormat.valueOf(info.get(new String("file_format")).toUpperCase());
+        PhysicalDiskFormat infoFormat = PhysicalDiskFormat.valueOf(info.get(new String("format")).toUpperCase());
         assertEquals(destFormat, infoFormat);
 
-        Long infoSize = Long.parseLong(info.get(new String("virtual_size")));
+        Long infoSize = Long.parseLong(info.get(new String("virtual-size")));
         assertEquals(Long.valueOf(srcSize), Long.valueOf(infoSize));
 
         File sf = new File(srcFileName);

plugins/network-elements/elastic-loadbalancer/src/com/cloud/network/lb/ElasticLoadBalancerManagerImpl.java

@@ -83,6 +83,7 @@ import com.cloud.offering.ServiceOffering;
 import com.cloud.offerings.dao.NetworkOfferingDao;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
+import com.cloud.storage.Storage;
 import com.cloud.user.Account;
 import com.cloud.user.AccountService;
 import com.cloud.utils.NumbersUtil;
@@ -294,8 +295,9 @@ public class ElasticLoadBalancerManagerImpl extends ManagerBase implements Elast
         _elasticLbVmRamSize = NumbersUtil.parseInt(configs.get(Config.ElasticLoadBalancerVmMemory.key()), DEFAULT_ELB_VM_RAMSIZE);
         _elasticLbvmCpuMHz = NumbersUtil.parseInt(configs.get(Config.ElasticLoadBalancerVmCpuMhz.key()), DEFAULT_ELB_VM_CPU_MHZ);
         _elasticLbvmNumCpu = NumbersUtil.parseInt(configs.get(Config.ElasticLoadBalancerVmNumVcpu.key()), 1);
-        _elasticLbVmOffering = new ServiceOfferingVO("System Offering For Elastic LB VM", _elasticLbvmNumCpu, _elasticLbVmRamSize, _elasticLbvmCpuMHz, 0, 0, true, null,
+        _elasticLbVmOffering = new ServiceOfferingVO("System Offering For Elastic LB VM", _elasticLbvmNumCpu,
-                useLocalStorage, true, null, true, VirtualMachine.Type.ElasticLoadBalancerVm, true);
+                _elasticLbVmRamSize, _elasticLbvmCpuMHz, 0, 0, true, null, Storage.ProvisioningType.THIN, useLocalStorage,
+                true, null, true, VirtualMachine.Type.ElasticLoadBalancerVm, true);
         _elasticLbVmOffering.setUniqueName(ServiceOffering.elbVmDefaultOffUniqueName);
         _elasticLbVmOffering = _serviceOfferingDao.persistSystemServiceOffering(_elasticLbVmOffering);
 

plugins/network-elements/internal-loadbalancer/src/org/apache/cloudstack/network/lb/InternalLoadBalancerVMManagerImpl.java

@@ -92,6 +92,7 @@ import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.dao.VMTemplateDao;
+import com.cloud.storage.Storage;
 import com.cloud.user.Account;
 import com.cloud.user.AccountManager;
 import com.cloud.user.User;
@@ -379,7 +380,8 @@ public class InternalLoadBalancerVMManagerImpl extends ManagerBase implements In
             boolean useLocalStorage = Boolean.parseBoolean(configs.get(Config.SystemVMUseLocalStorage.key()));
             ServiceOfferingVO newOff =
                 new ServiceOfferingVO("System Offering For Internal LB VM", 1, InternalLoadBalancerVMManager.DEFAULT_INTERNALLB_VM_RAMSIZE,
-                    InternalLoadBalancerVMManager.DEFAULT_INTERNALLB_VM_CPU_MHZ, null, null, true, null, useLocalStorage, true, null, true,
+                    InternalLoadBalancerVMManager.DEFAULT_INTERNALLB_VM_CPU_MHZ, null, null, true, null,
+                    Storage.ProvisioningType.THIN, useLocalStorage, true, null, true,
                     VirtualMachine.Type.InternalLoadBalancerVm, true);
             newOff.setUniqueName(ServiceOffering.internalLbVmDefaultOffUniqueName);
             newOff = _serviceOfferingDao.persistSystemServiceOffering(newOff);

plugins/network-elements/internal-loadbalancer/test/org/apache/cloudstack/internallbvmmgr/InternalLBVMManagerTest.java

@@ -24,6 +24,7 @@ import java.util.List;
 
 import javax.inject.Inject;
 
+import com.cloud.storage.Storage;
 import junit.framework.TestCase;
 
 import org.junit.Before;
@@ -117,8 +118,8 @@ public class InternalLBVMManagerTest extends TestCase {
     public void setUp() {
         //mock system offering creation as it's used by configure() method called by initComponentsLifeCycle
         Mockito.when(_accountMgr.getAccount(1L)).thenReturn(new AccountVO());
-        ServiceOfferingVO off =
+        ServiceOfferingVO off = new ServiceOfferingVO("alena", 1, 1,
-            new ServiceOfferingVO("alena", 1, 1, 1, 1, 1, false, "alena", false, false, null, false, VirtualMachine.Type.InternalLoadBalancerVm, false);
+                1, 1, 1, false, "alena", Storage.ProvisioningType.THIN, false, false, null, false, VirtualMachine.Type.InternalLoadBalancerVm, false);
         off = setId(off, 1);
         Mockito.when(_svcOffDao.persistSystemServiceOffering(Matchers.any(ServiceOfferingVO.class))).thenReturn(off);
 

plugins/network-elements/internal-loadbalancer/test/org/apache/cloudstack/internallbvmmgr/InternalLBVMServiceTest.java

@@ -20,6 +20,7 @@ import java.lang.reflect.Field;
 
 import javax.inject.Inject;
 
+import com.cloud.storage.Storage;
 import junit.framework.TestCase;
 
 import org.junit.After;
@@ -87,8 +88,8 @@ public class InternalLBVMServiceTest extends TestCase {
     public void setUp() {
         //mock system offering creation as it's used by configure() method called by initComponentsLifeCycle
         Mockito.when(_accountMgr.getAccount(1L)).thenReturn(new AccountVO());
-        ServiceOfferingVO off =
+        ServiceOfferingVO off = new ServiceOfferingVO("alena", 1, 1,
-            new ServiceOfferingVO("alena", 1, 1, 1, 1, 1, false, "alena", false, false, null, false, VirtualMachine.Type.InternalLoadBalancerVm, false);
+                1, 1, 1, false, "alena", Storage.ProvisioningType.THIN, false, false, null, false, VirtualMachine.Type.InternalLoadBalancerVm, false);
         off = setId(off, 1);
         Mockito.when(_svcOffDao.persistSystemServiceOffering(Matchers.any(ServiceOfferingVO.class))).thenReturn(off);
 

server/src/com/cloud/api/query/dao/DiskOfferingJoinDaoImpl.java

@@ -59,6 +59,7 @@ public class DiskOfferingJoinDaoImpl extends GenericDaoBase<DiskOfferingJoinVO,
         diskOfferingResponse.setId(offering.getUuid());
         diskOfferingResponse.setName(offering.getName());
         diskOfferingResponse.setDisplayText(offering.getDisplayText());
+        diskOfferingResponse.setProvisioningType(offering.getProvisioningType().toString());
         diskOfferingResponse.setCreated(offering.getCreated());
         diskOfferingResponse.setDiskSize(offering.getDiskSize() / (1024 * 1024 * 1024));
         diskOfferingResponse.setMinIops(offering.getMinIops());

server/src/com/cloud/api/query/dao/ServiceOfferingJoinDaoImpl.java

@@ -59,6 +59,7 @@ public class ServiceOfferingJoinDaoImpl extends GenericDaoBase<ServiceOfferingJo
         offeringResponse.setDefaultUse(offering.isDefaultUse());
         offeringResponse.setSystemVmType(offering.getSystemVmType());
         offeringResponse.setDisplayText(offering.getDisplayText());
+        offeringResponse.setProvisioningType(offering.getProvisioningType().toString());
         offeringResponse.setCpuNumber(offering.getCpu());
         offeringResponse.setCpuSpeed(offering.getSpeed());
         offeringResponse.setMemory(offering.getRamSize());

server/src/com/cloud/api/query/dao/VolumeJoinDaoImpl.java

@@ -99,6 +99,8 @@ public class VolumeJoinDaoImpl extends GenericDaoBase<VolumeJoinVO, Long> implem
             volResponse.setVirtualMachineDisplayName(volume.getVmDisplayName());
         }
 
+        volResponse.setProvisioningType(volume.getProvisioningType().toString());
+
         // Show the virtual size of the volume
         volResponse.setSize(volume.getSize());
 

server/src/com/cloud/api/query/vo/DiskOfferingJoinVO.java

@@ -23,6 +23,7 @@ import javax.persistence.Entity;
 import javax.persistence.Id;
 import javax.persistence.Table;
 
+import com.cloud.storage.Storage;
 import org.apache.cloudstack.api.Identity;
 import org.apache.cloudstack.api.InternalIdentity;
 
@@ -46,6 +47,9 @@ public class DiskOfferingJoinVO extends BaseViewVO implements InternalIdentity,
     @Column(name = "display_text")
     private String displayText;
 
+    @Column(name = "provisioning_type")
+    Storage.ProvisioningType provisioningType;
+
     @Column(name = "disk_size")
     long diskSize;
 
@@ -136,6 +140,10 @@ public class DiskOfferingJoinVO extends BaseViewVO implements InternalIdentity,
         return displayText;
     }
 
+    public Storage.ProvisioningType getProvisioningType(){
+        return provisioningType;
+    }
+
     public long getDiskSize() {
         return diskSize;
     }

server/src/com/cloud/api/query/vo/ServiceOfferingJoinVO.java

@@ -23,6 +23,8 @@ import javax.persistence.Entity;
 import javax.persistence.Id;
 import javax.persistence.Table;
 
+import com.cloud.storage.Storage;
+
 import org.apache.cloudstack.api.Identity;
 import org.apache.cloudstack.api.InternalIdentity;
 
@@ -45,6 +47,9 @@ public class ServiceOfferingJoinVO extends BaseViewVO implements InternalIdentit
     @Column(name = "display_text")
     private String displayText;
 
+    @Column(name = "provisioning_type")
+    Storage.ProvisioningType provisioningType;
+
     @Column(name = "tags", length = 4096)
     String tags;
 
@@ -156,6 +161,10 @@ public class ServiceOfferingJoinVO extends BaseViewVO implements InternalIdentit
         return displayText;
     }
 
+    public Storage.ProvisioningType getProvisioningType(){
+        return provisioningType;
+    }
+
     public String getTags() {
         return tags;
     }

server/src/com/cloud/api/query/vo/VolumeJoinVO.java

@@ -56,6 +56,10 @@ public class VolumeJoinVO extends BaseViewVO implements ControlledViewEntity {
     @Enumerated(EnumType.STRING)
     Volume.Type volumeType;
 
+    @Column(name = "provisioning_type")
+    @Enumerated(EnumType.STRING)
+    Storage.ProvisioningType provisioningType;
+
     @Column(name = "size")
     long size;
 
@@ -292,6 +296,10 @@ public class VolumeJoinVO extends BaseViewVO implements ControlledViewEntity {
         return volumeType;
     }
 
+    public Storage.ProvisioningType getProvisioningType(){
+        return provisioningType;
+    }
+
     public long getSize() {
         return size;
     }

server/src/com/cloud/configuration/ConfigurationManagerImpl.java

@@ -175,6 +175,7 @@ import com.cloud.service.ServiceOfferingDetailsVO;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.service.dao.ServiceOfferingDetailsDao;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.test.IPRangeConfig;
@@ -2018,19 +2019,24 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
             }
         }
 
-        return createServiceOffering(userId, cmd.getIsSystem(), vmType, cmd.getServiceOfferingName(), cpuNumber, memory, cpuSpeed, cmd.getDisplayText(), localStorageRequired,
+        return createServiceOffering(userId, cmd.getIsSystem(), vmType, cmd.getServiceOfferingName(), cpuNumber, memory, cpuSpeed, cmd.getDisplayText(),
-                offerHA, limitCpuUse, volatileVm, cmd.getTags(), cmd.getDomainId(), cmd.getHostTag(), cmd.getNetworkRate(), cmd.getDeploymentPlanner(), cmd.getDetails(),
+                cmd.getProvisioningType(), localStorageRequired, offerHA, limitCpuUse, volatileVm, cmd.getTags(), cmd.getDomainId(), cmd.getHostTag(),
-                cmd.isCustomizedIops(), cmd.getMinIops(), cmd.getMaxIops(), cmd.getBytesReadRate(), cmd.getBytesWriteRate(), cmd.getIopsReadRate(), cmd.getIopsWriteRate(),
+                cmd.getNetworkRate(), cmd.getDeploymentPlanner(), cmd.getDetails(), cmd.isCustomizedIops(), cmd.getMinIops(), cmd.getMaxIops(),
-                cmd.getHypervisorSnapshotReserve());
+                cmd.getBytesReadRate(), cmd.getBytesWriteRate(), cmd.getIopsReadRate(), cmd.getIopsWriteRate(), cmd.getHypervisorSnapshotReserve());
     }
 
-    protected ServiceOfferingVO createServiceOffering(long userId, boolean isSystem, VirtualMachine.Type vmType, String name, Integer cpu, Integer ramSize, Integer speed,
+    protected ServiceOfferingVO createServiceOffering(long userId, boolean isSystem, VirtualMachine.Type vmType,
-            String displayText, boolean localStorageRequired, boolean offerHA, boolean limitResourceUse, boolean volatileVm, String tags, Long domainId, String hostTag,
+            String name, Integer cpu, Integer ramSize, Integer speed, String displayText, String provisioningType, boolean localStorageRequired,
+            boolean offerHA, boolean limitResourceUse, boolean volatileVm,  String tags, Long domainId, String hostTag,
             Integer networkRate, String deploymentPlanner, Map<String, String> details, Boolean isCustomizedIops, Long minIops, Long maxIops,
             Long bytesReadRate, Long bytesWriteRate, Long iopsReadRate, Long iopsWriteRate, Integer hypervisorSnapshotReserve) {
+
+        ProvisioningType typedProvisioningType = ProvisioningType.getProvisioningType(provisioningType);
+
         tags = StringUtils.cleanupTags(tags);
-        ServiceOfferingVO offering = new ServiceOfferingVO(name, cpu, ramSize, speed, networkRate, null, offerHA, limitResourceUse, volatileVm, displayText, localStorageRequired,
+        ServiceOfferingVO offering = new ServiceOfferingVO(name, cpu, ramSize, speed, networkRate, null, offerHA,
-                false, tags, isSystem, vmType, domainId, hostTag, deploymentPlanner);
+                limitResourceUse, volatileVm, displayText, typedProvisioningType, localStorageRequired, false, tags, isSystem, vmType,
+                domainId, hostTag, deploymentPlanner);
 
         if (isCustomizedIops != null) {
             bytesReadRate = null;
@@ -2200,8 +2206,10 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
         }
     }
 
-    protected DiskOfferingVO createDiskOffering(Long domainId, String name, String description, Long numGibibytes, String tags, boolean isCustomized, boolean localStorageRequired,
+    protected DiskOfferingVO createDiskOffering(Long domainId, String name, String description, String provisioningType,
-            boolean isDisplayOfferingEnabled, Boolean isCustomizedIops, Long minIops, Long maxIops, Long bytesReadRate, Long bytesWriteRate, Long iopsReadRate, Long iopsWriteRate,
+            Long numGibibytes, String tags, boolean isCustomized, boolean localStorageRequired,
+            boolean isDisplayOfferingEnabled, Boolean isCustomizedIops, Long minIops, Long maxIops,
+            Long bytesReadRate, Long bytesWriteRate, Long iopsReadRate, Long iopsWriteRate,
             Integer hypervisorSnapshotReserve) {
         long diskSize = 0;// special case for custom disk offerings
         if (numGibibytes != null && (numGibibytes <= 0)) {
@@ -2209,6 +2217,7 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
         } else if (numGibibytes != null && (numGibibytes > _maxVolumeSizeInGb)) {
             throw new InvalidParameterValueException("The maximum size for a disk is " + _maxVolumeSizeInGb + " Gb.");
         }
+        ProvisioningType typedProvisioningType = ProvisioningType.getProvisioningType(provisioningType);
 
         if (numGibibytes != null) {
             diskSize = numGibibytes * 1024 * 1024 * 1024;
@@ -2251,7 +2260,8 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
         }
 
         tags = StringUtils.cleanupTags(tags);
-        DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, diskSize, tags, isCustomized, isCustomizedIops, minIops, maxIops);
+        DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, typedProvisioningType, diskSize, tags, isCustomized,
+                isCustomizedIops, minIops, maxIops);
         newDiskOffering.setUseLocalStorage(localStorageRequired);
         newDiskOffering.setDisplayOffering(isDisplayOfferingEnabled);
 
@@ -2285,6 +2295,7 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
     public DiskOffering createDiskOffering(CreateDiskOfferingCmd cmd) {
         String name = cmd.getOfferingName();
         String description = cmd.getDisplayText();
+        String provisioningType = cmd.getProvisioningType();
         Long numGibibytes = cmd.getDiskSize();
         boolean isDisplayOfferingEnabled = cmd.getDisplayOffering() != null ? cmd.getDisplayOffering() : true;
         boolean isCustomized = cmd.isCustomized() != null ? cmd.isCustomized() : false; // false
@@ -2320,7 +2331,8 @@ public class ConfigurationManagerImpl extends ManagerBase implements Configurati
         Long iopsWriteRate = cmd.getIopsWriteRate();
         Integer hypervisorSnapshotReserve = cmd.getHypervisorSnapshotReserve();
 
-        return createDiskOffering(domainId, name, description, numGibibytes, tags, isCustomized, localStorageRequired, isDisplayOfferingEnabled, isCustomizedIops, minIops,
+        return createDiskOffering(domainId, name, description, provisioningType, numGibibytes, tags, isCustomized,
+                localStorageRequired, isDisplayOfferingEnabled, isCustomizedIops, minIops,
                 maxIops, bytesReadRate, bytesWriteRate, iopsReadRate, iopsWriteRate, hypervisorSnapshotReserve);
     }
 

server/src/com/cloud/consoleproxy/ConsoleProxyManagerImpl.java

@@ -31,6 +31,7 @@ import javax.inject.Inject;
 import javax.naming.ConfigurationException;
 
 import org.apache.cloudstack.config.ApiServiceConfiguration;
+
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
 import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
@@ -100,6 +101,7 @@ import com.cloud.resource.ServerResource;
 import com.cloud.resource.UnableDeleteHostException;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
+import com.cloud.storage.Storage;
 import com.cloud.storage.StoragePoolStatus;
 import com.cloud.storage.VMTemplateStorageResourceAssoc.Status;
 import com.cloud.storage.VMTemplateVO;
@@ -1267,7 +1269,8 @@ public class ConsoleProxyManagerImpl extends ManagerBase implements ConsoleProxy
             int ramSize = NumbersUtil.parseInt(_configDao.getValue("console.ram.size"), DEFAULT_PROXY_VM_RAMSIZE);
             int cpuFreq = NumbersUtil.parseInt(_configDao.getValue("console.cpu.mhz"), DEFAULT_PROXY_VM_CPUMHZ);
             _serviceOffering =
-                new ServiceOfferingVO("System Offering For Console Proxy", 1, ramSize, cpuFreq, 0, 0, false, null, useLocalStorage, true, null, true,
+                new ServiceOfferingVO("System Offering For Console Proxy", 1, ramSize, cpuFreq, 0, 0, false, null,
+                    Storage.ProvisioningType.THIN, useLocalStorage, true, null, true,
                     VirtualMachine.Type.ConsoleProxy, true);
             _serviceOffering.setUniqueName(ServiceOffering.consoleProxyDefaultOffUniqueName);
             _serviceOffering = _offeringDao.persistSystemServiceOffering(_serviceOffering);

server/src/com/cloud/network/router/VirtualNetworkApplianceManagerImpl.java

@@ -226,6 +226,7 @@ import com.cloud.resource.ResourceManager;
 import com.cloud.server.ConfigurationServer;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeVO;
@@ -741,9 +742,8 @@ public class VirtualNetworkApplianceManagerImpl extends ManagerBase implements V
         _agentMgr.registerForHostEvents(new SshKeysDistriMonitor(_agentMgr, _hostDao, _configDao), true, false, false);
 
         final boolean useLocalStorage = Boolean.parseBoolean(configs.get(Config.SystemVMUseLocalStorage.key()));
-        _offering =
+        _offering = new ServiceOfferingVO("System Offering For Software Router", 1, _routerRamSize, _routerCpuMHz, null,
-                new ServiceOfferingVO("System Offering For Software Router", 1, _routerRamSize, _routerCpuMHz, null, null, true, null, useLocalStorage, true, null, true,
+                null, true, null, ProvisioningType.THIN, useLocalStorage, true, null, true, VirtualMachine.Type.DomainRouter, true);
-                        VirtualMachine.Type.DomainRouter, true);
         _offering.setUniqueName(ServiceOffering.routerDefaultOffUniqueName);
         _offering = _serviceOfferingDao.persistSystemServiceOffering(_offering);
 

server/src/com/cloud/server/ConfigurationServerImpl.java

@@ -96,6 +96,7 @@ import com.cloud.offerings.dao.NetworkOfferingServiceMapDao;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.storage.DiskOfferingVO;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.test.IPRangeConfig;
 import com.cloud.user.Account;
@@ -218,14 +219,14 @@ public class ConfigurationServerImpl extends ManagerBase implements Configuratio
             s_logger.debug("ConfigurationServer made secondary storage copy use realhostip.");
 
             // Save default service offerings
-            createServiceOffering(User.UID_SYSTEM, "Small Instance", 1, 512, 500, "Small Instance", false, false, null);
+            createServiceOffering(User.UID_SYSTEM, "Small Instance", 1, 512, 500, "Small Instance", ProvisioningType.THIN, false, false, null);
-            createServiceOffering(User.UID_SYSTEM, "Medium Instance", 1, 1024, 1000, "Medium Instance", false, false, null);
+            createServiceOffering(User.UID_SYSTEM, "Medium Instance", 1, 1024, 1000, "Medium Instance", ProvisioningType.THIN, false, false, null);
             // Save default disk offerings
-            createdefaultDiskOffering(null, "Small", "Small Disk, 5 GB", 5, null, false, false);
+            createdefaultDiskOffering(null, "Small", "Small Disk, 5 GB", ProvisioningType.THIN, 5, null, false, false);
-            createdefaultDiskOffering(null, "Medium", "Medium Disk, 20 GB", 20, null, false, false);
+            createdefaultDiskOffering(null, "Medium", "Medium Disk, 20 GB", ProvisioningType.THIN, 20, null, false, false);
-            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", 100, null, false, false);
+            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", ProvisioningType.THIN, 100, null, false, false);
-            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", 100, null, false, false);
+            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", ProvisioningType.THIN, 100, null, false, false);
-            createdefaultDiskOffering(null, "Custom", "Custom Disk", 0, null, true, false);
+            createdefaultDiskOffering(null, "Custom", "Custom Disk", ProvisioningType.THIN, 0, null, true, false);
 
             // Save the mount parent to the configuration table
             String mountParent = getMountParent();
@@ -1026,24 +1027,24 @@ public class ConfigurationServerImpl extends ManagerBase implements Configuratio
         return pod;
     }
 
-    private DiskOfferingVO createdefaultDiskOffering(Long domainId, String name, String description, int numGibibytes, String tags, boolean isCustomized,
+    private DiskOfferingVO createdefaultDiskOffering(Long domainId, String name, String description, ProvisioningType provisioningType,
-            boolean isSystemUse) {
+            int numGibibytes, String tags, boolean isCustomized, boolean isSystemUse) {
         long diskSize = numGibibytes;
         diskSize = diskSize * 1024 * 1024 * 1024;
         tags = cleanupTags(tags);
 
-        DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, diskSize, tags, isCustomized, null, null, null);
+        DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, provisioningType, diskSize, tags, isCustomized, null, null, null);
         newDiskOffering.setUniqueName("Cloud.Com-" + name);
         newDiskOffering.setSystemUse(isSystemUse);
         newDiskOffering = _diskOfferingDao.persistDeafultDiskOffering(newDiskOffering);
         return newDiskOffering;
     }
 
-    private ServiceOfferingVO createServiceOffering(long userId, String name, int cpu, int ramSize, int speed, String displayText, boolean localStorageRequired,
+    private ServiceOfferingVO createServiceOffering(long userId, String name, int cpu, int ramSize, int speed, String displayText,
-            boolean offerHA, String tags) {
+            ProvisioningType provisioningType, boolean localStorageRequired, boolean offerHA, String tags) {
         tags = cleanupTags(tags);
         ServiceOfferingVO offering =
-                new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, offerHA, displayText, localStorageRequired, false, tags, false, null, false);
+                new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, offerHA, displayText, provisioningType, localStorageRequired, false, tags, false, null, false);
         offering.setUniqueName("Cloud.Com-" + name);
         offering = _serviceOfferingDao.persistSystemServiceOffering(offering);
         return offering;

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

@@ -330,7 +330,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         return Transaction.execute(new TransactionCallback<VolumeVO>() {
             @Override
             public VolumeVO doInTransaction(TransactionStatus status) {
-        VolumeVO volume = new VolumeVO(volumeName, zoneId, -1, -1, -1, new Long(-1), null, null, 0, Volume.Type.DATADISK);
+                VolumeVO volume = new VolumeVO(volumeName, zoneId, -1, -1, -1, new Long(-1), null, null, Storage.ProvisioningType.THIN, 0, Volume.Type.DATADISK);
                 volume.setPoolId(null);
                 volume.setDataCenterId(zoneId);
                 volume.setPodId(null);
@@ -391,6 +391,7 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
         Long zoneId = cmd.getZoneId();
         Long diskOfferingId = null;
         DiskOfferingVO diskOffering = null;
+        Storage.ProvisioningType provisioningType;
         Long size = null;
         Long minIops = null;
         Long maxIops = null;
@@ -477,6 +478,8 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
                 }
             }
 
+            provisioningType = diskOffering.getProvisioningType();
+
             if (!validateVolumeSizeRange(size)) {// convert size from mb to gb
                 // for validation
                 throw new InvalidParameterValueException("Invalid size for custom volume creation: " + size + " ,max volume size is:" + _maxVolumeSizeInGb);
@@ -502,6 +505,8 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
             size = snapshotCheck.getSize(); // ; disk offering is used for tags
             // purposes
 
+            provisioningType = diskOffering.getProvisioningType();
+
             // one step operation - create volume in VM's cluster and attach it
             // to the VM
             Long vmId = cmd.getVirtualMachineId();
@@ -545,18 +550,19 @@ public class VolumeApiServiceImpl extends ManagerBase implements VolumeApiServic
             userSpecifiedName = getRandomVolumeName();
         }
 
-        VolumeVO volume = commitVolume(cmd, caller, ownerId, displayVolume, zoneId, diskOfferingId, size, minIops, maxIops, parentVolume, userSpecifiedName,
+        VolumeVO volume = commitVolume(cmd, caller, ownerId, displayVolume, zoneId, diskOfferingId, provisioningType, size,
-                _uuidMgr.generateUuid(Volume.class, cmd.getCustomId()));
+                minIops, maxIops, parentVolume, userSpecifiedName, _uuidMgr.generateUuid(Volume.class, cmd.getCustomId()));
 
         return volume;
     }
 
-    private VolumeVO commitVolume(final CreateVolumeCmd cmd, final Account caller, final long ownerId, final Boolean displayVolume, final Long zoneId,
+    private VolumeVO commitVolume(final CreateVolumeCmd cmd, final Account caller, final long ownerId, final Boolean displayVolume,
-            final Long diskOfferingId, final Long size, final Long minIops, final Long maxIops, final VolumeVO parentVolume, final String userSpecifiedName, final String uuid) {
+            final Long zoneId, final Long diskOfferingId, final Storage.ProvisioningType provisioningType, final Long size, final Long minIops, final Long maxIops, final VolumeVO parentVolume,
+            final String userSpecifiedName, final String uuid) {
         return Transaction.execute(new TransactionCallback<VolumeVO>() {
             @Override
             public VolumeVO doInTransaction(TransactionStatus status) {
-        VolumeVO volume = new VolumeVO(userSpecifiedName, -1, -1, -1, -1, new Long(-1), null, null, 0, Volume.Type.DATADISK);
+                VolumeVO volume = new VolumeVO(userSpecifiedName, -1, -1, -1, -1, new Long(-1), null, null, provisioningType, 0, Volume.Type.DATADISK);
                 volume.setPoolId(null);
                 volume.setUuid(uuid);
                 volume.setDataCenterId(zoneId);

server/src/com/cloud/test/DatabaseConfig.java

@@ -53,6 +53,7 @@ import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDaoImpl;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.dao.DiskOfferingDaoImpl;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.utils.PropertiesUtil;
 import com.cloud.utils.component.ComponentContext;
 import com.cloud.utils.db.DB;
@@ -906,6 +907,7 @@ public class DatabaseConfig {
         long id = Long.parseLong(_currentObjectParams.get("id"));
         String name = _currentObjectParams.get("name");
         String displayText = _currentObjectParams.get("displayText");
+        ProvisioningType provisioningType = ProvisioningType.valueOf(_currentObjectParams.get("provisioningType"));
         int cpu = Integer.parseInt(_currentObjectParams.get("cpu"));
         int ramSize = Integer.parseInt(_currentObjectParams.get("ramSize"));
         int speed = Integer.parseInt(_currentObjectParams.get("speed"));
@@ -928,7 +930,8 @@ public class DatabaseConfig {
         }
 
         ServiceOfferingVO serviceOffering =
-            new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, ha, displayText, useLocalStorage, false, null, false, null, false);
+            new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, ha, displayText,
+                    provisioningType, useLocalStorage, false, null, false, null, false);
 
         Long bytesReadRate = Long.parseLong(_currentObjectParams.get("bytesReadRate"));
         if ((bytesReadRate != null) && (bytesReadRate > 0))
@@ -971,6 +974,7 @@ public class DatabaseConfig {
         long domainId = Long.parseLong(_currentObjectParams.get("domainId"));
         String name = _currentObjectParams.get("name");
         String displayText = _currentObjectParams.get("displayText");
+        ProvisioningType provisioningType = ProvisioningType.valueOf(_currentObjectParams.get("provisioningtype"));
         long diskSpace = Long.parseLong(_currentObjectParams.get("diskSpace"));
         diskSpace = diskSpace * 1024 * 1024;
 //        boolean mirroring = Boolean.parseBoolean(_currentObjectParams.get("mirrored"));
@@ -990,7 +994,7 @@ public class DatabaseConfig {
             newTags.delete(newTags.length() - 1, newTags.length());
             tags = newTags.toString();
         }
-        DiskOfferingVO diskOffering = new DiskOfferingVO(domainId, name, displayText, diskSpace, tags, false, null, null, null);
+        DiskOfferingVO diskOffering = new DiskOfferingVO(domainId, name, displayText, provisioningType, diskSpace, tags, false, null, null, null);
         diskOffering.setUseLocalStorage(local);
 
         Long bytesReadRate = Long.parseLong(_currentObjectParams.get("bytesReadRate"));

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

@@ -116,7 +116,7 @@ public class VolumeApiServiceImplTest {
         TransactionLegacy txn = TransactionLegacy.open("runVolumeDaoImplTest");
         try {
             // volume of running vm id=1
-            VolumeVO volumeOfRunningVm = new VolumeVO("root", 1L, 1L, 1L, 1L, 1L, "root", "root", 1, null,
+            VolumeVO volumeOfRunningVm = new VolumeVO("root", 1L, 1L, 1L, 1L, 1L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             when(_svc._volsDao.findById(1L)).thenReturn(volumeOfRunningVm);
 
@@ -127,7 +127,7 @@ public class VolumeApiServiceImplTest {
             when(_svc._userVmDao.findById(1L)).thenReturn(runningVm);
 
             // volume of stopped vm id=2
-            VolumeVO volumeOfStoppedVm = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", 1, null,
+            VolumeVO volumeOfStoppedVm = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             volumeOfStoppedVm.setPoolId(1L);
             when(_svc._volsDao.findById(2L)).thenReturn(volumeOfStoppedVm);
@@ -146,7 +146,7 @@ public class VolumeApiServiceImplTest {
             hyperVVm.setDataCenterId(1L);
             when(_svc._userVmDao.findById(3L)).thenReturn(hyperVVm);
 
-            VolumeVO volumeOfStoppeHyperVVm = new VolumeVO("root", 1L, 1L, 1L, 1L, 3L, "root", "root", 1, null,
+            VolumeVO volumeOfStoppeHyperVVm = new VolumeVO("root", 1L, 1L, 1L, 1L, 3L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             volumeOfStoppeHyperVVm.setPoolId(1L);
             when(_svc._volsDao.findById(3L)).thenReturn(volumeOfStoppeHyperVVm);
@@ -158,7 +158,7 @@ public class VolumeApiServiceImplTest {
             StoragePoolVO managedPool = new StoragePoolVO();
             managedPool.setManaged(true);
             when(_svc._storagePoolDao.findById(2L)).thenReturn(managedPool);
-            VolumeVO managedPoolVolume = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", 1, null,
+            VolumeVO managedPoolVolume = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             managedPoolVolume.setPoolId(2L);
             when(_svc._volsDao.findById(4L)).thenReturn(managedPoolVolume);
@@ -177,7 +177,7 @@ public class VolumeApiServiceImplTest {
             when(correctRootVolume.getInstanceId()).thenReturn(null);
             when(_svc.volFactory.getVolume(6L)).thenReturn(correctRootVolume);
 
-            VolumeVO correctRootVolumeVO = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", 1, null,
+            VolumeVO correctRootVolumeVO = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             when(_svc._volsDao.findById(6L)).thenReturn(correctRootVolumeVO);
 
@@ -190,7 +190,7 @@ public class VolumeApiServiceImplTest {
             when(managedVolume.getPoolId()).thenReturn(2L);
             when(_svc.volFactory.getVolume(7L)).thenReturn(managedVolume);
 
-            VolumeVO managedVolume1 = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", 1, null,
+            VolumeVO managedVolume1 = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             managedVolume1.setPoolId(2L);
             managedVolume1.setDataCenterId(1L);
@@ -216,7 +216,7 @@ public class VolumeApiServiceImplTest {
             when(uploadedVolume.getState()).thenReturn(Volume.State.Uploaded);
             when(_svc.volFactory.getVolume(8L)).thenReturn(uploadedVolume);
 
-            VolumeVO upVolume = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", 1, null,
+            VolumeVO upVolume = new VolumeVO("root", 1L, 1L, 1L, 1L, 2L, "root", "root", Storage.ProvisioningType.THIN, 1, null,
                     null, "root", Volume.Type.ROOT);
             upVolume.setPoolId(1L);
             upVolume.setDataCenterId(1L);

server/test/com/cloud/vm/DeploymentPlanningManagerImplTest.java

@@ -81,6 +81,7 @@ import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.resource.ResourceManager;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.service.dao.ServiceOfferingDetailsDao;
+import com.cloud.storage.Storage.ProvisioningType;
 import com.cloud.storage.StorageManager;
 import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.storage.dao.GuestOSCategoryDao;
@@ -163,7 +164,8 @@ public class DeploymentPlanningManagerImplTest {
     @Test
     public void dataCenterAvoidTest() throws InsufficientServerCapacityException, AffinityConflictException {
         ServiceOfferingVO svcOffering =
-            new ServiceOfferingVO("testOffering", 1, 512, 500, 1, 1, false, false, false, "test dpm", false, false, null, false, VirtualMachine.Type.User, domainId,
+            new ServiceOfferingVO("testOffering", 1, 512, 500, 1, 1, false, false, false, "test dpm",
+                    ProvisioningType.THIN, false, false, null, false, VirtualMachine.Type.User, domainId,
                 null, "FirstFitPlanner");
         Mockito.when(vmProfile.getServiceOffering()).thenReturn(svcOffering);
 
@@ -177,7 +179,8 @@ public class DeploymentPlanningManagerImplTest {
     @Test
     public void plannerCannotHandleTest() throws InsufficientServerCapacityException, AffinityConflictException {
         ServiceOfferingVO svcOffering =
-            new ServiceOfferingVO("testOffering", 1, 512, 500, 1, 1, false, false, false, "test dpm", false, false, null, false, VirtualMachine.Type.User, domainId,
+            new ServiceOfferingVO("testOffering", 1, 512, 500, 1, 1, false, false, false, "test dpm",
+                    ProvisioningType.THIN, false, false, null, false, VirtualMachine.Type.User, domainId,
                 null, "UserDispersingPlanner");
         Mockito.when(vmProfile.getServiceOffering()).thenReturn(svcOffering);
 
@@ -192,7 +195,8 @@ public class DeploymentPlanningManagerImplTest {
     @Test
     public void emptyClusterListTest() throws InsufficientServerCapacityException, AffinityConflictException {
         ServiceOfferingVO svcOffering =
-            new ServiceOfferingVO("testOffering", 1, 512, 500, 1, 1, false, false, false, "test dpm", false, false, null, false, VirtualMachine.Type.User, domainId,
+            new ServiceOfferingVO("testOffering", 1, 512, 500, 1, 1, false, false, false, "test dpm",
+                ProvisioningType.THIN, false, false, null, false, VirtualMachine.Type.User, domainId,
                 null, "FirstFitPlanner");
         Mockito.when(vmProfile.getServiceOffering()).thenReturn(svcOffering);
 

server/test/com/cloud/vm/UserVmManagerTest.java

@@ -35,6 +35,7 @@ import java.lang.reflect.Field;
 import java.util.List;
 import java.util.UUID;
 
+import com.cloud.storage.Storage;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mock;
@@ -538,7 +539,8 @@ public class UserVmManagerTest {
         boolean useLocalStorage = false;
 
         ServiceOfferingVO serviceOffering =
-            new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, ha, displayText, useLocalStorage, false, null, false, null, false);
+            new ServiceOfferingVO(name, cpu, ramSize, speed, null, null, ha, displayText, Storage.ProvisioningType.THIN,
+                    useLocalStorage, false, null, false, null, false);
         return serviceOffering;
     }
 

server/test/org/apache/cloudstack/service/ServiceOfferingVOTest.java

@@ -21,6 +21,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.MockitoAnnotations;
 
+import com.cloud.storage.Storage;
 import com.cloud.service.ServiceOfferingVO;
 import com.cloud.vm.VirtualMachine;
 
@@ -31,8 +32,8 @@ public class ServiceOfferingVOTest {
     @Before
     public void setup() {
         MockitoAnnotations.initMocks(this);
-        offeringCustom = new ServiceOfferingVO("custom", null, null, 500, 10, 10, false, "custom", false, false, "", false, VirtualMachine.Type.User, false);
+        offeringCustom = new ServiceOfferingVO("custom", null, null, 500, 10, 10, false, "custom", Storage.ProvisioningType.THIN, false, false, "", false, VirtualMachine.Type.User, false);
-        offering = new ServiceOfferingVO("normal", 1, 1000, 500, 10, 10, false, "normal", false, false, "", false, VirtualMachine.Type.User, false);
+        offering = new ServiceOfferingVO("normal", 1, 1000, 500, 10, 10, false, "normal", Storage.ProvisioningType.THIN, false, false, "", false, VirtualMachine.Type.User, false);
     }
 
     // Test restoreVm when VM state not in running/stopped case

services/secondary-storage/controller/src/org/apache/cloudstack/secondarystorage/SecondaryStorageManagerImpl.java

@@ -32,6 +32,7 @@ import javax.inject.Inject;
 import javax.naming.ConfigurationException;
 
 import org.apache.cloudstack.config.ApiServiceConfiguration;
+
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
@@ -109,6 +110,7 @@ import com.cloud.storage.secondary.SecStorageVmAlertEventArgs;
 import com.cloud.storage.secondary.SecondaryStorageListener;
 import com.cloud.storage.secondary.SecondaryStorageVmAllocator;
 import com.cloud.storage.secondary.SecondaryStorageVmManager;
+import com.cloud.storage.Storage;
 import com.cloud.storage.template.TemplateConstants;
 import com.cloud.template.TemplateManager;
 import com.cloud.user.Account;
@@ -849,8 +851,8 @@ public class SecondaryStorageManagerImpl extends ManagerBase implements Secondar
             int cpuFreq = NumbersUtil.parseInt(_configDao.getValue("ssvm.cpu.mhz"), DEFAULT_SS_VM_CPUMHZ);
             _useLocalStorage = Boolean.parseBoolean(configs.get(Config.SystemVMUseLocalStorage.key()));
             _serviceOffering =
-                new ServiceOfferingVO("System Offering For Secondary Storage VM", 1, ramSize, cpuFreq, null, null, false, null, _useLocalStorage, true, null, true,
+                new ServiceOfferingVO("System Offering For Secondary Storage VM", 1, ramSize, cpuFreq, null, null, false, null,
-                    VirtualMachine.Type.SecondaryStorageVm, true);
+                        Storage.ProvisioningType.THIN,  _useLocalStorage, true, null, true, VirtualMachine.Type.SecondaryStorageVm, true);
             _serviceOffering.setUniqueName(ServiceOffering.ssvmDefaultOffUniqueName);
             _serviceOffering = _offeringDao.persistSystemServiceOffering(_serviceOffering);
 

setup/db/db/schema-440to450.sql

@@ -22,3 +22,204 @@
 -- Disable foreign key checking
 -- SET foreign_key_checks = 0;
 
+ALTER TABLE `cloud`.`volumes` ADD COLUMN `provisioning_type` VARCHAR(32) NOT NULL DEFAULT 'Off' COMMENT 'pre allocation setting of the volume';
+ALTER TABLE `cloud`.`disk_offering` ADD COLUMN `provisioning_type` VARCHAR(32) NOT NULL DEFAULT 'Off' COMMENT 'pre allocation setting of the volume';
+
+DROP VIEW IF EXISTS `cloud`.`disk_offering_view`;
+CREATE VIEW `cloud`.`disk_offering_view` AS
+    select
+        disk_offering.id,
+        disk_offering.uuid,
+        disk_offering.name,
+        disk_offering.display_text,
+        disk_offering.provisioning_type,
+        disk_offering.disk_size,
+        disk_offering.min_iops,
+        disk_offering.max_iops,
+        disk_offering.created,
+        disk_offering.tags,
+        disk_offering.customized,
+        disk_offering.customized_iops,
+        disk_offering.removed,
+        disk_offering.use_local_storage,
+        disk_offering.system_use,
+        disk_offering.hv_ss_reserve,
+        disk_offering.bytes_read_rate,
+        disk_offering.bytes_write_rate,
+        disk_offering.iops_read_rate,
+        disk_offering.iops_write_rate,
+        disk_offering.cache_mode,
+        disk_offering.sort_key,
+        disk_offering.type,
+        disk_offering.display_offering,
+        domain.id domain_id,
+        domain.uuid domain_uuid,
+        domain.name domain_name,
+        domain.path domain_path
+    from
+        `cloud`.`disk_offering`
+            left join
+        `cloud`.`domain` ON disk_offering.domain_id = domain.id
+    where
+        disk_offering.state='ACTIVE';
+
+
+DROP VIEW IF EXISTS `cloud`.`service_offering_view`;
+CREATE VIEW `cloud`.`service_offering_view` AS
+    select
+        service_offering.id,
+        disk_offering.uuid,
+        disk_offering.name,
+        disk_offering.display_text,
+        disk_offering.provisioning_type,
+        disk_offering.created,
+        disk_offering.tags,
+        disk_offering.removed,
+        disk_offering.use_local_storage,
+        disk_offering.system_use,
+        disk_offering.customized_iops,
+        disk_offering.min_iops,
+        disk_offering.max_iops,
+        disk_offering.hv_ss_reserve,
+        disk_offering.bytes_read_rate,
+        disk_offering.bytes_write_rate,
+        disk_offering.iops_read_rate,
+        disk_offering.iops_write_rate,
+        disk_offering.cache_mode,
+        service_offering.cpu,
+        service_offering.speed,
+        service_offering.ram_size,
+        service_offering.nw_rate,
+        service_offering.mc_rate,
+        service_offering.ha_enabled,
+        service_offering.limit_cpu_use,
+        service_offering.host_tag,
+        service_offering.default_use,
+        service_offering.vm_type,
+        service_offering.sort_key,
+        service_offering.is_volatile,
+        service_offering.deployment_planner,
+        domain.id domain_id,
+        domain.uuid domain_uuid,
+        domain.name domain_name,
+        domain.path domain_path
+    from
+        `cloud`.`service_offering`
+            inner join
+        `cloud`.`disk_offering` ON service_offering.id = disk_offering.id
+            left join
+        `cloud`.`domain` ON disk_offering.domain_id = domain.id
+    where
+        disk_offering.state='Active';
+
+DROP VIEW IF EXISTS `cloud`.`volume_view`;
+CREATE VIEW `cloud`.`volume_view` AS
+    select
+        volumes.id,
+        volumes.uuid,
+        volumes.name,
+        volumes.device_id,
+        volumes.volume_type,
+        volumes.provisioning_type,
+        volumes.size,
+        volumes.min_iops,
+        volumes.max_iops,
+        volumes.created,
+        volumes.state,
+        volumes.attached,
+        volumes.removed,
+        volumes.pod_id,
+        volumes.display_volume,
+        volumes.format,
+        volumes.path,
+        volumes.chain_info,
+        account.id account_id,
+        account.uuid account_uuid,
+        account.account_name account_name,
+        account.type account_type,
+        domain.id domain_id,
+        domain.uuid domain_uuid,
+        domain.name domain_name,
+        domain.path domain_path,
+        projects.id project_id,
+        projects.uuid project_uuid,
+        projects.name project_name,
+        data_center.id data_center_id,
+        data_center.uuid data_center_uuid,
+        data_center.name data_center_name,
+        data_center.networktype data_center_type,
+        vm_instance.id vm_id,
+        vm_instance.uuid vm_uuid,
+        vm_instance.name vm_name,
+        vm_instance.state vm_state,
+        vm_instance.vm_type,
+        user_vm.display_name vm_display_name,
+        volume_store_ref.size volume_store_size,
+        volume_store_ref.download_pct,
+        volume_store_ref.download_state,
+        volume_store_ref.error_str,
+        volume_store_ref.created created_on_store,
+        disk_offering.id disk_offering_id,
+        disk_offering.uuid disk_offering_uuid,
+        disk_offering.name disk_offering_name,
+        disk_offering.display_text disk_offering_display_text,
+        disk_offering.use_local_storage,
+        disk_offering.system_use,
+        disk_offering.bytes_read_rate,
+        disk_offering.bytes_write_rate,
+        disk_offering.iops_read_rate,
+        disk_offering.iops_write_rate,
+        disk_offering.cache_mode,
+        storage_pool.id pool_id,
+        storage_pool.uuid pool_uuid,
+        storage_pool.name pool_name,
+        cluster.hypervisor_type,
+        vm_template.id template_id,
+        vm_template.uuid template_uuid,
+        vm_template.extractable,
+        vm_template.type template_type,
+        resource_tags.id tag_id,
+        resource_tags.uuid tag_uuid,
+        resource_tags.key tag_key,
+        resource_tags.value tag_value,
+        resource_tags.domain_id tag_domain_id,
+        resource_tags.account_id tag_account_id,
+        resource_tags.resource_id tag_resource_id,
+        resource_tags.resource_uuid tag_resource_uuid,
+        resource_tags.resource_type tag_resource_type,
+        resource_tags.customer tag_customer,
+        async_job.id job_id,
+        async_job.uuid job_uuid,
+        async_job.job_status job_status,
+        async_job.account_id job_account_id
+    from
+        `cloud`.`volumes`
+            inner join
+        `cloud`.`account` ON volumes.account_id = account.id
+            inner join
+        `cloud`.`domain` ON volumes.domain_id = domain.id
+            left join
+        `cloud`.`projects` ON projects.project_account_id = account.id
+            left join
+        `cloud`.`data_center` ON volumes.data_center_id = data_center.id
+            left join
+        `cloud`.`vm_instance` ON volumes.instance_id = vm_instance.id
+            left join
+        `cloud`.`user_vm` ON user_vm.id = vm_instance.id
+            left join
+        `cloud`.`volume_store_ref` ON volumes.id = volume_store_ref.volume_id
+            left join
+        `cloud`.`disk_offering` ON volumes.disk_offering_id = disk_offering.id
+            left join
+        `cloud`.`storage_pool` ON volumes.pool_id = storage_pool.id
+            left join
+        `cloud`.`cluster` ON storage_pool.cluster_id = cluster.id
+            left join
+        `cloud`.`vm_template` ON volumes.template_id = vm_template.id OR volumes.iso_id = vm_template.id
+            left join
+        `cloud`.`resource_tags` ON resource_tags.resource_id = volumes.id
+            and resource_tags.resource_type = 'Volume'
+            left join
+        `cloud`.`async_job` ON async_job.instance_id = volumes.id
+            and async_job.instance_type = 'Volume'
+            and async_job.job_status = 0;

test/integration/smoke/test_disk_offerings.py

@@ -38,6 +38,18 @@ class Services:
                                         "displaytext": "Disk offering",
                                         "disksize": 1   # in GB
                                 },
+                         "sparse": {
+                                        "name": "Sparse Type Disk offering",
+                                        "displaytext": "Sparse Type Disk offering",
+                                        "disksize": 1,   # in GB
+                                        "provisioningtype" : "sparse"
+                                },
+                                 "fat": {
+                                        "name": "Fat Type Disk offering",
+                                        "displaytext": "Fat Type Disk offering",
+                                        "disksize": 1,   # in GB
+                                        "provisioningtype" : "fat"
+                                }
                          }
 
 class TestCreateDiskOffering(cloudstackTestCase):
@@ -102,6 +114,86 @@ class TestCreateDiskOffering(cloudstackTestCase):
                         )
         return
 
+    @attr(hypervisor="kvm")
+    @attr(tags = ["advanced", "basic", "eip", "sg", "advancedns", "simulator", "smoke"])
+    def test_02_create_sparse_type_disk_offering(self):
+        """Test to create  a sparse type disk offering"""
+
+        # Validate the following:
+        # 1. createDiskOfferings should return valid info for new offering
+        # 2. The Cloud Database contains the valid information
+
+        disk_offering = DiskOffering.create(
+                                        self.apiclient,
+                                        self.services["sparse"]
+                                        )
+        self.cleanup.append(disk_offering)
+
+        self.debug("Created Disk offering with ID: %s" % disk_offering.id)
+
+        list_disk_response = list_disk_offering(
+                                                self.apiclient,
+                                                id=disk_offering.id
+                                                )
+        self.assertEqual(
+                            isinstance(list_disk_response, list),
+                            True,
+                            "Check list response returns a valid list"
+                        )
+        self.assertNotEqual(
+                            len(list_disk_response),
+                            0,
+                            "Check Disk offering is created"
+                        )
+        disk_response = list_disk_response[0]
+
+        self.assertEqual(
+                            disk_response.provisioningtype,
+                            self.services["sparse"]["provisioningtype"],
+                            "Check provisionig type in createServiceOffering"
+                        )
+        return
+
+
+    @attr(hypervisor="kvm")
+    @attr(tags = ["advanced", "basic", "eip", "sg", "advancedns", "simulator", "smoke"])
+    def test_04_create_fat_type_disk_offering(self):
+        """Test to create  a sparse type disk offering"""
+
+        # Validate the following:
+        # 1. createDiskOfferings should return valid info for new offering
+        # 2. The Cloud Database contains the valid information
+
+        disk_offering = DiskOffering.create(
+                                        self.apiclient,
+                                        self.services["fat"]
+                                        )
+        self.cleanup.append(disk_offering)
+
+        self.debug("Created Disk offering with ID: %s" % disk_offering.id)
+
+        list_disk_response = list_disk_offering(
+                                                self.apiclient,
+                                                id=disk_offering.id
+                                                )
+        self.assertEqual(
+                            isinstance(list_disk_response, list),
+                            True,
+                            "Check list response returns a valid list"
+                        )
+        self.assertNotEqual(
+                            len(list_disk_response),
+                            0,
+                            "Check Disk offering is created"
+                        )
+        disk_response = list_disk_response[0]
+
+        self.assertEqual(
+                            disk_response.provisioningtype,
+                            self.services["fat"]["provisioningtype"],
+                            "Check provisionig type in createServiceOffering"
+                        )
+        return
 
 class TestDiskOfferings(cloudstackTestCase):
 

test/integration/smoke/test_volumes.py

@@ -62,6 +62,12 @@ class Services:
                                     "name": "Small",
                                     "disksize": 1
                         },
+                        "sparse_disk_offering": {
+                                    "displaytext": "Sparse",
+                                    "name": "Sparse",
+                                    "provisioningtype": "sparse",
+                                    "disksize": 1
+                        },
                         'resized_disk_offering': {
                                     "displaytext": "Resized",
                                     "name": "Resized",
@@ -102,6 +108,10 @@ class TestCreateVolume(cloudstackTestCase):
                                     cls.api_client,
                                     cls.services["disk_offering"]
                                     )
+        cls.sparse_disk_offering = DiskOffering.create(
+                                    cls.api_client,
+                                    cls.services["sparse_disk_offering"]
+                                    )
         cls.custom_disk_offering = DiskOffering.create(
                                     cls.api_client,
                                     cls.services["disk_offering"],
@@ -172,6 +182,18 @@ class TestCreateVolume(cloudstackTestCase):
             self.debug("Created a volume with ID: %s" % volume.id)
             self.volumes.append(volume)
 
+        if self.virtual_machine.hypervisor == "KVM":
+            sparse_volume = Volume.create(
+                                        self.apiClient,
+                                        self.services,
+                                        zoneid=self.zone.id,
+                                        account=self.account.name,
+                                        domainid=self.account.domainid,
+                                        diskofferingid=self.sparse_disk_offering.id
+                                        )
+            self.debug("Created a sparse volume: %s" % sparse_volume.id)
+            self.volumes.append(sparse_volume)
+
         volume = Volume.create_custom_disk(
                                     self.apiClient,
                                     self.services,

ui/dictionary.jsp

@@ -536,6 +536,7 @@ dictionary = {
 'label.disk.read.bytes': '<fmt:message key="label.disk.read.bytes" />',
 'label.disk.read.io': '<fmt:message key="label.disk.read.io" />',
 'label.disk.offering': '<fmt:message key="label.disk.offering" />',
+'label.disk.provisioningtype': '<fmt:message key="label.disk.provisioningtype" />',
 'label.disk.size': '<fmt:message key="label.disk.size" />',
 'label.disk.size.gb': '<fmt:message key="label.disk.size.gb" />',
 'label.disk.total': '<fmt:message key="label.disk.total" />',

ui/scripts/configuration.js

@@ -95,6 +95,28 @@
                                             });
                                         }
                                     },
+                                    provisioningType: {
+                                        label: 'label.disk.provisioningtype',
+                                        docID: 'helpComputeOfferingProvisioningType',
+                                        select: function(args) {
+                                            var items = [];
+                                            items.push({
+                                                id: 'thin',
+                                                description: 'thin'
+                                            });
+                                            items.push({
+                                                id: 'sparse',
+                                                description: 'sparse'
+                                            });
+                                            items.push({
+                                                id: 'fat',
+                                                description: 'fat'
+                                            });
+                                            args.response.success({
+                                                data: items
+                                            });
+                                        }
+                                    },
                                     isCustomized: {
                                         label: 'label.custom',                                       
                                         isBoolean: true,
@@ -456,6 +478,7 @@
                                     name: args.data.name,
                                     displaytext: args.data.description,
                                     storageType: args.data.storageType,
+                                    provisioningType :args.data.provisioningType,
                                     customized: (args.data.isCustomized == "on")
                                 };
                                 
@@ -721,6 +744,9 @@
                                     storagetype: {
                                         label: 'label.storage.type'
                                     },
+                                    provisioningtype: {
+                                        label: 'label.disk.provisioningtype'
+                                    },
                                     cpunumber: {
                                         label: 'label.num.cpu.cores'
                                     },
@@ -937,6 +963,28 @@
                                             });
                                         }
                                     },
+                                    provisioningType: {
+                                        label: 'label.disk.provisioningtype',
+                                        docID: 'helpDiskOfferingProvisioningType',
+                                        select: function(args) {
+                                            var items = [];
+                                            items.push({
+                                                id: 'thin',
+                                                description: 'thin'
+                                            });
+                                            items.push({
+                                                id: 'sparse',
+                                                description: 'sparse'
+                                            });
+                                            items.push({
+                                                id: 'fat',
+                                                description: 'fat'
+                                            });
+                                            args.response.success({
+                                                data: items
+                                            });
+                                        }
+                                    },
                                     cpuNumber: {
                                         label: 'label.num.cpu.cores',
                                         docID: 'helpSystemOfferingCPUCores',
@@ -1064,6 +1112,7 @@
                                     displaytext: args.data.description,
                                     systemvmtype: args.data.systemvmtype,
                                     storageType: args.data.storageType,
+                                    provisioningType: args.data.provisioningType,
                                     cpuNumber: args.data.cpuNumber,
                                     cpuSpeed: args.data.cpuSpeed,
                                     memory: args.data.memory
@@ -1271,6 +1320,9 @@
                                     storagetype: {
                                         label: 'label.storage.type'
                                     },
+                                    provisioningtype: {
+                                        label: 'label.disk.provisioningtype'
+                                    },
                                     cpunumber: {
                                         label: 'label.num.cpu.cores'
                                     },
@@ -1447,6 +1499,28 @@
                                             });
                                         }
                                     },
+                                    provisioningType: {
+                                        label: 'label.disk.provisioningtype',
+                                        docID: 'helpDiskOfferingProvisioningType',
+                                        select: function(args) {
+                                            var items = [];
+                                            items.push({
+                                                id: 'thin',
+                                                description: 'thin'
+                                            });
+                                            items.push({
+                                                id: 'sparse',
+                                                description: 'sparse'
+                                            });
+                                            items.push({
+                                                id: 'fat',
+                                                description: 'fat'
+                                            });
+                                            args.response.success({
+                                                data: items
+                                            });
+                                        }
+                                    },
                                     isCustomized: {
                                         label: 'label.custom.disk.size',
                                         docID: 'helpDiskOfferingCustomDiskSize',
@@ -1671,6 +1745,7 @@
                                     displaytext: args.data.description,
                                     storageType: args.data.storageType,
                                     cacheMode: args.data.cacheMode,
+                                    provisioningType: args.data.provisioningType,
                                     customized: (args.data.isCustomized == "on")
                                 };
 
@@ -1917,6 +1992,9 @@
                                     },
                                     storagetype: {
                                         label: 'label.storage.type'
+                                    },
+                                    provisioningtype: {
+                                        label: 'label.disk.provisioningtype'
                                     }
                                 }],
 

ui/scripts/docs.js

@@ -218,6 +218,10 @@ cloudStack.docs = {
         desc: 'Type of disk for the VM. Local storage is attached to the hypervisor host where the VM is running. Shared storage is accessible via NFS.',
         externalLink: ''
     },
+    helpComputeOfferingProvisioningType: {
+        desc: 'Provisioning type to create a volume. Thin and sparse is lazy allocation. fat is eager allocation.',
+        externalLink: ''
+    },
     helpComputeOfferingCPUCores: {
         desc: 'The number of cores which should be allocated to a VM with this offering',
         externalLink: ''
@@ -282,6 +286,10 @@ cloudStack.docs = {
         desc: 'A short description of the offering that can be displayed to users',
         externalLink: ''
     },
+    helpDiskOfferingProvisioningType: {
+        desc: 'Provisioning type to create a volume. Thin and sparse is lazy allocation. fat is eager allocation.',
+        externalLink: ''
+    },
     helpDiskOfferingStorageType: {
         desc: 'Type of disk for the VM. Local is attached to the hypervisor host where the VM is running. Shared is storage accessible via NFS.',
         externalLink: ''

ui/scripts/storage.js

@@ -1402,6 +1402,9 @@
                                     storagetype: {
                                         label: 'label.storage.type'
                                     },
+                                    provisioningtype: {
+                                        label: 'label.disk.provisioningtype'
+                                    },
                                     hypervisor: {
                                         label: 'label.hypervisor'
                                     },