CS-14277

Support for local data disk. Currently enable/disable config is at zone level, in subsequent checkins it can be made more granular.
    Following changes are made:
    - Create disk offering API now takes an extra parameter to denote storage type (local or shared). This is similar to storage type in service offering.
    - Create/delete of data volume on local storage
    - Attach/detach for local data volumes. Re-attach is allowed as long as vm host and data volume storage pool host is same.
    - Migration of VM instance is not supported if it uses local root or data volumes.
    - Migrate is not supported for local volumes.
    - Zone level config to enable/disable local storage usage for service and disk offerings.
    - Local storage gets discovered when a host is added/reconnected if zone level config is enabled. When disabled existing local storages are not removed but any new local storage is not added.
    - Deploy VM command validates service and disk offerings based on local storage config.
    - Upgrade uses the global config 'use.local.storage' to set the zone level config for local storage.
(cherry picked from commit 62710aed37606168012a0ed255a876c8e7954010)

api/src/com/cloud/api/ApiConstants.java

@@ -294,6 +294,7 @@ public class ApiConstants {
     public static final String DHCP_RANGE = "dhcprange";
     public static final String UUID = "uuid";
     public static final String SECURITY_GROUP_EANBLED = "securitygroupenabled";
+    public static final String LOCAL_STORAGE_ENABLED = "localstorageenabled";
     public static final String GUEST_IP_TYPE = "guestiptype";
     public static final String XEN_NETWORK_LABEL = "xennetworklabel";
     public static final String KVM_NETWORK_LABEL = "kvmnetworklabel";

api/src/com/cloud/api/commands/CreateDiskOfferingCmd.java

@@ -26,6 +26,7 @@ import com.cloud.api.Parameter;
 import com.cloud.api.ServerApiException;
 import com.cloud.api.response.DiskOfferingResponse;
 import com.cloud.offering.DiskOffering;
+import com.cloud.offering.ServiceOffering;
 import com.cloud.user.Account;
 
 @Implementation(description="Creates a disk offering.", responseObject=DiskOfferingResponse.class)
@@ -57,6 +58,9 @@ public class CreateDiskOfferingCmd extends BaseCmd {
     @Parameter(name=ApiConstants.DOMAIN_ID, type=CommandType.LONG, description="the ID of the containing domain, null for public offerings")
     private Long domainId; 
 
+    @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();
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -84,6 +88,11 @@ public class CreateDiskOfferingCmd extends BaseCmd {
     public Long getDomainId(){
     	return domainId;
     }
+
+    public String getStorageType() {
+        return storageType;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////

api/src/com/cloud/api/commands/CreateZoneCmd.java

@@ -73,6 +73,9 @@ public class CreateZoneCmd extends BaseCmd {
     @Parameter(name=ApiConstants.SECURITY_GROUP_EANBLED, type=CommandType.BOOLEAN, description="true if network is security group enabled, false otherwise")
     private Boolean securitygroupenabled;
 
+    @Parameter(name=ApiConstants.LOCAL_STORAGE_ENABLED, type=CommandType.BOOLEAN, description="true if local storage offering enabled, false otherwise")
+    private Boolean localStorageEnabled;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -124,6 +127,12 @@ public class CreateZoneCmd extends BaseCmd {
         return securitygroupenabled;
     }
 
+    public Boolean getLocalStorageEnabled() {
+        if (localStorageEnabled == null) {
+            return false;
+        }
+        return localStorageEnabled;
+    }
 
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////

api/src/com/cloud/api/commands/DeployVMCmd.java

@@ -375,13 +375,23 @@ public class DeployVMCmd extends BaseAsyncCreateCmd {
                 throw new InvalidParameterValueException("Unable to use template " + templateId);
             }
 
+            DiskOffering diskOffering = null;
             if (diskOfferingId != null) {
-                DiskOffering diskOffering = _configService.getDiskOffering(diskOfferingId);
+                diskOffering = _configService.getDiskOffering(diskOfferingId);
                 if (diskOffering == null) {
                     throw new InvalidParameterValueException("Unable to find disk offering " + diskOfferingId);
                 }
             }
 
+            if (!zone.isLocalStorageEnabled()) {
+                if (serviceOffering.getUseLocalStorage()) {
+                    throw new InvalidParameterValueException("Zone is not configured to use local storage but service offering " + serviceOffering.getName() + " uses it");
+                }
+                if (diskOffering != null && diskOffering.getUseLocalStorage()) {
+                    throw new InvalidParameterValueException("Zone is not configured to use local storage but disk offering " + diskOffering.getName() + " uses it");
+                }
+            }
+
             UserVm vm = null;
             if (getHypervisor() == HypervisorType.BareMetal) {
                 vm = _bareMetalVmService.createVirtualMachine(this);

api/src/com/cloud/api/commands/UpdateZoneCmd.java

@@ -82,6 +82,9 @@ public class UpdateZoneCmd extends BaseCmd {
     @Parameter(name=ApiConstants.DNS_SEARCH_ORDER, type=CommandType.LIST, collectionType = CommandType.STRING, description="the dns search order list")
     private List<String> dnsSearchOrder;   
 
+    @Parameter(name=ApiConstants.LOCAL_STORAGE_ENABLED, type=CommandType.BOOLEAN, description="true if local storage offering enabled, false otherwise")
+    private Boolean localStorageEnabled;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -137,6 +140,11 @@ public class UpdateZoneCmd extends BaseCmd {
     public List<String> getDnsSearchOrder() {
         return dnsSearchOrder;
     }   
+
+    public Boolean getLocalStorageEnabled() {
+        return localStorageEnabled;
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////

api/src/com/cloud/api/response/DiskOfferingResponse.java

@@ -51,6 +51,9 @@ public class DiskOfferingResponse extends BaseResponse {
     @SerializedName(ApiConstants.TAGS) @Param(description="the tags for the disk offering")
     private String tags;
 
+    @SerializedName("storagetype") @Param(description="the storage type for this disk offering")
+    private String storageType;
+
     public Long getId() {
         return id.getValue();
     }
@@ -123,4 +126,11 @@ public class DiskOfferingResponse extends BaseResponse {
         this.customized = customized;
     }
 
+    public String getStorageType() {
+        return storageType;
+    }
+
+    public void setStorageType(String storageType) {
+        this.storageType = storageType;
+    }
 }

api/src/com/cloud/api/response/ZoneResponse.java

@@ -86,6 +86,9 @@ public class ZoneResponse extends BaseResponse {
     @SerializedName("capacity")  @Param(description="the capacity of the Zone", responseObject = CapacityResponse.class)
     private List<CapacityResponse> capacitites;
 
+    @SerializedName(ApiConstants.LOCAL_STORAGE_ENABLED) @Param(description="true if local storage offering enabled, false otherwise")
+    private boolean localStorageEnabled;
+
     public void setId(Long id) {
         this.id.setValue(id);
     }
@@ -165,4 +168,8 @@ public class ZoneResponse extends BaseResponse {
 	public void setDomainName(String domainName) {
 		this.domainName = domainName;
 	}
+
+    public void setLocalStorageEnabled(boolean localStorageEnabled) {
+        this.localStorageEnabled = localStorageEnabled;
+    }
 }

api/src/com/cloud/dc/DataCenter.java

@@ -75,4 +75,5 @@ public interface DataCenter extends Grouping {
 
     String getZoneToken();
 
+    boolean isLocalStorageEnabled();
 }

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

@@ -1124,6 +1124,7 @@ label.linklocal.ip=Link Local IP Adddress
 label.load.balancer=Load Balancer
 label.loading=Loading
 label.local=Local
+label.local.storage.enabled=Local Storage Enabled
 label.login=Login
 label.logout=Logout
 label.lun=LUN

console-proxy/pom.xml

@@ -138,7 +138,7 @@
           </execution>
         </executions>
       </plugin>
-      <plugin>
+      <!--plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>exec-maven-plugin</artifactId>
         <version>1.2.1</version>
@@ -163,7 +163,7 @@
             <argument>authorized_keys</argument>
           </arguments>
         </configuration>
-      </plugin>
+      </plugin-->
     </plugins>
   </build>
 </project>

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

@@ -463,6 +463,7 @@ public class ApiResponseHelper implements ResponseGenerator {
         }
         diskOfferingResponse.setTags(offering.getTags());
         diskOfferingResponse.setCustomized(offering.isCustomized());
+        diskOfferingResponse.setStorageType(offering.getUseLocalStorage() ? ServiceOffering.StorageType.local.toString() : ServiceOffering.StorageType.shared.toString());
         diskOfferingResponse.setObjectName("diskoffering");
         return diskOfferingResponse;
     }
@@ -954,6 +955,7 @@ public class ApiResponseHelper implements ResponseGenerator {
         zoneResponse.setId(dataCenter.getId());
         zoneResponse.setName(dataCenter.getName());
         zoneResponse.setSecurityGroupsEnabled(ApiDBUtils.isSecurityGroupEnabledInZone(dataCenter.getId()));
+        zoneResponse.setLocalStorageEnabled(dataCenter.isLocalStorageEnabled());
 
         if ((dataCenter.getDescription() != null) && !dataCenter.getDescription().equalsIgnoreCase("null")) {
             zoneResponse.setDescription(dataCenter.getDescription());
@@ -1121,24 +1123,6 @@ public class ApiResponseHelper implements ResponseGenerator {
         
         populateOwner(volResponse, volume);
 
-        String storageType;
-        try {
-            if (volume.getPoolId() == null) {
-                if (volume.getState() == Volume.State.Allocated || volume.getState() == Volume.State.UploadOp) {
-                    /* set it as shared, so the UI can attach it to VM */
-                    storageType = "shared";
-                } else {
-                    storageType = "unknown";
-                }
-            } else {
-                storageType = ApiDBUtils.volumeIsOnSharedStorage(volume.getId()) ? ServiceOffering.StorageType.shared.toString() : ServiceOffering.StorageType.local.toString();
-            }
-        } catch (InvalidParameterValueException e) {
-            s_logger.error(e.getMessage(), e);
-            throw new ServerApiException(BaseCmd.INTERNAL_ERROR, "Volume " + volume.getName() + " does not have a valid ID");
-        }
-
-        volResponse.setStorageType(storageType);
         if (volume.getVolumeType().equals(Volume.Type.ROOT)) {
             volResponse.setServiceOfferingId(volume.getDiskOfferingId());
         } else {
@@ -1153,6 +1137,7 @@ public class ApiResponseHelper implements ResponseGenerator {
             volResponse.setDiskOfferingName(diskOffering.getName());
             volResponse.setDiskOfferingDisplayText(diskOffering.getDisplayText());
         }
+        volResponse.setStorageType(diskOffering.getUseLocalStorage() ? ServiceOffering.StorageType.local.toString() : ServiceOffering.StorageType.shared.toString());
 
         Long poolId = volume.getPoolId();
         String poolName = (poolId == null) ? "none" : ApiDBUtils.findStoragePoolById(poolId).getName();

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

@@ -191,7 +191,6 @@ public enum Config {
 	ManagementHostIPAdr("Advanced", ManagementServer.class, String.class, "host", "localhost", "The ip address of management server", null),
 	ManagementNetwork("Advanced", ManagementServer.class, String.class, "management.network.cidr", null, "The cidr of management server network", null),
 	EventPurgeDelay("Advanced", ManagementServer.class, Integer.class, "event.purge.delay", "15", "Events older than specified number days will be purged. Set this value to 0 to never delete events", null),
-    UseLocalStorage("Advanced", ManagementServer.class, Boolean.class, "use.local.storage", "false", "Should we use the local storage if it's available?", null),
 	SecStorageVmMTUSize("Advanced", AgentManager.class, Integer.class, "secstorage.vm.mtu.size", String.valueOf(SecondaryStorageVmManager.DEFAULT_SS_VM_MTUSIZE), "MTU size (in Byte) of storage network in secondary storage vms", null),
 	MaxTemplateAndIsoSize("Advanced",  ManagementServer.class, Long.class, "max.template.iso.size", "50", "The maximum size for a downloaded template or ISO (in GB).", null),
 	SecStorageAllowedInternalDownloadSites("Advanced", ManagementServer.class, String.class, "secstorage.allowed.internal.sites", null, "Comma separated list of cidrs internal to the datacenter that can host template download servers, please note 0.0.0.0 is not a valid site", null),

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

@@ -93,7 +93,7 @@ public interface ConfigurationManager extends ConfigurationService, Manager {
      * @param isCustomized
      * @return newly created disk offering
      */
-    DiskOfferingVO createDiskOffering(Long domainId, String name, String description, Long numGibibytes, String tags, boolean isCustomized);
+    DiskOfferingVO createDiskOffering(Long domainId, String name, String description, Long numGibibytes, String tags, boolean isCustomized, boolean localStorageRequired);
 
     /**
      * Creates a new pod
@@ -133,7 +133,7 @@ public interface ConfigurationManager extends ConfigurationService, Manager {
      * @throws
      */
     DataCenterVO createZone(long userId, String zoneName, String dns1, String dns2, String internalDns1, String internalDns2, String guestCidr, String domain, Long domainId, NetworkType zoneType, String allocationState,
-            String networkDomain, boolean isSecurityGroupEnabled);
+            String networkDomain, boolean isSecurityGroupEnabled, boolean isLocalStorageEnabled);
 
     /**
      * Deletes a VLAN from the database, along with all of its IP addresses. Will not delete VLANs that have allocated

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

@@ -1364,6 +1364,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
         String dhcpProvider = cmd.getDhcpProvider();
         Map<?, ?> detailsMap = cmd.getDetails();
         String networkDomain = cmd.getDomain();
+        Boolean localStorageEnabled = cmd.getLocalStorageEnabled();
 
         Map<String, String> newDetails = new HashMap<String, String>();
         if (detailsMap != null) {
@@ -1470,6 +1471,9 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
         zone.setInternalDns1(internalDns1);
         zone.setInternalDns2(internalDns2);
         zone.setGuestNetworkCidr(guestCidr);
+        if (localStorageEnabled != null) {
+            zone.setLocalStorageEnabled(localStorageEnabled.booleanValue());
+        }
 
         if (networkDomain != null) {
             if (networkDomain.isEmpty()) {
@@ -1543,7 +1547,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
     @Override
     @DB
     public DataCenterVO createZone(long userId, String zoneName, String dns1, String dns2, String internalDns1, String internalDns2, String guestCidr, String domain, Long domainId,
-            NetworkType zoneType, String allocationStateStr, String networkDomain, boolean isSecurityGroupEnabled) {
+            NetworkType zoneType, String allocationStateStr, String networkDomain, boolean isSecurityGroupEnabled, boolean isLocalStorageEnabled) {
 
         // checking the following params outside checkzoneparams method as we do
         // not use these params for updatezone
@@ -1569,7 +1573,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
         try {
             txn.start();
             // Create the new zone in the database
-            DataCenterVO zone = new DataCenterVO(zoneName, null, dns1, dns2, internalDns1, internalDns2, guestCidr, domain, domainId, zoneType, zoneToken, networkDomain, isSecurityGroupEnabled);
+            DataCenterVO zone = new DataCenterVO(zoneName, null, dns1, dns2, internalDns1, internalDns2, guestCidr, domain, domainId, zoneType, zoneToken, networkDomain, isSecurityGroupEnabled, isLocalStorageEnabled);
             if (allocationStateStr != null && !allocationStateStr.isEmpty()) {
                 Grouping.AllocationState allocationState = Grouping.AllocationState.valueOf(allocationStateStr);
                 zone.setAllocationState(allocationState);
@@ -1649,6 +1653,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
         String allocationState = cmd.getAllocationState();
         String networkDomain = cmd.getDomain();
         boolean isSecurityGroupEnabled = cmd.getSecuritygroupenabled();
+        boolean isLocalStorageEnabled = cmd.getLocalStorageEnabled();
 
         if (allocationState == null) {
             allocationState = Grouping.AllocationState.Disabled.toString();
@@ -1682,7 +1687,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
         }
 
         return createZone(userId, zoneName, dns1, dns2, internalDns1, internalDns2, guestCidr, domainVO != null ? domainVO.getName() : null, domainId, zoneType, allocationState, networkDomain,
-                isSecurityGroupEnabled);
+                isSecurityGroupEnabled, isLocalStorageEnabled);
     }
 
     @Override
@@ -1857,7 +1862,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
 
     @Override
     @ActionEvent(eventType = EventTypes.EVENT_DISK_OFFERING_CREATE, eventDescription = "creating disk offering")
-    public DiskOfferingVO createDiskOffering(Long domainId, String name, String description, Long numGibibytes, String tags, boolean isCustomized) {
+    public DiskOfferingVO createDiskOffering(Long domainId, String name, String description, Long numGibibytes, String tags, boolean isCustomized, boolean localStorageRequired) {
         long diskSize = 0;// special case for custom disk offerings
         if (numGibibytes != null && (numGibibytes <= 0)) {
             throw new InvalidParameterValueException("Please specify a disk size of at least 1 Gb.");
@@ -1875,6 +1880,7 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
 
         tags = cleanupTags(tags);
         DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, diskSize, tags, isCustomized);
+        newDiskOffering.setUseLocalStorage(localStorageRequired);
         UserContext.current().setEventDetails("Disk offering id=" + newDiskOffering.getId());
         DiskOfferingVO offering = _diskOfferingDao.persist(newDiskOffering);
         if (offering != null) {
@@ -1904,7 +1910,17 @@ public class ConfigurationManagerImpl implements ConfigurationManager, Configura
             throw new InvalidParameterValueException("Disksize is required for non-customized disk offering");
         }
 
-        return createDiskOffering(domainId, name, description, numGibibytes, tags, isCustomized);
+        boolean localStorageRequired = false;
+        String storageType = cmd.getStorageType();
+        if (storageType != null) {
+            if (storageType.equalsIgnoreCase(ServiceOffering.StorageType.local.toString())) {
+                localStorageRequired = true;
+            } else if (!storageType.equalsIgnoreCase(ServiceOffering.StorageType.shared.toString())) {
+                throw new InvalidParameterValueException("Invalid storage type " + storageType + " specified, valid types are: 'local' and 'shared'");
+            }
+        }
+
+        return createDiskOffering(domainId, name, description, numGibibytes, tags, isCustomized, localStorageRequired);
     }
 
     @Override

server/src/com/cloud/dc/DataCenterVO.java

@@ -127,6 +127,9 @@ public class DataCenterVO implements DataCenter, Identity {
     @Column(name="is_security_group_enabled")
     boolean securityGroupEnabled;
 
+    @Column(name="is_local_storage_enabled")
+    boolean localStorageEnabled;
+
     @Override
     public String getDnsProvider() {
         return dnsProvider;
@@ -173,13 +176,13 @@ public class DataCenterVO implements DataCenter, Identity {
     }
     
     public DataCenterVO(long id, String name, String description, String dns1, String dns2, String dns3, String dns4, String guestCidr, String domain, Long domainId, NetworkType zoneType, String zoneToken, String domainSuffix) {
-        this(name, description, dns1, dns2, dns3, dns4, guestCidr, domain, domainId, zoneType, zoneToken, domainSuffix, false);
+        this(name, description, dns1, dns2, dns3, dns4, guestCidr, domain, domainId, zoneType, zoneToken, domainSuffix, false, false);
         this.id = id;
         this.allocationState = Grouping.AllocationState.Enabled;
         this.uuid = UUID.randomUUID().toString();
 	}
 
-    public DataCenterVO(String name, String description, String dns1, String dns2, String dns3, String dns4, String guestCidr, String domain, Long domainId, NetworkType zoneType, String zoneToken, String domainSuffix, boolean securityGroupEnabled) {
+    public DataCenterVO(String name, String description, String dns1, String dns2, String dns3, String dns4, String guestCidr, String domain, Long domainId, NetworkType zoneType, String zoneToken, String domainSuffix, boolean securityGroupEnabled, boolean localStorageEnabled) {
         this.name = name;
         this.description = description;
         this.dns1 = dns1;
@@ -192,7 +195,7 @@ public class DataCenterVO implements DataCenter, Identity {
         this.networkType = zoneType;
         this.allocationState = Grouping.AllocationState.Enabled;
         this.securityGroupEnabled = securityGroupEnabled;
-        
+        this.localStorageEnabled = localStorageEnabled;
 
         if (zoneType == NetworkType.Advanced) {
             loadBalancerProvider = Provider.VirtualRouter.getName();
@@ -345,6 +348,15 @@ public class DataCenterVO implements DataCenter, Identity {
         this.securityGroupEnabled = enabled;
     }
 
+    @Override
+    public boolean isLocalStorageEnabled() {
+        return localStorageEnabled;
+    }
+
+    public void setLocalStorageEnabled(boolean enabled) {
+        this.localStorageEnabled = enabled;
+    }
+
     @Override
     public Map<String, String> getDetails() {
         return details;

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

@@ -31,6 +31,8 @@ import com.cloud.agent.api.StoragePoolInfo;
 import com.cloud.capacity.Capacity;
 import com.cloud.capacity.CapacityVO;
 import com.cloud.capacity.dao.CapacityDao;
+import com.cloud.dc.DataCenterVO;
+import com.cloud.dc.dao.DataCenterDao;
 import com.cloud.exception.ConnectionException;
 import com.cloud.host.HostVO;
 import com.cloud.host.Status;
@@ -48,6 +50,7 @@ public class LocalStoragePoolListener implements Listener {
     @Inject StoragePoolHostDao _storagePoolHostDao;
     @Inject CapacityDao _capacityDao;
     @Inject StorageManager _storageMgr;
+    @Inject DataCenterDao _dcDao;
 
     @Override
     public int getTimeout() {
@@ -87,6 +90,11 @@ public class LocalStoragePoolListener implements Listener {
             return;
         }
 
+        DataCenterVO dc = _dcDao.findById(host.getDataCenterId());
+        if (dc == null || !dc.isLocalStorageEnabled()) {
+            return;
+        }
+
         try {
             StoragePoolVO pool = _storagePoolDao.findPoolByHostPath(host.getDataCenterId(), host.getPodId(), pInfo.getHost(), pInfo.getHostPath(), pInfo.getUuid());
         	if(pool == null && host.getHypervisorType() == HypervisorType.VMware) {

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

@@ -126,6 +126,7 @@ import com.cloud.host.dao.HostDao;
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.hypervisor.HypervisorGuruManager;
 import com.cloud.network.NetworkManager;
+import com.cloud.offering.ServiceOffering;
 import com.cloud.org.Grouping;
 import com.cloud.org.Grouping.AllocationState;
 import com.cloud.projects.Project.ListProjectResourcesCriteria;
@@ -457,13 +458,13 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         return false;
     }
 
-    protected StoragePoolVO findStoragePool(DiskProfile dskCh, final DataCenterVO dc, HostPodVO pod, Long clusterId, VMInstanceVO vm, final Set<StoragePool> avoid) {
+    protected StoragePoolVO findStoragePool(DiskProfile dskCh, final DataCenterVO dc, HostPodVO pod, Long clusterId, Long hostId, VMInstanceVO vm, final Set<StoragePool> avoid) {
 
         VirtualMachineProfile<VMInstanceVO> profile = new VirtualMachineProfileImpl<VMInstanceVO>(vm);
         Enumeration<StoragePoolAllocator> en = _storagePoolAllocators.enumeration();
         while (en.hasMoreElements()) {
             final StoragePoolAllocator allocator = en.nextElement();
-            final List<StoragePool> poolList = allocator.allocateToPool(dskCh, profile, dc.getId(), pod.getId(), clusterId, avoid, 1);
+            final List<StoragePool> poolList = allocator.allocateToPool(dskCh, profile, dc.getId(), pod.getId(), clusterId, hostId, avoid, 1);
             if (poolList != null && !poolList.isEmpty()) {
                 return (StoragePoolVO) poolList.get(0);
             }
@@ -571,7 +572,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         while ((pod = _resourceMgr.findPod(null, null, dc, account.getId(), podsToAvoid)) != null) {
             podsToAvoid.add(pod.first().getId());
             // Determine what storage pool to store the volume in
-            while ((pool = findStoragePool(dskCh, dc, pod.first(), null, null, poolsToAvoid)) != null) {
+            while ((pool = findStoragePool(dskCh, dc, pod.first(), null, null, null, poolsToAvoid)) != null) {
                 poolsToAvoid.add(pool);
                 volumeFolder = pool.getPath();
                 if (s_logger.isDebugEnabled()) {
@@ -741,7 +742,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
     	DiskProfile dskCh = createDiskCharacteristics(volume, template, dc, diskOffering);
     	dskCh.setHyperType(vm.getHypervisorType());
     	// Find a suitable storage to create volume on 
-    	StoragePoolVO destPool = findStoragePool(dskCh, dc, pod, clusterId, vm, avoidPools);
+	StoragePoolVO destPool = findStoragePool(dskCh, dc, pod, clusterId, null, vm, avoidPools);
     	
     	// Copy the volume from secondary storage to the destination storage pool    	  	
     	stateTransitTo(volume, Event.CopyRequested);
@@ -818,7 +819,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
                 break;
             }
 
-            pool = findStoragePool(dskCh, dc, pod, clusterId, vm, avoidPools);
+            pool = findStoragePool(dskCh, dc, pod, clusterId, vm.getHostId(), vm, avoidPools);
             if (pool == null) {
                 s_logger.warn("Unable to find storage poll when create volume " + volume.getName());
                 break;
@@ -988,10 +989,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         int wrks = NumbersUtil.parseInt(workers, 10);
         _executor = Executors.newScheduledThreadPool(wrks, new NamedThreadFactory("StorageManager-Scavenger"));
 
-        boolean localStorage = Boolean.parseBoolean(configs.get(Config.UseLocalStorage.key()));
-        if (localStorage) {
         _agentMgr.registerForHostEvents(ComponentLocator.inject(LocalStoragePoolListener.class), true, false, false);
-        }
 
         String maxVolumeSizeInGbString = configDao.getValue("storage.max.volume.size");
         _maxVolumeSizeInGb = NumbersUtil.parseLong(maxVolumeSizeInGbString, 2000);
@@ -1713,7 +1711,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         dskCh.setHyperType(dataDiskHyperType);
         DataCenterVO destPoolDataCenter = _dcDao.findById(destPoolDcId);
         HostPodVO destPoolPod = _podDao.findById(destPoolPodId);
-        StoragePoolVO destPool = findStoragePool(dskCh, destPoolDataCenter, destPoolPod, destPoolClusterId, null, new HashSet<StoragePool>());
+        StoragePoolVO destPool = findStoragePool(dskCh, destPoolDataCenter, destPoolPod, destPoolClusterId, null, null, new HashSet<StoragePool>());
         String secondaryStorageURL = getSecondaryStorageURL(volume.getDataCenterId());
 
         if (destPool == null) {
@@ -1894,6 +1892,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
 
         Long zoneId = cmd.getZoneId();
         Long diskOfferingId = null;
+        DiskOfferingVO diskOffering = null;
         Long size = null;
 
         // validate input parameters before creating the volume
@@ -1913,11 +1912,9 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
                     throw new InvalidParameterValueException("Disk size must be larger than 0");
                 }
             }
-            if (diskOfferingId == null) {
+
-                throw new InvalidParameterValueException("Missing parameter(s),either a positive volume size or a valid disk offering id must be specified.");
-            }
             // Check that the the disk offering is specified
-            DiskOfferingVO diskOffering = _diskOfferingDao.findById(diskOfferingId);
+            diskOffering = _diskOfferingDao.findById(diskOfferingId);
             if ((diskOffering == null) || diskOffering.getRemoved() != null || !DiskOfferingVO.Type.Disk.equals(diskOffering.getType())) {
                 throw new InvalidParameterValueException("Please specify a valid disk offering.");
             }
@@ -1959,7 +1956,8 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
                 throw new InvalidParameterValueException("Snapshot id=" + snapshotId + " is not in " + Snapshot.Status.BackedUp + " state yet and can't be used for volume creation");
             }
 
-            diskOfferingId = (cmd.getDiskOfferingId() != null) ? cmd.getDiskOfferingId() : snapshotCheck.getDiskOfferingId();
+            diskOfferingId = snapshotCheck.getDiskOfferingId();
+            diskOffering = _diskOfferingDao.findById(diskOfferingId);
             zoneId = snapshotCheck.getDataCenterId();
             size = snapshotCheck.getSize(); // ; disk offering is used for tags purposes
 
@@ -1988,12 +1986,27 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
             throw new PermissionDeniedException("Cannot perform this operation, Zone is currently disabled: " + zoneId);
         }
 
-        // Check that there is a shared primary storage pool in the specified zone
+        // If local storage is disabled then creation of volume with local disk offering not allowed
+        if (!zone.isLocalStorageEnabled() && diskOffering.getUseLocalStorage()) {
+            throw new InvalidParameterValueException("Zone is not configured to use local storage but volume's disk offering " + diskOffering.getName() + " uses it");
+        }
+
+        // Check that there is appropriate primary storage pool in the specified zone
         List<StoragePoolVO> storagePools = _storagePoolDao.listByDataCenterId(zoneId);
-        boolean sharedPoolExists = false;
+        boolean appropriatePoolExists = false;
+        if (!diskOffering.getUseLocalStorage()) {
             for (StoragePoolVO storagePool : storagePools) {
                 if (storagePool.isShared()) {
-                sharedPoolExists = true;
+                    appropriatePoolExists = true;
+                    break;
+                }
+            }
+        } else {
+            for (StoragePoolVO storagePool : storagePools) {
+                if (storagePool.isLocal()) {
+                    appropriatePoolExists = true;
+                    break;
+                }
             }
         }
 
@@ -2003,8 +2016,9 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
             throw new InvalidParameterValueException("There is no workable host in data center id " + zoneId + ", please check hosts' agent status and see if they are disabled");
         }
 
-        if (!sharedPoolExists) {
+        if (!appropriatePoolExists) {
-            throw new InvalidParameterValueException("Please specify a zone that has at least one shared primary storage pool.");
+            String storageType = diskOffering.getUseLocalStorage() ? ServiceOffering.StorageType.local.toString() : ServiceOffering.StorageType.shared.toString();
+            throw new InvalidParameterValueException("Volume's disk offering uses " + storageType + " storage, please specify a zone that has at least one " + storageType + " primary storage pool.");
         }
 
         String userSpecifiedName = cmd.getVolumeName();
@@ -3047,7 +3061,11 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
 
         StoragePool destPool = _storagePoolDao.findById(storagePoolId);
         if (destPool == null) {
-            throw new InvalidParameterValueException("Faild to find the destination storage pool: " + storagePoolId);
+            throw new InvalidParameterValueException("Failed to find the destination storage pool: " + storagePoolId);
+        }
+
+        if (!volumeOnSharedStoragePool(vol)) {
+            throw new InvalidParameterValueException("Migration of volume from local storage pool is not supported");
         }
 
         List<Volume> vols = new ArrayList<Volume>();

server/src/com/cloud/storage/allocator/AbstractStoragePoolAllocator.java

@@ -193,14 +193,14 @@ public abstract class AbstractStoragePoolAllocator extends AdapterBase implement
 	
 	
 	@Override
-	public List<StoragePool> allocateToPool(DiskProfile dskCh, VirtualMachineProfile<? extends VirtualMachine> vmProfile, long dcId, long podId, Long clusterId, Set<? extends StoragePool> avoids, int returnUpTo) {
+	public List<StoragePool> allocateToPool(DiskProfile dskCh, VirtualMachineProfile<? extends VirtualMachine> vmProfile, long dcId, long podId, Long clusterId, Long hostId, Set<? extends StoragePool> avoids, int returnUpTo) {
 	    
 	    ExcludeList avoid = new ExcludeList();
 	    for(StoragePool pool : avoids){
 	    	avoid.addPool(pool.getId());
 	    }
 	    
-	    DataCenterDeployment plan = new DataCenterDeployment(dcId, podId, clusterId, null, null, null);
+	    DataCenterDeployment plan = new DataCenterDeployment(dcId, podId, clusterId, hostId, null, null);
 	    return allocateToPool(dskCh, vmProfile, plan, avoid, returnUpTo);
 	}
 

server/src/com/cloud/storage/allocator/FirstFitStoragePoolAllocator.java

@@ -30,6 +30,7 @@ import org.apache.log4j.Logger;
 
 import com.cloud.deploy.DeploymentPlan;
 import com.cloud.deploy.DeploymentPlanner.ExcludeList;
+import com.cloud.offering.ServiceOffering;
 import com.cloud.server.StatsCollector;
 import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.dao.DiskOfferingDao;
@@ -85,7 +86,8 @@ public class FirstFitStoragePoolAllocator extends AbstractStoragePoolAllocator {
         List<StoragePoolVO> pools = _storagePoolDao.findPoolsByTags(dcId, podId, clusterId, dskCh.getTags(), null);
         if (pools.size() == 0) {
             if (s_logger.isDebugEnabled()) {
-    			s_logger.debug("No storage pools available for allocation, returning");
+                String storageType = dskCh.useLocalStorage() ? ServiceOffering.StorageType.local.toString() : ServiceOffering.StorageType.shared.toString();
+                s_logger.debug("No storage pools available for " + storageType + " volume allocation, returning");
             }
             return suitablePools;
         }

server/src/com/cloud/storage/allocator/LocalStoragePoolAllocator.java

@@ -35,6 +35,8 @@ import com.cloud.offering.ServiceOffering;
 import com.cloud.service.dao.ServiceOfferingDao;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.StoragePoolHostVO;
+import com.cloud.storage.StoragePoolVO;
+import com.cloud.storage.Volume;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.StoragePoolHostDao;
 import com.cloud.utils.DateUtil;
@@ -101,6 +103,21 @@ public class LocalStoragePoolAllocator extends FirstFitStoragePoolAllocator {
             s_logger.debug("LocalStoragePoolAllocator trying to find storage pool to fit the vm");
         }
 
+        // data disk and host identified from deploying vm (attach volume case)
+        if (dskCh.getType() == Volume.Type.DATADISK && plan.getHostId() != null) {
+            List<StoragePoolHostVO> hostPools = _poolHostDao.listByHostId(plan.getHostId());
+            for (StoragePoolHostVO hostPool: hostPools) {
+                StoragePoolVO pool = _storagePoolDao.findById(hostPool.getPoolId());
+                if (pool != null && pool.isLocal()) {
+                    s_logger.debug("Found suitable local storage pool " + pool.getId() + ", adding to list");
+                    suitablePools.add(pool);
+                }
+
+                if (suitablePools.size() == returnUpTo) {
+                    break;
+                }
+            }
+        } else {
             List<StoragePool> availablePool;
             while (!(availablePool = super.allocateToPool(dskCh, vmProfile, plan, myAvoids, 1)).isEmpty()) {
                 StoragePool pool = availablePool.get(0);
@@ -115,6 +132,7 @@ public class LocalStoragePoolAllocator extends FirstFitStoragePoolAllocator {
                     break;
                 }
             }
+        }
 
         if (s_logger.isDebugEnabled()) {
             s_logger.debug("LocalStoragePoolAllocator returning " + suitablePools.size() + " suitable storage pools");

server/src/com/cloud/storage/allocator/StoragePoolAllocator.java

@@ -33,7 +33,7 @@ import com.cloud.vm.VirtualMachineProfile;
 public interface StoragePoolAllocator extends Adapter {
 	
 	//keeping since storageMgr is using this API for some existing functionalities
-	List<StoragePool> allocateToPool(DiskProfile dskCh, VirtualMachineProfile<? extends VirtualMachine> vmProfile, long dcId, long podId, Long clusterId, Set<? extends StoragePool> avoids, int returnUpTo);	
+	List<StoragePool> allocateToPool(DiskProfile dskCh, VirtualMachineProfile<? extends VirtualMachine> vmProfile, long dcId, long podId, Long clusterId, Long hostId, Set<? extends StoragePool> avoids, int returnUpTo);
 	
 	String chooseStorageIp(VirtualMachine vm, Host host, Host storage);
 

server/src/com/cloud/storage/allocator/UseLocalForRootAllocator.java

@@ -24,23 +24,29 @@ import javax.naming.ConfigurationException;
 
 import com.cloud.configuration.Config;
 import com.cloud.configuration.dao.ConfigurationDao;
+import com.cloud.dc.DataCenterVO;
+import com.cloud.dc.dao.DataCenterDao;
 import com.cloud.deploy.DeploymentPlan;
 import com.cloud.deploy.DeploymentPlanner.ExcludeList;
 import com.cloud.host.Host;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.Volume.Type;
 import com.cloud.utils.component.ComponentLocator;
+import com.cloud.utils.component.Inject;
 import com.cloud.vm.DiskProfile;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.VirtualMachineProfile;
 
 @Local(value=StoragePoolAllocator.class)
 public class UseLocalForRootAllocator extends LocalStoragePoolAllocator implements StoragePoolAllocator {
-    boolean _useLocalStorage;
+
+    @Inject
+    DataCenterDao _dcDao;
 
     @Override
     public List<StoragePool> allocateToPool(DiskProfile dskCh, VirtualMachineProfile<? extends VirtualMachine> vmProfile, DeploymentPlan plan, ExcludeList avoid, int returnUpTo) {
-        if (!_useLocalStorage) {
+        DataCenterVO dc = _dcDao.findById(plan.getDataCenterId());
+        if (!dc.isLocalStorageEnabled()) {
             return null;
         }
         
@@ -55,13 +61,6 @@ public class UseLocalForRootAllocator extends LocalStoragePoolAllocator implemen
     @Override
     public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
         super.configure(name, params);
-        
-        ComponentLocator locator = ComponentLocator.getCurrentLocator();
-        ConfigurationDao configDao = locator.getDao(ConfigurationDao.class);
-        Map<String, String> dbParams = configDao.getConfiguration(params);
-        
-        _useLocalStorage = Boolean.parseBoolean(dbParams.get(Config.UseLocalStorage.toString()));
-        
         return true;
     }
     

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

@@ -205,8 +205,7 @@ public class DatabaseConfig {
         s_configurationDescriptions.put("expunge.interval", "the interval to wait before running the expunge thread");
         s_configurationDescriptions.put("network.throttling.rate", "default data transfer rate in megabits per second allowed per user");
         s_configurationDescriptions.put("multicast.throttling.rate", "default multicast rate in megabits per second allowed");
-        s_configurationDescriptions.put("use.local.storage", "Indicates whether to use local storage pools or shared storage pools for user VMs");
+        s_configurationDescriptions.put("system.vm.use.local.storage", "Indicates whether to use local storage pools or shared storage pools for system VMs.");
-        s_configurationDescriptions.put("use.local.storage", "Indicates whether to use local storage pools or shared storage pools for system VMs.");
         s_configurationDescriptions.put("snapshot.poll.interval", "The time interval in seconds when the management server polls for snapshots to be scheduled.");
         s_configurationDescriptions.put("snapshot.max.hourly", "Maximum hourly snapshots for a volume");
         s_configurationDescriptions.put("snapshot.max.daily", "Maximum daily snapshots for a volume");
@@ -280,7 +279,6 @@ public class DatabaseConfig {
         s_configurationComponents.put("storage.overwrite.provisioning", "UserVmManager");
         s_configurationComponents.put("init", "none");
         s_configurationComponents.put("system.vm.use.local.storage", "ManagementServer");
-        s_configurationComponents.put("use.local.storage", "ManagementServer");
         s_configurationComponents.put("snapshot.poll.interval", "SnapshotManager");
         s_configurationComponents.put("snapshot.max.hourly", "SnapshotManager");
         s_configurationComponents.put("snapshot.max.daily", "SnapshotManager");
@@ -334,7 +332,6 @@ public class DatabaseConfig {
         s_defaultConfigurationValues.put("event.purge.interval", "86400");
         s_defaultConfigurationValues.put("account.cleanup.interval", "86400");
         s_defaultConfigurationValues.put("system.vm.use.local.storage", "false");
-        s_defaultConfigurationValues.put("use.local.storage", "false");
         s_defaultConfigurationValues.put("init", "false");
         s_defaultConfigurationValues.put("cpu.overprovisioning.factor", "1");
         s_defaultConfigurationValues.put("mem.overprovisioning.factor", "1");

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

@@ -588,6 +588,15 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             throw new InvalidParameterValueException("Please specify a VM that is in the same zone as the volume.");
         }
 
+        // If local storage is disabled then attaching a volume with local disk offering not allowed
+        DataCenterVO dataCenter = _dcDao.findById(volume.getDataCenterId());
+        if (!dataCenter.isLocalStorageEnabled()) {
+            DiskOfferingVO diskOffering = _diskOfferingDao.findById(volume.getDiskOfferingId());
+            if (diskOffering.getUseLocalStorage()) {
+                throw new InvalidParameterValueException("Zone is not configured to use local storage but volume's disk offering " + diskOffering.getName() + " uses it");
+            }
+        }
+
         //permission check
         _accountMgr.checkAccess(caller, null, true, volume, vm);
 
@@ -601,11 +610,6 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             }
         }
 
-        //If the volume is Ready, check that the volume is stored on shared storage
-        if (!(Volume.State.Allocated.equals(volume.getState()) || Volume.State.UploadOp.equals(volume.getState())) && !_storageMgr.volumeOnSharedStoragePool(volume)) {
-            throw new InvalidParameterValueException("Please specify a volume that has been created on a shared storage pool.");
-        }
-
         if ( !(Volume.State.Allocated.equals(volume.getState()) || Volume.State.Ready.equals(volume.getState()) || Volume.State.UploadOp.equals(volume.getState())) ) {
             throw new InvalidParameterValueException("Volume state must be in Allocated, Ready or in Uploaded state");
         }
@@ -700,10 +704,11 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             DiskOfferingVO volumeDiskOffering = _diskOfferingDao.findById(volume.getDiskOfferingId());
             String[] volumeTags = volumeDiskOffering.getTagsArray();
 
+            boolean isVolumeOnSharedPool = !volumeDiskOffering.getUseLocalStorage();
             StoragePoolVO sourcePool = _storagePoolDao.findById(volume.getPoolId());
-            List<StoragePoolVO> sharedVMPools = _storagePoolDao.findPoolsByTags(vmRootVolumePool.getDataCenterId(), vmRootVolumePool.getPodId(), vmRootVolumePool.getClusterId(), volumeTags, true);
+            List<StoragePoolVO> matchingVMPools = _storagePoolDao.findPoolsByTags(vmRootVolumePool.getDataCenterId(), vmRootVolumePool.getPodId(), vmRootVolumePool.getClusterId(), volumeTags, isVolumeOnSharedPool);
             boolean moveVolumeNeeded = true;
-            if (sharedVMPools.size() == 0) {
+            if (matchingVMPools.size() == 0) {
                 String poolType;
                 if (vmRootVolumePool.getClusterId() != null) {
                     poolType = "cluster";
@@ -714,15 +719,20 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
                 }
                 throw new CloudRuntimeException("There are no storage pools in the VM's " + poolType + " with all of the volume's tags (" + volumeDiskOffering.getTags() + ").");
             } else {
+                long sourcePoolId = sourcePool.getId();
                 Long sourcePoolDcId = sourcePool.getDataCenterId();
                 Long sourcePoolPodId = sourcePool.getPodId();
                 Long sourcePoolClusterId = sourcePool.getClusterId();
-                for (StoragePoolVO vmPool : sharedVMPools) {
+                for (StoragePoolVO vmPool : matchingVMPools) {
+                    long vmPoolId = vmPool.getId();
                     Long vmPoolDcId = vmPool.getDataCenterId();
                     Long vmPoolPodId = vmPool.getPodId();
                     Long vmPoolClusterId = vmPool.getClusterId();
 
-                    if (sourcePoolDcId == vmPoolDcId && sourcePoolPodId == vmPoolPodId && sourcePoolClusterId == vmPoolClusterId) {
+                    // Moving a volume is not required if storage pools belongs to same cluster in case of shared volume or
+                    // identical storage pool in case of local
+                    if (sourcePoolDcId == vmPoolDcId && sourcePoolPodId == vmPoolPodId && sourcePoolClusterId == vmPoolClusterId
+                            && (isVolumeOnSharedPool || sourcePoolId == vmPoolId)) {
                         moveVolumeNeeded = false;
                         break;
                     }
@@ -730,12 +740,16 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             }
 
             if (moveVolumeNeeded) {
+                if (isVolumeOnSharedPool) {
                     // Move the volume to a storage pool in the VM's zone, pod, or cluster
                     try {
                         volume = _storageMgr.moveVolume(volume, vmRootVolumePool.getDataCenterId(), vmRootVolumePool.getPodId(), vmRootVolumePool.getClusterId(), dataDiskHyperType);
                     } catch (ConcurrentOperationException e) {
                         throw new CloudRuntimeException(e.toString());
                     }
+                } else {
+                    throw new CloudRuntimeException("Failed to attach local data volume " + volume.getName() + " to VM " + vm.getDisplayName() + " as migration of local data volume is not allowed");
+                }
             }
         }
 
@@ -838,11 +852,6 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             throw new InvalidParameterValueException("The specified volume is not attached to a VM.");
         }
 
-        // Check that the volume is stored on shared storage
-        if (volume.getState() != Volume.State.Allocated && !_storageMgr.volumeOnSharedStoragePool(volume)) {
-            throw new InvalidParameterValueException("Please specify a volume that has been created on a shared storage pool.");
-        } 
-
         // Check that the VM is in the correct state
         UserVmVO vm = _vmDao.findById(vmId);
         if (vm.getState() != State.Running && vm.getState() != State.Stopped && vm.getState() != State.Destroyed) {
@@ -3276,6 +3285,25 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
 
     }
 
+    private boolean isVMUsingLocalStorage(VMInstanceVO vm)
+    {
+        boolean usesLocalStorage = false;
+        ServiceOfferingVO svcOffering = _serviceOfferingDao.findById(vm.getServiceOfferingId());
+        if (svcOffering.getUseLocalStorage()) {
+            usesLocalStorage = true;
+        } else {
+            List<VolumeVO> volumes = _volsDao.findByInstanceAndType(vm.getId(), Volume.Type.DATADISK);
+            for (VolumeVO vol : volumes) {
+                DiskOfferingVO diskOffering = _diskOfferingDao.findById(vol.getDiskOfferingId());
+                if (diskOffering.getUseLocalStorage()) {
+                    usesLocalStorage = true;
+                    break;
+                }
+            }
+        }
+        return usesLocalStorage;
+    }
+
     @Override
     @ActionEvent(eventType = EventTypes.EVENT_VM_MIGRATE, eventDescription = "migrating VM", async = true)
     public VirtualMachine migrateVirtualMachine(Long vmId, Host destinationHost) throws ResourceUnavailableException, ConcurrentOperationException, ManagementServerException, VirtualMachineMigrationException {
@@ -3308,8 +3336,7 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             throw new InvalidParameterValueException("Unsupported Hypervisor Type for VM migration, we support XenServer/VMware/KVM only");
         }
 
-        ServiceOfferingVO svcOffering = _serviceOfferingDao.findById(vm.getServiceOfferingId());
+        if (isVMUsingLocalStorage(vm)) {
-        if (svcOffering.getUseLocalStorage()) {
             if (s_logger.isDebugEnabled()) {
                 s_logger.debug(vm + " is using Local Storage, cannot migrate this VM.");
             }

setup/db/create-schema.sql

@@ -547,6 +547,7 @@ CREATE TABLE  `cloud`.`data_center` (
   `allocation_state` varchar(32) NOT NULL DEFAULT 'Enabled' COMMENT 'Is this data center enabled for allocation for new resources',
   `zone_token` varchar(255),
   `is_security_group_enabled` tinyint NOT NULL DEFAULT 0 COMMENT '1: enabled, 0: not',
+  `is_local_storage_enabled` tinyint NOT NULL DEFAULT 0 COMMENT 'Is local storage offering enabled for this data center; 1: enabled, 0: not',
   `removed` datetime COMMENT 'date removed if not null',
   PRIMARY KEY  (`id`),
   CONSTRAINT `fk_data_center__domain_id` FOREIGN KEY (`domain_id`) REFERENCES `domain`(`id`),

setup/db/db/schema-302to40.sql

@@ -246,6 +246,10 @@ DEALLOCATE PREPARE stmt1;
 AlTER TABLE physical_network_service_providers ADD CONSTRAINT `fk_pnetwork_service_providers__physical_network_id` FOREIGN KEY (`physical_network_id`) REFERENCES `physical_network`(`id`) ON DELETE CASCADE;
 UPDATE `cloud`.`configuration` SET description='In second, timeout for creating volume from snapshot' WHERE name='create.volume.from.snapshot.wait';
 
+ALTER TABLE `cloud`.`data_center` ADD COLUMN `is_local_storage_enabled` tinyint NOT NULL DEFAULT 0 COMMENT 'Is local storage offering enabled for this data center; 1: enabled, 0: not';
+UPDATE `cloud`.`data_center` SET `is_local_storage_enabled` = IF ((SELECT `value` FROM `cloud`.`configuration` WHERE `name`='use.local.storage')='true', 1, 0) WHERE `removed` IS NULL;
+DELETE FROM `cloud`.`configuration` where name='use.local.storage';
+
 ALTER TABLE `cloud`.`hypervisor_capabilities` ADD COLUMN `max_data_volumes_limit` int unsigned DEFAULT 6 COMMENT 'Max. data volumes per VM supported by hypervisor';
 UPDATE `cloud`.`hypervisor_capabilities` SET `max_data_volumes_limit`=13 WHERE `hypervisor_type`='XenServer' AND (`hypervisor_version`='6.0' OR `hypervisor_version`='6.0.2');
 INSERT INTO `cloud`.`configuration` (`category`, `instance`, `component`, `name`, `value`, `description`) VALUES ('Advanced', 'DEFAULT', 'management-server', 'event.purge.interval', '86400', 'The interval (in seconds) to wait before running the event purge thread');

ui/scripts/configuration.js

@@ -774,6 +774,15 @@
                     label: 'label.description',
                     validation: { required: true }
                   },
+                  storageType: {
+                    label: 'label.storage.type',
+                    select: function(args) {
+                      var items = [];
+                      items.push({id: 'shared', description: 'shared'});
+                      items.push({id: 'local', description: 'local'});
+                      args.response.success({data: items});
+                    }
+                  },
                   isCustomized: {
                     label: 'label.custom.disk.size',
                     isBoolean: true,
@@ -822,6 +831,7 @@
                 array1.push("&name=" + args.data.name);
                 array1.push("&displaytext=" + todb(args.data.description));
 
+                array1.push("&storageType=" + todb(args.data.storageType));
                 array1.push("&customized=" + (args.data.isCustomized=="on"));
                 if(args.$form.find('.form-item[rel=disksize]').css("display") != "none")
                   array1.push("&disksize=" + args.data.disksize);
@@ -941,7 +951,8 @@
                       }
                     },
                     tags: { label: 'label.storage.tags' },
-                    domain: { label: 'label.domain' }
+                    domain: { label: 'label.domain' },
+                    storagetype: { label: 'label.storage.type' }
                   }
                 ],
 

ui/scripts/storage.js

@@ -1327,21 +1327,19 @@
       }
       else { //jsonObj.type == "DATADISK"
         if (jsonObj.virtualmachineid != null) {
-          if (jsonObj.storagetype == "shared" && (jsonObj.vmstate == "Running" || jsonObj.vmstate == "Stopped" || jsonObj.vmstate == "Destroyed")) {
+          if (jsonObj.vmstate == "Running" || jsonObj.vmstate == "Stopped" || jsonObj.vmstate == "Destroyed") {
             allowedActions.push("detachDisk");
           }
         }
         else { // Disk not attached
           allowedActions.push("remove");
-					if(jsonObj.state == "Ready" && isAdmin()) {
+          if(jsonObj.state == "Ready" && isAdmin() && jsonObj.storagetype == "shared") {
             allowedActions.push("migrateToAnotherStorage");
           }
-          if (jsonObj.storagetype == "shared") {
           allowedActions.push("attachDisk");
         }
       }
     }
-    }
     return allowedActions;
   };
 

ui/scripts/system.js

@@ -3973,6 +3973,7 @@
                       array1.push("&internaldns1=" + todb(args.data.internaldns1));
                       array1.push("&internaldns2=" + todb(args.data.internaldns2));  //internaldns2 can be empty ("") when passed to API
                       array1.push("&domain=" + todb(args.data.domain));
+                      array1.push("&localstorageenabled=" + todb(args.data.localstorageenabled));
                       $.ajax({
                         url: createURL("updateZone&id=" + args.context.physicalResources[0].id + array1.join("")),
                         dataType: "json",
@@ -4016,6 +4017,16 @@
                         domain: {
                           label: 'label.network.domain',
                           isEditable: true
+                        },
+                        localstorageenabled: {
+                          label: 'label.local.storage.enabled',
+                          converter: function(args) {
+                            if(args)
+                              return "true";
+                            else
+                              return "false";
+                          },
+                          isEditable: true
                         }
                       }
                     ],