CLOUDSTACK-4757. During template registration, after template download analyze OVA template to identify additional disks and create Datadisk templates for each of the additional disks.

api/src/com/cloud/agent/api/to/DatadiskTemplateTO.java

@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package com.cloud.agent.api.to;
+
+public class DatadiskTemplateTO {
+    private long id;
+    private String uniqueName;
+    private String path;
+    private Long virtualSize;
+    private Long fileSize;
+
+    public DatadiskTemplateTO() {
+    }
+
+    public DatadiskTemplateTO(long id, String uniqueName, String path, Long virtualSize, Long fileSize) {
+        this.id = id;
+        this.uniqueName = uniqueName;
+        this.path = path;
+        this.virtualSize = virtualSize;
+        this.fileSize = fileSize;
+    }
+
+    public long getId() {
+        return id;
+    }
+
+    public void setId(long id) {
+        this.id = id;
+    }
+
+    public String getUniqueName() {
+        return uniqueName;
+    }
+
+    public void setUniqueName(String uniqueName) {
+        this.uniqueName = uniqueName;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public void setPath(String path) {
+        this.path = path;
+    }
+
+    public Long getVirtualSize() {
+        return virtualSize;
+    }
+
+    public void setVirtualSize(Long virtualSize) {
+        this.virtualSize = virtualSize;
+    }
+
+    public Long getFileSize() {
+        return fileSize;
+    }
+}

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

@@ -112,7 +112,8 @@ public class Storage {
         SYSTEM, /* routing, system vm template */
         BUILTIN, /* buildin template */
         PERHOST, /* every host has this template, don't need to install it in secondary storage */
-        USER /* User supplied template/iso */
+        USER, /* User supplied template/iso */
+        DATADISK /* Template corresponding to a datadisk(non root disk) present in an OVA */
     }
 
     public static enum StoragePoolType {

api/src/com/cloud/template/VirtualMachineTemplate.java

@@ -100,4 +100,6 @@ public interface VirtualMachineTemplate extends ControlledEntity, Identity, Inte
     Map getDetails();
 
     boolean isDynamicallyScalable();
+
+    Long getParentTemplateId();
 }

core/src/com/cloud/agent/api/storage/CreateDatadiskTemplateAnswer.java

@@ -0,0 +1,38 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.agent.api.storage;
+
+import org.apache.cloudstack.storage.to.TemplateObjectTO;
+
+import com.cloud.agent.api.Answer;
+
+public class CreateDatadiskTemplateAnswer extends Answer {
+    private TemplateObjectTO dataDiskTemplate = null;
+
+    public CreateDatadiskTemplateAnswer(TemplateObjectTO dataDiskTemplate) {
+        super(null);
+        this.dataDiskTemplate = dataDiskTemplate;
+    }
+
+    public TemplateObjectTO getDataDiskTemplate() {
+        return dataDiskTemplate;
+    }
+
+    public CreateDatadiskTemplateAnswer(String errMsg) {
+        super(null, false, errMsg);
+    }
+}

core/src/com/cloud/agent/api/storage/CreateDatadiskTemplateCommand.java

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.agent.api.storage;
+
+import com.cloud.agent.api.Command;
+import com.cloud.agent.api.to.DataTO;
+
+public final class CreateDatadiskTemplateCommand extends Command {
+    private DataTO dataDiskTemplate;
+    private String path;
+    private long fileSize;
+
+    public CreateDatadiskTemplateCommand(DataTO dataDiskTemplate, String path, long fileSize) {
+        super();
+        this.dataDiskTemplate = dataDiskTemplate;
+        this.path = path;
+        this.fileSize = fileSize;
+    }
+
+    protected CreateDatadiskTemplateCommand() {
+        super();
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return false;
+    }
+
+    public DataTO getDataDiskTemplate() {
+        return dataDiskTemplate;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public long getFileSize() {
+        return fileSize;
+    }
+}

core/src/com/cloud/agent/api/storage/GetDatadisksAnswer.java

@@ -0,0 +1,40 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.agent.api.storage;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import com.cloud.agent.api.Answer;
+import com.cloud.utils.Ternary;
+
+public class GetDatadisksAnswer extends Answer {
+    List<Ternary<String, Long, Long>> dataDiskDetails = new ArrayList<Ternary<String, Long, Long>>();
+
+    public GetDatadisksAnswer(List<Ternary<String, Long, Long>> dataDiskDetails) {
+        super(null);
+        this.dataDiskDetails = dataDiskDetails;
+    }
+
+    public List<Ternary<String, Long, Long>> getDataDiskDetails() {
+        return dataDiskDetails;
+    }
+
+    public GetDatadisksAnswer(String errMsg) {
+        super(null, false, errMsg);
+    }
+}

core/src/com/cloud/agent/api/storage/GetDatadisksCommand.java

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package com.cloud.agent.api.storage;
+
+import com.cloud.agent.api.Command;
+import com.cloud.agent.api.to.DataTO;
+
+public final class GetDatadisksCommand extends Command {
+    private DataTO data;
+
+    public GetDatadisksCommand(DataTO data) {
+        super();
+        this.data = data;
+    }
+
+    protected GetDatadisksCommand() {
+        super();
+    }
+
+    @Override
+    public boolean executeInSequence() {
+        return false;
+    }
+
+    public DataTO getData() {
+        return data;
+    }
+
+}

core/src/com/cloud/storage/template/OVAProcessor.java

@@ -26,10 +26,12 @@ import javax.xml.parsers.DocumentBuilderFactory;
 import org.apache.log4j.Logger;
 import org.w3c.dom.Document;
 import org.w3c.dom.Element;
+import org.w3c.dom.NodeList;
 
 import com.cloud.exception.InternalErrorException;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.StorageLayer;
+import com.cloud.utils.Pair;
 import com.cloud.utils.component.AdapterBase;
 import com.cloud.utils.script.Script;
 
@@ -131,6 +133,52 @@ public class OVAProcessor extends AdapterBase implements Processor {
         }
     }
 
+    public Pair<Long, Long> getDiskDetails(String ovfFilePath, String diskName) throws InternalErrorException {
+        long virtualSize = 0;
+        long fileSize = 0;
+        String fileId = null;
+        try {
+            Document ovfDoc = null;
+            ovfDoc = DocumentBuilderFactory.newInstance().newDocumentBuilder().parse(new File(ovfFilePath));
+            NodeList disks = ovfDoc.getElementsByTagName("Disk");
+            NodeList files = ovfDoc.getElementsByTagName("File");
+            for (int j = 0; j < files.getLength(); j++) {
+                Element file = (Element)files.item(j);
+                if (file.getAttribute("ovf:href").equals(diskName)) {
+                    fileSize = Long.parseLong(file.getAttribute("ovf:size"));
+                    fileId = file.getAttribute("ovf:id");
+                    break;
+                }
+            }
+            for (int i = 0; i < disks.getLength(); i++) {
+                Element disk = (Element)disks.item(i);
+                if (disk.getAttribute("ovf:fileRef").equals(fileId)) {
+                    virtualSize = Long.parseLong(disk.getAttribute("ovf:capacity"));
+                    String allocationUnits = disk.getAttribute("ovf:capacityAllocationUnits");
+                    if ((virtualSize != 0) && (allocationUnits != null)) {
+                        long units = 1;
+                        if (allocationUnits.equalsIgnoreCase("KB") || allocationUnits.equalsIgnoreCase("KiloBytes") || allocationUnits.equalsIgnoreCase("byte * 2^10")) {
+                            units = 1024;
+                        } else if (allocationUnits.equalsIgnoreCase("MB") || allocationUnits.equalsIgnoreCase("MegaBytes") || allocationUnits.equalsIgnoreCase("byte * 2^20")) {
+                            units = 1024 * 1024;
+                        } else if (allocationUnits.equalsIgnoreCase("GB") || allocationUnits.equalsIgnoreCase("GigaBytes") || allocationUnits.equalsIgnoreCase("byte * 2^30")) {
+                            units = 1024 * 1024 * 1024;
+                        }
+                        virtualSize = virtualSize * units;
+                    } else {
+                        throw new InternalErrorException("Failed to read capacity and capacityAllocationUnits from the OVF file: " + ovfFilePath);
+                    }
+                    break;
+                }
+            }
+            return new Pair<Long, Long>(virtualSize, fileSize);
+        } catch (Exception e) {
+            String msg = "Unable to parse OVF XML document to get the virtual disk size due to" + e;
+            s_logger.error(msg);
+            throw new InternalErrorException(msg);
+        }
+    }
+
     private String getOVFFilePath(String srcOVAFileName) {
         File file = new File(srcOVAFileName);
         assert (_storage != null);

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

@@ -41,6 +41,8 @@ public class TemplateObjectTO implements DataTO {
     private Long size;
     private Long physicalSize;
     private Hypervisor.HypervisorType hypervisorType;
+    private boolean bootable;
+    private String uniqueName;
 
     public TemplateObjectTO() {
 
@@ -70,6 +72,9 @@ public class TemplateObjectTO implements DataTO {
         this.accountId = template.getAccountId();
         this.name = template.getUniqueName();
         this.format = template.getFormat();
+        this.uniqueName = template.getUniqueName();
+        this.size = template.getSize();
+
         if (template.getDataStore() != null) {
             this.imageDataStore = template.getDataStore().getTO();
         }
@@ -204,10 +209,26 @@ public class TemplateObjectTO implements DataTO {
         return physicalSize;
     }
 
+    public void setIsBootable(boolean bootable) {
+        this.bootable = bootable;
+    }
+
+    public boolean isBootable() {
+        return bootable;
+    }
+
     public void setPhysicalSize(Long physicalSize) {
         this.physicalSize = physicalSize;
     }
 
+    public String getUniqueName() {
+        return this.uniqueName;
+    }
+
+    public void setUniqueName(String uniqueName) {
+        this.uniqueName = uniqueName;
+    }
+
     @Override
     public String toString() {
         return new StringBuilder("TemplateTO[id=").append(id).append("|origUrl=").append(origUrl).append("|name").append(name).append("]").toString();

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

@@ -20,6 +20,8 @@ package org.apache.cloudstack.engine.subsystem.api.storage;
 
 import java.util.List;
 
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
+
 public interface EndPointSelector {
     EndPoint select(DataObject srcData, DataObject destData);
 
@@ -35,5 +37,5 @@ public interface EndPointSelector {
 
     EndPoint select(Scope scope, Long storeId);
 
-    EndPoint selectHypervisorHost(Scope scope);
+    EndPoint selectHypervisorHostByType(Scope scope, HypervisorType htype);
 }

engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/TemplateService.java

@@ -18,12 +18,15 @@
  */
 package org.apache.cloudstack.engine.subsystem.api.storage;
 
+import java.util.List;
+
 import org.apache.cloudstack.framework.async.AsyncCallFuture;
 import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 import org.apache.cloudstack.storage.command.CommandResult;
 
 import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.storage.StoragePool;
+import com.cloud.utils.Ternary;
 
 public interface TemplateService {
 
@@ -65,4 +68,8 @@ public interface TemplateService {
     void associateTemplateToZone(long templateId, Long zoneId);
 
     void associateCrosszoneTemplatesToZone(long dcId);
+
+    List<Ternary<String, Long, Long>> getDatadiskTemplates(TemplateInfo template);
+
+    AsyncCallFuture<TemplateApiResult> createDatadiskTemplateAsync(TemplateInfo parentTemplate, TemplateInfo dataDiskTemplate, String path, long fileSize);
 }

engine/api/src/org/apache/cloudstack/storage/image/datastore/ImageStoreEntity.java

@@ -18,16 +18,20 @@
  */
 package org.apache.cloudstack.storage.image.datastore;
 
+import java.util.List;
 import java.util.Set;
 
+import org.apache.cloudstack.engine.subsystem.api.storage.CreateCmdResult;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
 import org.apache.cloudstack.engine.subsystem.api.storage.TemplateInfo;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
+import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 
 import com.cloud.storage.ImageStore;
 import com.cloud.storage.Storage.ImageFormat;
+import com.cloud.utils.Ternary;
 
 public interface ImageStoreEntity extends DataStore, ImageStore {
     TemplateInfo getTemplate(long templateId);
@@ -43,4 +47,8 @@ public interface ImageStoreEntity extends DataStore, ImageStore {
     String getMountPoint(); // get the mount point on ssvm.
 
     String createEntityExtractUrl(String installPath, ImageFormat format, DataObject dataObject);  // get the entity download URL
+
+    List<Ternary<String, Long, Long>> getDatadiskTemplates(DataObject obj);
+
+    Void createDataDiskTemplateAsync(TemplateInfo dataDiskTemplate, String path, long fileSize, AsyncCompletionCallback<CreateCmdResult> callback);
 }

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

@@ -146,6 +146,9 @@ public class VMTemplateVO implements VirtualMachineTemplate {
     @Column(name = "dynamically_scalable")
     protected boolean dynamicallyScalable;
 
+    @Column(name = "parent_template_id")
+    private Long parentTemplateId;
+
     @Override
     public String getUniqueName() {
         return uniqueName;
@@ -636,4 +639,13 @@ public class VMTemplateVO implements VirtualMachineTemplate {
     public Class<?> getEntityType() {
         return VirtualMachineTemplate.class;
     }
+
+    @Override
+    public Long getParentTemplateId() {
+        return parentTemplateId;
+    }
+
+    public void setParentTemplateId(Long parentTemplateId) {
+        this.parentTemplateId = parentTemplateId;
+    }
 }

engine/storage/image/src/org/apache/cloudstack/storage/image/TemplateServiceImpl.java

@@ -91,6 +91,7 @@ import com.cloud.template.TemplateManager;
 import com.cloud.user.Account;
 import com.cloud.user.AccountManager;
 import com.cloud.user.ResourceLimitService;
+import com.cloud.utils.Ternary;
 import com.cloud.utils.UriUtils;
 import com.cloud.utils.db.GlobalLock;
 import com.cloud.utils.exception.CloudRuntimeException;
@@ -134,6 +135,8 @@ public class TemplateServiceImpl implements TemplateService {
     ConfigurationDao _configDao;
     @Inject
     StorageCacheManager _cacheMgr;
+    @Inject
+    TemplateDataFactory imageFactory;
 
     class TemplateOpContext<T> extends AsyncRpcContext<T> {
         final TemplateObject template;
@@ -887,4 +890,74 @@ public class TemplateServiceImpl implements TemplateService {
             }
         }
     }
+
+    @Override
+    public List<Ternary<String, Long, Long>> getDatadiskTemplates(TemplateInfo template) {
+        List<Ternary<String, Long, Long>> dataDiskDetails = new ArrayList<Ternary<String, Long, Long>>();
+        ImageStoreEntity tmpltStore = (ImageStoreEntity)template.getDataStore();
+        dataDiskDetails = tmpltStore.getDatadiskTemplates(template);
+        return dataDiskDetails;
+    }
+
+    private class CreateDataDiskTemplateContext<T> extends AsyncRpcContext<T> {
+        private final DataObject dataDiskTemplate;
+        private final AsyncCallFuture<TemplateApiResult> future;
+
+        public CreateDataDiskTemplateContext(AsyncCompletionCallback<T> callback, DataObject dataDiskTemplate, AsyncCallFuture<TemplateApiResult> future) {
+            super(callback);
+            this.dataDiskTemplate = dataDiskTemplate;
+            this.future = future;
+        }
+
+        public AsyncCallFuture<TemplateApiResult> getFuture() {
+            return this.future;
+        }
+    }
+
+    @Override
+    public AsyncCallFuture<TemplateApiResult> createDatadiskTemplateAsync(TemplateInfo parentTemplate, TemplateInfo dataDiskTemplate, String path, long fileSize) {
+        AsyncCallFuture<TemplateApiResult> future = new AsyncCallFuture<TemplateApiResult>();
+        // Make an entry for Datadisk template in template_store_ref table
+        DataStore store = parentTemplate.getDataStore();
+        TemplateObject dataDiskTemplateOnStore = (TemplateObject)store.create(dataDiskTemplate);
+        dataDiskTemplateOnStore.processEvent(ObjectInDataStoreStateMachine.Event.CreateOnlyRequested);
+        try {
+            CreateDataDiskTemplateContext<TemplateApiResult> context = new CreateDataDiskTemplateContext<TemplateApiResult>(null, dataDiskTemplateOnStore, future);
+            AsyncCallbackDispatcher<TemplateServiceImpl, CreateCmdResult> caller = AsyncCallbackDispatcher.create(this);
+            caller.setCallback(caller.getTarget().createDataDiskTemplateCallback(null, null)).setContext(context);
+            ImageStoreEntity tmpltStore = (ImageStoreEntity)parentTemplate.getDataStore();
+            tmpltStore.createDataDiskTemplateAsync(dataDiskTemplate, path, fileSize, caller);
+        } catch (CloudRuntimeException ex) {
+            dataDiskTemplateOnStore.processEvent(ObjectInDataStoreStateMachine.Event.OperationFailed);
+            TemplateApiResult result = new TemplateApiResult(dataDiskTemplate);
+            result.setResult(ex.getMessage());
+            if (future != null) {
+                future.complete(result);
+            }
+        }
+        return future;
+    }
+
+    protected Void createDataDiskTemplateCallback(AsyncCallbackDispatcher<TemplateServiceImpl, CreateCmdResult> callback, CreateDataDiskTemplateContext<TemplateApiResult> context) {
+        if (s_logger.isDebugEnabled()) {
+            s_logger.debug("Performing create datadisk template cross callback after completion");
+        }
+        DataObject dataDiskTemplate = context.dataDiskTemplate;
+        AsyncCallFuture<TemplateApiResult> future = context.getFuture();
+        CreateCmdResult result = callback.getResult();
+        TemplateApiResult dataDiskTemplateResult = new TemplateApiResult((TemplateObject)dataDiskTemplate);
+        try {
+            if (result.isSuccess()) {
+                dataDiskTemplate.processEvent(Event.OperationSuccessed, result.getAnswer());
+            } else {
+                dataDiskTemplate.processEvent(Event.OperationFailed);
+                dataDiskTemplateResult.setResult(result.getResult());
+            }
+        } catch (Exception e) {
+            s_logger.debug("Failed to process copy template cross zones callback", e);
+            dataDiskTemplateResult.setResult(e.toString());
+        }
+        future.complete(dataDiskTemplateResult);
+        return null;
+    }
 }

engine/storage/image/src/org/apache/cloudstack/storage/image/store/ImageStoreImpl.java

@@ -19,6 +19,7 @@
 package org.apache.cloudstack.storage.image.store;
 
 import java.util.Date;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 
@@ -26,6 +27,7 @@ import javax.inject.Inject;
 
 import org.apache.log4j.Logger;
 
+import org.apache.cloudstack.engine.subsystem.api.storage.CreateCmdResult;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreDriver;
 import org.apache.cloudstack.engine.subsystem.api.storage.ImageStoreProvider;
@@ -35,6 +37,7 @@ import org.apache.cloudstack.engine.subsystem.api.storage.TemplateInfo;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
 import org.apache.cloudstack.engine.subsystem.api.storage.ZoneScope;
 import org.apache.cloudstack.framework.async.AsyncCallFuture;
+import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 import org.apache.cloudstack.storage.command.CommandResult;
 import org.apache.cloudstack.storage.datastore.ObjectInDataStoreManager;
 import org.apache.cloudstack.storage.datastore.db.ImageStoreVO;
@@ -47,6 +50,7 @@ import com.cloud.capacity.dao.CapacityDao;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.dao.VMTemplateDao;
+import com.cloud.utils.Ternary;
 import com.cloud.utils.component.ComponentContext;
 
 public class ImageStoreImpl implements ImageStoreEntity {
@@ -203,4 +207,14 @@ public class ImageStoreImpl implements ImageStoreEntity {
         return driver.createEntityExtractUrl(this, installPath, format, dataObject);
     }
 
+    @Override
+    public List<Ternary<String, Long, Long>> getDatadiskTemplates(DataObject obj) {
+        return driver.getDatadiskTemplates(obj);
+    }
+
+    @Override
+    public Void createDataDiskTemplateAsync(TemplateInfo dataDiskTemplate, String path, long fileSize, AsyncCompletionCallback<CreateCmdResult> callback) {
+        return driver.createDataDiskTemplateAsync(dataDiskTemplate, path, fileSize, callback);
+
+    }
 }

engine/storage/image/src/org/apache/cloudstack/storage/image/store/TemplateObject.java

@@ -38,6 +38,7 @@ import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreVO;
 import org.apache.cloudstack.storage.to.TemplateObjectTO;
 
 import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateAnswer;
 import com.cloud.agent.api.to.DataObjectType;
 import com.cloud.agent.api.to.DataTO;
 import com.cloud.exception.ConcurrentOperationException;
@@ -220,6 +221,16 @@ public class TemplateObject implements TemplateInfo {
                         templateVO.setSize(newTemplate.getSize());
                         imageDao.update(templateVO.getId(), templateVO);
                     }
+                } else if (answer instanceof CreateDatadiskTemplateAnswer) {
+                    CreateDatadiskTemplateAnswer createAnswer = (CreateDatadiskTemplateAnswer)answer;
+                    TemplateObjectTO dataDiskTemplate = createAnswer.getDataDiskTemplate();
+                    TemplateDataStoreVO templateStoreRef = templateStoreDao.findByStoreTemplate(getDataStore().getId(), dataDiskTemplate.getId());
+                    templateStoreRef.setInstallPath(dataDiskTemplate.getPath());
+                    templateStoreRef.setDownloadPercent(100);
+                    templateStoreRef.setDownloadState(Status.DOWNLOADED);
+                    templateStoreRef.setSize(dataDiskTemplate.getSize());
+                    templateStoreRef.setPhysicalSize(dataDiskTemplate.getPhysicalSize());
+                    templateStoreDao.update(templateStoreRef.getId(), templateStoreRef);
                 }
             }
             objectInStoreMgr.update(this, event);

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

@@ -47,6 +47,7 @@ import com.cloud.host.HostVO;
 import com.cloud.host.Status;
 import com.cloud.host.dao.HostDao;
 import com.cloud.hypervisor.Hypervisor;
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.ScopeType;
 import com.cloud.storage.Storage.TemplateType;
@@ -66,6 +67,7 @@ public class DefaultEndPointSelector implements EndPointSelector {
         "select h.id from host h, storage_pool_host_ref s  where h.status = 'Up' and h.type = 'Routing' and h.resource_state = 'Enabled' and"
             + " h.id = s.host_id and s.pool_id = ? ";
     private String findOneHypervisorHostInScope = "select h.id from host h where h.status = 'Up' and h.hypervisor_type is not null ";
+    private String findOneHypervisorHostInScopeByType = "select h.id from host h where h.status = 'Up' and h.hypervisor_type = ? ";
 
     protected boolean moveBetweenPrimaryImage(DataStore srcStore, DataStore destStore) {
         DataStoreRole srcRole = srcStore.getRole();
@@ -346,9 +348,13 @@ public class DefaultEndPointSelector implements EndPointSelector {
     }
 
     @Override
-    public EndPoint selectHypervisorHost(Scope scope) {
+    public EndPoint selectHypervisorHostByType(Scope scope, HypervisorType htype) {
         StringBuilder sbuilder = new StringBuilder();
-        sbuilder.append(findOneHypervisorHostInScope);
+        if (htype != null) {
+            sbuilder.append(findOneHypervisorHostInScopeByType);
+        } else {
+            sbuilder.append(findOneHypervisorHostInScope);
+        }
         if (scope.getScopeType() == ScopeType.ZONE) {
             sbuilder.append(" and h.data_center_id = ");
             sbuilder.append(scope.getScopeId());
@@ -366,6 +372,9 @@ public class DefaultEndPointSelector implements EndPointSelector {
 
         try {
             pstmt = txn.prepareStatement(sql);
+            if (htype != null) {
+                pstmt.setString(1, htype.toString());
+            }
             rs = pstmt.executeQuery();
             while (rs.next()) {
                 long id = rs.getLong(1);

engine/storage/src/org/apache/cloudstack/storage/image/BaseImageStoreDriverImpl.java

@@ -20,7 +20,9 @@ package org.apache.cloudstack.storage.image;
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
 import java.util.Date;
+import java.util.List;
 import java.util.Map;
 
 import javax.inject.Inject;
@@ -33,6 +35,7 @@ import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.EndPoint;
 import org.apache.cloudstack.engine.subsystem.api.storage.EndPointSelector;
+import org.apache.cloudstack.engine.subsystem.api.storage.TemplateInfo;
 import org.apache.cloudstack.framework.async.AsyncCallbackDispatcher;
 import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 import org.apache.cloudstack.framework.async.AsyncRpcContext;
@@ -43,18 +46,27 @@ import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreVO;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
+import org.apache.cloudstack.storage.endpoint.DefaultEndPointSelector;
 
 import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateCommand;
 import com.cloud.agent.api.storage.DownloadAnswer;
+import com.cloud.agent.api.storage.GetDatadisksAnswer;
+import com.cloud.agent.api.storage.GetDatadisksCommand;
 import com.cloud.agent.api.storage.Proxy;
 import com.cloud.agent.api.to.DataObjectType;
 import com.cloud.agent.api.to.DataTO;
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.storage.VMTemplateStorageResourceAssoc;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.VolumeVO;
 import com.cloud.storage.dao.VMTemplateDao;
+import com.cloud.storage.dao.VMTemplateDetailsDao;
 import com.cloud.storage.dao.VolumeDao;
 import com.cloud.storage.download.DownloadMonitor;
+import com.cloud.user.ResourceLimitService;
+import com.cloud.user.dao.AccountDao;
+import com.cloud.utils.Ternary;
 
 public abstract class BaseImageStoreDriverImpl implements ImageStoreDriver {
     private static final Logger s_logger = Logger.getLogger(BaseImageStoreDriverImpl.class);
@@ -69,9 +81,17 @@ public abstract class BaseImageStoreDriverImpl implements ImageStoreDriver {
     @Inject
     TemplateDataStoreDao _templateStoreDao;
     @Inject
+    VMTemplateDetailsDao _templateDetailsDao;
+    @Inject
     EndPointSelector _epSelector;
     @Inject
-    ConfigurationDao configDao;
+    ConfigurationDao configDao;;
+    @Inject
+    DefaultEndPointSelector _defaultEpSelector;
+    @Inject
+    AccountDao _accountDao;
+    @Inject
+    ResourceLimitService _resourceLimitMgr;
     protected String _proxy = null;
 
     protected Proxy getHttpProxy() {
@@ -143,6 +163,7 @@ public abstract class BaseImageStoreDriverImpl implements ImageStoreDriver {
         DataObject obj = context.data;
         DataStore store = obj.getDataStore();
 
+        VMTemplateVO vmTemplate = _templateDao.findById(obj.getId());
         TemplateDataStoreVO tmpltStoreVO = _templateStoreDao.findByStoreTemplate(store.getId(), obj.getId());
         if (tmpltStoreVO != null) {
             if (tmpltStoreVO.getDownloadState() == VMTemplateStorageResourceAssoc.Status.DOWNLOADED) {
@@ -182,7 +203,6 @@ public abstract class BaseImageStoreDriverImpl implements ImageStoreDriver {
                 templateDaoBuilder.setChecksum(answer.getCheckSum());
                 _templateDao.update(obj.getId(), templateDaoBuilder);
             }
-
             CreateCmdResult result = new CreateCmdResult(null, null);
             caller.complete(result);
         }
@@ -271,4 +291,54 @@ public abstract class BaseImageStoreDriverImpl implements ImageStoreDriver {
     @Override
     public void resize(DataObject data, AsyncCompletionCallback<CreateCmdResult> callback) {
     }
+
+    @Override
+    public List<Ternary<String, Long, Long>> getDatadiskTemplates(DataObject obj) {
+        List<Ternary<String, Long, Long>> dataDiskDetails = new ArrayList<Ternary<String, Long, Long>>();
+        if (s_logger.isDebugEnabled()) {
+            s_logger.debug("Get the data disks present in the OVA template");
+        }
+        DataStore store = obj.getDataStore();
+        GetDatadisksCommand cmd = new GetDatadisksCommand(obj.getTO());
+        EndPoint ep = _defaultEpSelector.selectHypervisorHostByType(store.getScope(), HypervisorType.VMware);
+        Answer answer = null;
+        if (ep == null) {
+            String errMsg = "No remote endpoint to send command, check if host or ssvm is down?";
+            s_logger.error(errMsg);
+            answer = new Answer(cmd, false, errMsg);
+        } else {
+            answer = ep.sendMessage(cmd);
+        }
+        if (answer != null && answer.getResult()) {
+            GetDatadisksAnswer getDatadisksAnswer = (GetDatadisksAnswer)answer;
+            dataDiskDetails = getDatadisksAnswer.getDataDiskDetails(); // Details - Disk path, virtual size
+        }
+        return dataDiskDetails;
+    }
+
+    @Override
+    public Void createDataDiskTemplateAsync(TemplateInfo dataDiskTemplate, String path, long fileSize,
+            AsyncCompletionCallback<CreateCmdResult> callback) {
+        Answer answer = null;
+        String errMsg = null;
+        if (s_logger.isDebugEnabled()) {
+            s_logger.debug("Create Datadisk template: " + dataDiskTemplate.getId());
+        }
+        CreateDatadiskTemplateCommand cmd = new CreateDatadiskTemplateCommand(dataDiskTemplate.getTO(), path, fileSize);
+        EndPoint ep = _defaultEpSelector.selectHypervisorHostByType(dataDiskTemplate.getDataStore().getScope(), HypervisorType.VMware);
+        if (ep == null) {
+            errMsg = "No remote endpoint to send command, check if host or ssvm is down?";
+            s_logger.error(errMsg);
+            answer = new Answer(cmd, false, errMsg);
+        } else {
+            answer = ep.sendMessage(cmd);
+        }
+        if (answer != null && !answer.getResult()) {
+            errMsg = answer.getDetails();
+        }
+        CreateCmdResult result = new CreateCmdResult(null, answer);
+        result.setResult(errMsg);
+        callback.complete(result);
+        return null;
+    }
 }

engine/storage/src/org/apache/cloudstack/storage/image/ImageStoreDriver.java

@@ -18,12 +18,22 @@
  */
 package org.apache.cloudstack.storage.image;
 
+import java.util.List;
+
+import org.apache.cloudstack.engine.subsystem.api.storage.CreateCmdResult;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreDriver;
+import org.apache.cloudstack.engine.subsystem.api.storage.TemplateInfo;
+import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 
 import com.cloud.storage.Storage.ImageFormat;
+import com.cloud.utils.Ternary;
 
 public interface ImageStoreDriver extends DataStoreDriver {
     String createEntityExtractUrl(DataStore store, String installPath, ImageFormat format, DataObject dataObject);
+
+    List<Ternary<String, Long, Long>> getDatadiskTemplates(DataObject obj);
+
+    Void createDataDiskTemplateAsync(TemplateInfo dataDiskTemplate, String path, long fileSize, AsyncCompletionCallback<CreateCmdResult> callback);
 }

plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/manager/VmwareStorageManager.java

@@ -27,7 +27,9 @@ import com.cloud.agent.api.CreateVolumeFromSnapshotCommand;
 import com.cloud.agent.api.DeleteVMSnapshotCommand;
 import com.cloud.agent.api.RevertToVMSnapshotCommand;
 import com.cloud.agent.api.storage.CopyVolumeCommand;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateCommand;
 import com.cloud.agent.api.storage.CreateEntityDownloadURLCommand;
+import com.cloud.agent.api.storage.GetDatadisksCommand;
 import com.cloud.agent.api.storage.PrimaryStorageDownloadCommand;
 
 public interface VmwareStorageManager {
@@ -49,6 +51,10 @@ public interface VmwareStorageManager {
 
     Answer execute(VmwareHostService hostService, RevertToVMSnapshotCommand cmd);
 
+    Answer execute(VmwareHostService hostService, CreateDatadiskTemplateCommand cmd);
+
+    Answer execute(VmwareHostService hostService, GetDatadisksCommand cmd);
+
     boolean execute(VmwareHostService hostService, CreateEntityDownloadURLCommand cmd);
 
     public void createOva(String path, String name);

plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/manager/VmwareStorageManagerImpl.java

@@ -57,8 +57,12 @@ import com.cloud.agent.api.RevertToVMSnapshotAnswer;
 import com.cloud.agent.api.RevertToVMSnapshotCommand;
 import com.cloud.agent.api.storage.CopyVolumeAnswer;
 import com.cloud.agent.api.storage.CopyVolumeCommand;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateAnswer;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateCommand;
 import com.cloud.agent.api.storage.CreateEntityDownloadURLCommand;
 import com.cloud.agent.api.storage.CreatePrivateTemplateAnswer;
+import com.cloud.agent.api.storage.GetDatadisksAnswer;
+import com.cloud.agent.api.storage.GetDatadisksCommand;
 import com.cloud.agent.api.storage.PrimaryStorageDownloadAnswer;
 import com.cloud.agent.api.storage.PrimaryStorageDownloadCommand;
 import com.cloud.agent.api.to.DataObjectType;
@@ -80,6 +84,7 @@ import com.cloud.storage.JavaStorageLayer;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.StorageLayer;
 import com.cloud.storage.Volume;
+import com.cloud.storage.resource.VmwareStorageLayoutHelper;
 import com.cloud.storage.template.OVAProcessor;
 import com.cloud.utils.NumbersUtil;
 import com.cloud.utils.Pair;
@@ -543,6 +548,160 @@ public class VmwareStorageManagerImpl implements VmwareStorageManager {
         return new CreateVolumeFromSnapshotAnswer(cmd, success, details, newVolumeName);
     }
 
+    @Override
+    public Answer execute(VmwareHostService hostService, GetDatadisksCommand cmd) {
+        List<Ternary<String, Long, Long>> datDiskDetails = new ArrayList<Ternary<String, Long, Long>>();
+        DataTO srcData = cmd.getData();
+        TemplateObjectTO template = (TemplateObjectTO)srcData;
+        DataStoreTO srcStore = srcData.getDataStore();
+        if (!(srcStore instanceof NfsTO)) {
+            return new CreateDatadiskTemplateAnswer("Unsupported protocol");
+        }
+        NfsTO nfsImageStore = (NfsTO)srcStore;
+        String secondaryStorageUrl = nfsImageStore.getUrl();
+        assert (secondaryStorageUrl != null);
+        String secondaryStorageUuid = HypervisorHostHelper.getSecondaryDatastoreUUID(secondaryStorageUrl).replace("-", "");
+        String templateUrl = secondaryStorageUrl + "/" + srcData.getPath();
+        Pair<String, String> templateInfo = VmwareStorageLayoutHelper.decodeTemplateRelativePathAndNameFromUrl(secondaryStorageUrl, templateUrl, template.getName());
+        String templateRelativeFolderPath = templateInfo.first();
+
+        VmwareContext context = hostService.getServiceContext(cmd);
+        try {
+            VmwareHypervisorHost hyperHost = hostService.getHyperHost(context, cmd);
+
+            ManagedObjectReference morDs = HypervisorHostHelper.findDatastoreWithBackwardsCompatibility(hyperHost, secondaryStorageUuid);
+            DatastoreMO datastoreMo = new DatastoreMO(context, morDs);
+
+            String secondaryMountPoint = _mountService.getMountPoint(secondaryStorageUrl);
+            s_logger.info("Secondary storage mount point: " + secondaryMountPoint);
+
+            String srcOVAFileName = VmwareStorageLayoutHelper.getTemplateOnSecStorageFilePath(secondaryMountPoint, templateRelativeFolderPath, templateInfo.second(),
+                    ImageFormat.OVA.getFileExtension());
+
+            String ovfFilePath = getOVFFilePath(srcOVAFileName);
+            if (ovfFilePath == null) {
+                Script command = new Script("tar", 0, s_logger);
+                command.add("--no-same-owner");
+                command.add("-xf", srcOVAFileName);
+                command.setWorkDir(secondaryMountPoint + "/" + templateRelativeFolderPath);
+                s_logger.info("Executing command: " + command.toString());
+                String result = command.execute();
+                if (result != null) {
+                    String msg = "Unable to unpack snapshot OVA file at: " + srcOVAFileName;
+                    s_logger.error(msg);
+                    throw new Exception(msg);
+                }
+            }
+
+            ovfFilePath = getOVFFilePath(srcOVAFileName);
+            if (ovfFilePath == null) {
+                String msg = "Unable to locate OVF file in template package directory: " + srcOVAFileName;
+                s_logger.error(msg);
+                throw new Exception(msg);
+            }
+
+            s_logger.debug("Reading OVF " + ovfFilePath + " to retrive the number of disks present in OVA");
+            List<Pair<String, Boolean>> ovfVolumeDetails = HypervisorHostHelper.readOVF(hyperHost, ovfFilePath, datastoreMo);
+
+            // Get the virtual size of data disk
+            for (Pair<String, Boolean> ovfVolumeDetail : ovfVolumeDetails) {
+                if (ovfVolumeDetail.second()) { // ROOT disk
+                    continue;
+                }
+                String dataDiskPath = ovfVolumeDetail.first();
+                String diskName = dataDiskPath.substring((dataDiskPath.lastIndexOf(File.separator)) + 1);
+                Pair<Long, Long> diskDetails = new OVAProcessor().getDiskDetails(ovfFilePath, diskName);
+                datDiskDetails.add(new Ternary<String, Long, Long>(dataDiskPath, diskDetails.first(), diskDetails.second()));
+            }
+        } catch (Exception e) {
+            String msg = "Get Datadisk Template Count failed due to " + e.getMessage();
+            s_logger.error(msg);
+            return new GetDatadisksAnswer(msg);
+        }
+        return new GetDatadisksAnswer(datDiskDetails);
+    }
+
+    @Override
+    public Answer execute(VmwareHostService hostService, CreateDatadiskTemplateCommand cmd) {
+        TemplateObjectTO diskTemplate = new TemplateObjectTO();
+        TemplateObjectTO dataDiskTemplate = (TemplateObjectTO)cmd.getDataDiskTemplate();
+        DataStoreTO dataStore = dataDiskTemplate.getDataStore();
+        if (!(dataStore instanceof NfsTO)) {
+            return new CreateDatadiskTemplateAnswer("Unsupported protocol");
+        }
+        NfsTO nfsImageStore = (NfsTO)dataStore;
+        String secondaryStorageUrl = nfsImageStore.getUrl();
+        assert (secondaryStorageUrl != null);
+        String secondaryStorageUuid = HypervisorHostHelper.getSecondaryDatastoreUUID(secondaryStorageUrl).replace("-", "");
+
+        VmwareContext context = hostService.getServiceContext(cmd);
+        try {
+            VmwareHypervisorHost hyperHost = hostService.getHyperHost(context, cmd);
+
+            ManagedObjectReference morDs = HypervisorHostHelper.findDatastoreWithBackwardsCompatibility(hyperHost, secondaryStorageUuid);
+            DatastoreMO datastoreMo = new DatastoreMO(context, morDs);
+
+            String secondaryMountPoint = _mountService.getMountPoint(secondaryStorageUrl);
+            s_logger.info("Secondary storage mount point: " + secondaryMountPoint);
+
+            long templateId = dataDiskTemplate.getId();
+            String templateUniqueName = dataDiskTemplate.getUniqueName();
+            String dataDiskPath = cmd.getPath();
+            long virtualSize = dataDiskTemplate.getSize();
+            long fileSize = cmd.getFileSize();
+            String diskName = dataDiskPath.substring((dataDiskPath.lastIndexOf(File.separator)) + 1);
+            long physicalSize = new File(dataDiskPath).length();
+            String dataDiskTemplateFolderPath = getTemplateRelativeDirInSecStorage(dataDiskTemplate.getAccountId(), dataDiskTemplate.getId());
+            String dataDiskTemplateFolderFullPath = secondaryMountPoint + "/" + dataDiskTemplateFolderPath;
+
+            // Create folder to hold datadisk template
+            synchronized (dataDiskTemplateFolderPath.intern()) {
+                Script command = new Script(false, "mkdir", _timeout, s_logger);
+                command.add("-p");
+                command.add(dataDiskTemplateFolderFullPath);
+                String result = command.execute();
+                if (result != null) {
+                    String msg = "Unable to prepare template directory: " + dataDiskTemplateFolderPath + ", storage: " + secondaryStorageUrl + ", error msg: " + result;
+                    s_logger.error(msg);
+                    throw new Exception(msg);
+                }
+            }
+
+            // Copy Datadisk VMDK from parent template folder to Datadisk template folder
+            synchronized (dataDiskPath.intern()) {
+                Script command = new Script(false, "cp", _timeout, s_logger);
+                command.add(dataDiskPath);
+                command.add(dataDiskTemplateFolderFullPath);
+                String result = command.execute();
+                if (result != null) {
+                    String msg = "Unable to copy VMDK from parent template folder to datadisk template folder" + ", error msg: " + result;
+                    s_logger.error(msg);
+                    throw new Exception(msg);
+                }
+            }
+
+            String ovfName = diskName.substring(0, diskName.lastIndexOf("-"));
+            String datastorePath = String.format("[%s] %s", datastoreMo.getName(), dataDiskTemplateFolderPath);
+
+            // Create OVF for Datadisk
+            s_logger.debug("Creating OVF file for datadisk " + diskName + " in " + dataDiskTemplateFolderFullPath);
+            HypervisorHostHelper.createOvfFile(hyperHost, diskName, ovfName, datastorePath, dataDiskTemplateFolderFullPath, virtualSize, fileSize, morDs);
+
+            postCreatePrivateTemplate(dataDiskTemplateFolderFullPath, templateId, templateUniqueName, physicalSize, virtualSize);
+            writeMetaOvaForTemplate(dataDiskTemplateFolderFullPath, ovfName + ".ovf", diskName, templateUniqueName, physicalSize);
+
+            diskTemplate.setId(templateId);
+            diskTemplate.setPath(dataDiskTemplateFolderPath + "/" + templateUniqueName + ".ova");
+            diskTemplate.setSize(virtualSize);
+            diskTemplate.setPhysicalSize(physicalSize);
+        } catch (Exception e) {
+            String msg = "Create Datadisk template failed due to " + e.getMessage();
+            s_logger.error(msg);
+            return new CreateDatadiskTemplateAnswer(msg);
+        }
+        return new CreateDatadiskTemplateAnswer(diskTemplate);
+    }
+
     // templateName: name in secondary storage
     // templateUuid: will be used at hypervisor layer
     private void copyTemplateFromSecondaryToPrimary(VmwareHypervisorHost hyperHost, DatastoreMO datastoreMo, String secondaryStorageUrl,

plugins/hypervisors/vmware/src/com/cloud/hypervisor/vmware/resource/VmwareResource.java

@@ -188,8 +188,12 @@ import com.cloud.agent.api.routing.SetNetworkACLCommand;
 import com.cloud.agent.api.routing.SetSourceNatCommand;
 import com.cloud.agent.api.storage.CopyVolumeAnswer;
 import com.cloud.agent.api.storage.CopyVolumeCommand;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateAnswer;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateCommand;
 import com.cloud.agent.api.storage.CreatePrivateTemplateAnswer;
 import com.cloud.agent.api.storage.DestroyCommand;
+import com.cloud.agent.api.storage.GetDatadisksAnswer;
+import com.cloud.agent.api.storage.GetDatadisksCommand;
 import com.cloud.agent.api.storage.MigrateVolumeAnswer;
 import com.cloud.agent.api.storage.MigrateVolumeCommand;
 import com.cloud.agent.api.storage.PrimaryStorageDownloadAnswer;
@@ -430,6 +434,10 @@ public class VmwareResource implements StoragePoolResource, ServerResource, Vmwa
                 answer = execute((GetStorageStatsCommand)cmd);
             } else if (clz == PrimaryStorageDownloadCommand.class) {
                 answer = execute((PrimaryStorageDownloadCommand)cmd);
+            } else if (clz == CreateDatadiskTemplateCommand.class) {
+                answer = execute((CreateDatadiskTemplateCommand)cmd);
+            } else if (clz == GetDatadisksCommand.class) {
+                answer = execute((GetDatadisksCommand)cmd);
             } else if (clz == GetVncPortCommand.class) {
                 answer = execute((GetVncPortCommand)cmd);
             } else if (clz == SetupCommand.class) {
@@ -3724,6 +3732,44 @@ public class VmwareResource implements StoragePoolResource, ServerResource, Vmwa
         }
     }
 
+    protected GetDatadisksAnswer execute(GetDatadisksCommand cmd) {
+        if (s_logger.isInfoEnabled()) {
+            s_logger.info("Executing resource GetDatadisksCommand: " + _gson.toJson(cmd));
+        }
+        try {
+            VmwareContext context = getServiceContext();
+            VmwareManager mgr = context.getStockObject(VmwareManager.CONTEXT_STOCK_NAME);
+            return (GetDatadisksAnswer)mgr.getStorageManager().execute(this, cmd);
+        } catch (Throwable e) {
+            if (e instanceof RemoteException) {
+                s_logger.warn("Encounter remote exception to vCenter, invalidate VMware session context");
+                invalidateServiceContext();
+            }
+            String msg = "GetDatadisksCommand failed due to " + VmwareHelper.getExceptionMessage(e);
+            s_logger.error(msg, e);
+            return new GetDatadisksAnswer(msg);
+        }
+    }
+
+    protected CreateDatadiskTemplateAnswer execute(CreateDatadiskTemplateCommand cmd) {
+        if (s_logger.isInfoEnabled()) {
+            s_logger.info("Executing resource CreateDatadiskTemplatesCommand: " + _gson.toJson(cmd));
+        }
+        try {
+            VmwareContext context = getServiceContext();
+            VmwareManager mgr = context.getStockObject(VmwareManager.CONTEXT_STOCK_NAME);
+            return (CreateDatadiskTemplateAnswer)mgr.getStorageManager().execute(this, cmd);
+        } catch (Throwable e) {
+            if (e instanceof RemoteException) {
+                s_logger.warn("Encounter remote exception to vCenter, invalidate VMware session context");
+                invalidateServiceContext();
+            }
+            String msg = "CreateDatadiskTemplatesCommand failed due to " + VmwareHelper.getExceptionMessage(e);
+            s_logger.error(msg, e);
+            return new CreateDatadiskTemplateAnswer(msg);
+        }
+    }
+
     protected Answer execute(PvlanSetupCommand cmd) {
         // Pvlan related operations are performed in the start/stop command paths
         // for vmware. This function is implemented to support mgmt layer code

plugins/hypervisors/vmware/src/com/cloud/storage/resource/VmwareSecondaryStorageResourceHandler.java

@@ -33,7 +33,9 @@ import com.cloud.agent.api.CreatePrivateTemplateFromSnapshotCommand;
 import com.cloud.agent.api.CreatePrivateTemplateFromVolumeCommand;
 import com.cloud.agent.api.CreateVolumeFromSnapshotCommand;
 import com.cloud.agent.api.storage.CopyVolumeCommand;
+import com.cloud.agent.api.storage.CreateDatadiskTemplateCommand;
 import com.cloud.agent.api.storage.CreateEntityDownloadURLCommand;
+import com.cloud.agent.api.storage.GetDatadisksCommand;
 import com.cloud.agent.api.storage.PrimaryStorageDownloadCommand;
 import com.cloud.hypervisor.vmware.manager.VmwareHostService;
 import com.cloud.hypervisor.vmware.manager.VmwareStorageManager;
@@ -98,6 +100,10 @@ public class VmwareSecondaryStorageResourceHandler implements SecondaryStorageRe
                 answer = storageSubsystemHandler.handleStorageCommands((StorageSubSystemCommand)cmd);
             } else if (cmd instanceof CreateEntityDownloadURLCommand) {
                 answer = execute((CreateEntityDownloadURLCommand)cmd);
+            } else if (cmd instanceof CreateDatadiskTemplateCommand) {
+                answer = execute((CreateDatadiskTemplateCommand)cmd);
+            } else if (cmd instanceof GetDatadisksCommand) {
+                answer = execute((GetDatadisksCommand)cmd);
             } else {
                 answer = _resource.defaultAction(cmd);
             }
@@ -174,6 +180,14 @@ public class VmwareSecondaryStorageResourceHandler implements SecondaryStorageRe
         return _storageMgr.execute(this, cmd);
     }
 
+    private Answer execute(CreateDatadiskTemplateCommand cmd) {
+        return _storageMgr.execute(this, cmd);
+    }
+
+    private Answer execute(GetDatadisksCommand cmd) {
+        return _storageMgr.execute(this, cmd);
+    }
+
     @Override
     public VmwareContext getServiceContext(Command cmd) {
         String guid = cmd.getContextParam("guid");

plugins/hypervisors/xen/src/com/cloud/hypervisor/XenServerGuru.java

@@ -145,7 +145,7 @@ public class XenServerGuru extends HypervisorGuruBase implements HypervisorGuru
                 DataStoreTO destStore = destData.getDataStore();
                 if (srcStore instanceof NfsTO && destStore instanceof NfsTO) {
                     HostVO host = hostDao.findById(hostId);
-                    EndPoint ep = endPointSelector.selectHypervisorHost(new ZoneScope(host.getDataCenterId()));
+                    EndPoint ep = endPointSelector.selectHypervisorHostByType(new ZoneScope(host.getDataCenterId()), null);
                     host = hostDao.findById(ep.getId());
                     hostDao.loadDetails(host);
                     String snapshotHotFixVersion = host.getDetail(XenserverConfigs.XS620HotFix);

server/src/com/cloud/template/HypervisorTemplateAdapter.java

@@ -19,6 +19,7 @@ package com.cloud.template;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -48,6 +49,7 @@ import org.apache.cloudstack.framework.async.AsyncCompletionCallback;
 import org.apache.cloudstack.framework.async.AsyncRpcContext;
 import org.apache.cloudstack.framework.messagebus.MessageBus;
 import org.apache.cloudstack.framework.messagebus.PublishScope;
+import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreVO;
 import org.apache.cloudstack.storage.image.datastore.ImageStoreEntity;
 
@@ -72,9 +74,13 @@ import com.cloud.storage.dao.VMTemplateZoneDao;
 import com.cloud.storage.download.DownloadMonitor;
 import com.cloud.user.Account;
 import com.cloud.utils.Pair;
+import com.cloud.utils.Ternary;
 import com.cloud.utils.UriUtils;
 import com.cloud.utils.db.DB;
 import com.cloud.utils.db.EntityManager;
+import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionCallbackNoReturn;
+import com.cloud.utils.db.TransactionStatus;
 import com.cloud.utils.exception.CloudRuntimeException;
 
 @Local(value = TemplateAdapter.class)
@@ -96,6 +102,8 @@ public class HypervisorTemplateAdapter extends TemplateAdapterBase {
     @Inject
     AlertManager alertMgr;
     @Inject
+    TemplateDataStoreDao _tmplStoreDao;
+    @Inject
     VMTemplateZoneDao templateZoneDao;
     @Inject
     EndPointSelector _epSelector;
@@ -290,6 +298,10 @@ public class HypervisorTemplateAdapter extends TemplateAdapterBase {
         TemplateInfo template = context.template;
         if (result.isSuccess()) {
             VMTemplateVO tmplt = _tmpltDao.findById(template.getId());
+            // Check if OVA contains additional data disks. If yes, create Datadisk templates for each of the additional datadisk present in the OVA
+            if (template.getFormat().equals(ImageFormat.OVA)) {
+                createDataDiskTemplates(template);
+            }
             // need to grant permission for public templates
             if (tmplt.isPublicTemplate()) {
                 _messageBus.publish(_name, TemplateManager.MESSAGE_REGISTER_PUBLIC_TEMPLATE_EVENT, PublishScope.LOCAL, tmplt.getId());
@@ -331,6 +343,60 @@ public class HypervisorTemplateAdapter extends TemplateAdapterBase {
         return null;
     }
 
+    private void createDataDiskTemplates(TemplateInfo parentTemplate) {
+        TemplateApiResult result = null;
+        VMTemplateVO template = _tmpltDao.findById(parentTemplate.getId());
+        DataStore imageStore = parentTemplate.getDataStore();
+        List<Ternary<String, Long, Long>> dataDiskTemplates = imageService.getDatadiskTemplates(parentTemplate);
+        s_logger.error("Found " + dataDiskTemplates.size() + " Datadisk templates for template: " + parentTemplate.getId());
+        int diskCount = 1;
+        for (Ternary<String, Long, Long> dataDiskTemplate : dataDiskTemplates) {
+            // Make an entry in vm_template table
+            final long templateId = _templateDao.getNextInSequence(Long.class, "id");
+            VMTemplateVO templateVO = new VMTemplateVO(templateId, template.getName() + "-DataDiskTemplate-" + diskCount, template.getFormat(), false, false, false,
+                    TemplateType.DATADISK, template.getUrl(), template.requiresHvm(), template.getBits(), template.getAccountId(), null,
+                    template.getDisplayText() + "-DataDiskTemplate", false, 0, false, template.getHypervisorType(), null, null, false, false);
+            templateVO.setParentTemplateId(template.getId());
+            templateVO.setSize(dataDiskTemplate.second());
+            templateVO = _templateDao.persist(templateVO);
+            // Make sync call to create Datadisk templates in image store
+            TemplateInfo dataDiskTemplateInfo = imageFactory.getTemplate(templateVO.getId(), imageStore);
+            AsyncCallFuture<TemplateApiResult> future = imageService.createDatadiskTemplateAsync(parentTemplate, dataDiskTemplateInfo, dataDiskTemplate.first(),
+                    dataDiskTemplate.third());
+            try {
+                result = future.get();
+                if (result.isSuccess()) {
+                    // Make an entry in template_zone_ref table
+                    if (imageStore.getScope().getScopeType() == ScopeType.REGION) {
+                        imageService.associateTemplateToZone(templateId, null);
+                    } else if (imageStore.getScope().getScopeType() == ScopeType.ZONE) {
+                        Long zoneId = ((ImageStoreEntity)imageStore).getDataCenterId();
+                        VMTemplateZoneVO templateZone = new VMTemplateZoneVO(zoneId, templateId, new Date());
+                        _tmpltZoneDao.persist(templateZone);
+                    }
+                    _resourceLimitMgr.incrementResourceCount(template.getAccountId(), ResourceType.secondary_storage, templateVO.getSize());
+                } else {
+                    // Cleanup Datadisk template enries in case of failure
+                    Transaction.execute(new TransactionCallbackNoReturn() {
+                        @Override
+                        public void doInTransactionWithoutResult(TransactionStatus status) {
+                            _tmplStoreDao.deletePrimaryRecordsForTemplate(templateId);
+                            _tmpltZoneDao.deletePrimaryRecordsForTemplate(templateId);
+                            _tmpltDao.expunge(templateId);
+                        }
+                    });
+                    // Continue to create the remaining Datadisk templates even if creation of 1 Datadisk template failes
+                    s_logger.error("Creation of Datadisk: " + templateVO.getId() + " failed: " + result.getResult());
+                    continue;
+                }
+            } catch (Exception e) {
+                s_logger.error("Creation of Datadisk: " + templateVO.getId() + " failed: " + result.getResult());
+                continue;
+            }
+            diskCount++;
+        }
+    }
+
     @Override
     @DB
     public boolean delete(TemplateProfile profile) {

server/src/com/cloud/template/TemplateManagerImpl.java

@@ -680,6 +680,11 @@ public class TemplateManagerImpl extends ManagerBase implements TemplateManager,
             throw new InvalidParameterValueException("Unable to find template with id");
         }
 
+        // Verify template is not Datadisk template
+        if (template.getTemplateType().equals(TemplateType.DATADISK)) {
+            throw new InvalidParameterValueException("Template " + template.getId() + " is of type Datadisk. Cannot copy Datadisk templates.");
+        }
+
         DataStore srcSecStore = null;
         if (sourceZoneId != null) {
             // template is on zone-wide secondary storage

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

@@ -223,3 +223,5 @@ CREATE VIEW `cloud`.`volume_view` AS
         `cloud`.`async_job` ON async_job.instance_id = volumes.id
             and async_job.instance_type = 'Volume'
             and async_job.job_status = 0;
+
+ALTER TABLE `cloud`.`vm_template` ADD COLUMN `parent_template_id` bigint(20) unsigned DEFAULT NULL COMMENT 'If datadisk template, then id of the root template this template belongs to';

vmware-base/src/com/cloud/hypervisor/vmware/mo/HypervisorHostHelper.java

@@ -85,7 +85,6 @@ import com.cloud.hypervisor.vmware.util.VmwareHelper;
 import com.cloud.network.Networks.BroadcastDomainType;
 import com.cloud.utils.ActionDelegate;
 import com.cloud.utils.Pair;
-import com.cloud.utils.Ternary;
 import com.cloud.utils.cisco.n1kv.vsm.NetconfHelper;
 import com.cloud.utils.cisco.n1kv.vsm.PolicyMap;
 import com.cloud.utils.cisco.n1kv.vsm.PortProfile;
@@ -104,7 +103,6 @@ public class HypervisorHostHelper {
 
     // make vmware-base loosely coupled with cloud-specific stuff, duplicate VLAN.UNTAGGED constant here
     private static final String UNTAGGED_VLAN_NAME = "untagged";
-    private static final String VMDK_PACK_DIR = "ova";
     private static final String OVA_OPTION_KEY_BOOTDISK = "cloud.ova.bootdisk";
 
     public static VirtualMachineMO findVmFromObjectContent(VmwareContext context, ObjectContent[] ocs, String name, String instanceNameCustomField) {
@@ -142,6 +140,10 @@ public class HypervisorHostHelper {
         return morDs;
     }
 
+    public static String getSecondaryDatastoreUUID(String storeUrl) {
+        return UUID.nameUUIDFromBytes(storeUrl.getBytes()).toString();
+    }
+
     public static DatastoreMO getHyperHostDatastoreMO(VmwareHypervisorHost hyperHost, String datastoreName) throws Exception {
         ObjectContent[] ocs = hyperHost.getDatastorePropertiesOnHyperHost(new String[] {"name"});
         if (ocs != null && ocs.length > 0) {
@@ -1188,10 +1190,13 @@ public class HypervisorHostHelper {
         return false;
     }
 
-    public static List<Ternary<String, Long, Boolean>> readOVF(VmwareHypervisorHost host, String ovfFilePath, DatastoreMO dsMo, ManagedObjectReference morRp,
-            ManagedObjectReference morHost) throws Exception {
+    public static List<Pair<String, Boolean>> readOVF(VmwareHypervisorHost host, String ovfFilePath, DatastoreMO dsMo) throws Exception {
+        List<Pair<String, Boolean>> ovfVolumeInfos = new ArrayList<Pair<String, Boolean>>();
+        List<String> files = new ArrayList<String>();
 
+        ManagedObjectReference morRp = host.getHyperHostOwnerResourcePool();
         assert (morRp != null);
+        ManagedObjectReference morHost = host.getMor();
 
         String importEntityName = UUID.randomUUID().toString();
         OvfCreateImportSpecParams importSpecParams = new OvfCreateImportSpecParams();
@@ -1211,7 +1216,7 @@ public class HypervisorHostHelper {
             throw new Exception(msg);
         }
 
-        if(!ovfImportResult.getError().isEmpty()) {
+        if (!ovfImportResult.getError().isEmpty()) {
             for (LocalizedMethodFault fault : ovfImportResult.getError()) {
                 s_logger.error("createImportSpec error: " + fault.getLocalizedMessage());
             }
@@ -1224,8 +1229,7 @@ public class HypervisorHostHelper {
             }
         }
 
-        VirtualMachineImportSpec importSpec = new VirtualMachineImportSpec();
-        importSpec = (VirtualMachineImportSpec)ovfImportResult.getImportSpec();
+        VirtualMachineImportSpec importSpec = (VirtualMachineImportSpec)ovfImportResult.getImportSpec();
         if (importSpec == null) {
             String msg = "createImportSpec() failed to create import specification for OVF template at " + ovfFilePath;
             s_logger.error(msg);
@@ -1233,15 +1237,8 @@ public class HypervisorHostHelper {
         }
 
         File ovfFile = new File(ovfFilePath);
-        int diskCount = 0;
-        long sizeKb = 0;
-        List<Ternary<String, Long, Boolean>> ovfVolumeInfos = new ArrayList<Ternary<String, Long, Boolean>>();
-        Ternary<String, Long, Boolean> ovfVolumeInfo = null;
-        List<String> files = new ArrayList<String>();
-        String absFile = null;
-
         for (OvfFileItem ovfFileItem : ovfImportResult.getFileItem()) {
-            absFile = ovfFile.getParent() + File.separator + ovfFileItem.getPath();
+            String absFile = ovfFile.getParent() + File.separator + ovfFileItem.getPath();
             files.add(absFile);
         }
 
@@ -1256,19 +1253,21 @@ public class HypervisorHostHelper {
             }
         }
 
-        Boolean osDisk = true;
+        int diskCount = 0;
+        int deviceCount = 0;
         List<VirtualDeviceConfigSpec> deviceConfigList = config.getDeviceChange();
         for (VirtualDeviceConfigSpec deviceSpec : deviceConfigList) {
+            Boolean osDisk = false;
             VirtualDevice device = deviceSpec.getDevice();
             if (device instanceof VirtualDisk) {
-                sizeKb = ((VirtualDisk)device).getCapacityInKB();
-                if (diskCount == osDiskSeqNumber) {
+                if ((osDiskSeqNumber == 0 && diskCount == 0) || osDiskSeqNumber == deviceCount) {
                     osDisk = true;
                 }
-                diskCount++;
-                ovfVolumeInfo = new Ternary<String, Long, Boolean>(files.get(diskCount), sizeKb, osDisk);
+                Pair<String, Boolean> ovfVolumeInfo = new Pair<String, Boolean>(files.get(diskCount), osDisk);
                 ovfVolumeInfos.add(ovfVolumeInfo);
+                diskCount++;
             }
+            deviceCount++;
         }
         return ovfVolumeInfos;
     }
@@ -1338,46 +1337,48 @@ public class HypervisorHostHelper {
         return paramVal;
     }
 
-    public static void createOvfFile(VmwareHypervisorHost host, String diskFileName, String ovfName, String dir, long size, ManagedObjectReference morDs) throws Exception {
+    public static void createOvfFile(VmwareHypervisorHost host, String diskFileName, String ovfName, String datastorePath, String templatePath, long diskCapacity, long fileSize,
+            ManagedObjectReference morDs) throws Exception {
         VmwareContext context = host.getContext();
+        ManagedObjectReference morOvf = context.getServiceContent().getOvfManager();
         VirtualMachineMO workerVmMo = HypervisorHostHelper.createWorkerVM(host, new DatastoreMO(context, morDs), ovfName);
         if (workerVmMo == null)
             throw new Exception("Unable to find just-created worker VM");
 
-        String[] disks = {dir + File.separator + diskFileName};
-        boolean bSuccess = false;
+        String[] disks = {datastorePath + File.separator + diskFileName};
         try {
             VirtualMachineConfigSpec vmConfigSpec = new VirtualMachineConfigSpec();
             VirtualDeviceConfigSpec deviceConfigSpec = new VirtualDeviceConfigSpec();
 
-            VirtualDisk device = VmwareHelper.prepareDiskDevice(workerVmMo, null, -1, disks, morDs, -1, 1);
-            device.setCapacityInKB(size);
-
+            // Reconfigure worker VM with datadisk
+            VirtualDevice device = VmwareHelper.prepareDiskDevice(workerVmMo, null, -1, disks, morDs, -1, 1);
             deviceConfigSpec.setDevice(device);
             deviceConfigSpec.setOperation(VirtualDeviceConfigSpecOperation.ADD);
             vmConfigSpec.getDeviceChange().add(deviceConfigSpec);
             workerVmMo.configureVm(vmConfigSpec);
+
+            // Write OVF descriptor file
+            OvfCreateDescriptorParams ovfDescParams = new OvfCreateDescriptorParams();
             String deviceId = File.separator + workerVmMo.getMor().getValue() + File.separator + "VirtualIDEController0:0";
-            bSuccess = true;
             OvfFile ovfFile = new OvfFile();
             ovfFile.setPath(diskFileName);
             ovfFile.setDeviceId(deviceId);
-            ovfFile.setSize(size);
-            // write OVF descriptor file
-            OvfCreateDescriptorParams ovfDescParams = new OvfCreateDescriptorParams();
+            ovfFile.setSize(fileSize);
+            ovfFile.setCapacity(diskCapacity);
             ovfDescParams.getOvfFiles().add(ovfFile);
-            ManagedObjectReference morOvf = context.getServiceContent().getOvfManager();
             OvfCreateDescriptorResult ovfCreateDescriptorResult = context.getService().createDescriptor(morOvf, workerVmMo.getMor(), ovfDescParams);
 
-            String ovfPath = dir + File.separator + ovfName + ".ovf";
-            FileWriter out = new FileWriter(ovfPath);
-            out.write(ovfCreateDescriptorResult.getOvfDescriptor());
-            out.close();
-        } finally {
-            if (!bSuccess) {
-                workerVmMo.detachAllDisks();
-                workerVmMo.destroy();
+            String ovfPath = templatePath + File.separator + ovfName + ".ovf";
+            try {
+                FileWriter out = new FileWriter(ovfPath);
+                out.write(ovfCreateDescriptorResult.getOvfDescriptor());
+                out.close();
+            } catch (Exception e) {
+                throw e;
             }
+        } finally {
+            workerVmMo.detachAllDisks();
+            workerVmMo.destroy();
         }
     }