diff --git a/api/src/com/cloud/agent/api/storage/DownloadCommand.java b/api/src/com/cloud/agent/api/storage/DownloadCommand.java
index fc955068054..126ee46cdd5 100644
--- a/api/src/com/cloud/agent/api/storage/DownloadCommand.java
+++ b/api/src/com/cloud/agent/api/storage/DownloadCommand.java
@@ -14,6 +14,7 @@ package com.cloud.agent.api.storage;
 
 import java.net.URI;
 
+import com.cloud.storage.Volume;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.template.VirtualMachineTemplate;
 
@@ -41,6 +42,10 @@ public class DownloadCommand extends AbstractDownloadCommand {
 		}
 	}
 	
+    public static enum ResourceType {
+        VOLUME, TEMPLATE
+    }
+	
 	public static class Proxy {
 		private String _host;
 		private int _port;
@@ -97,6 +102,7 @@ public class DownloadCommand extends AbstractDownloadCommand {
 	private Proxy _proxy;
 	private Long maxDownloadSizeInBytes = null;
 	private long id;
+	private ResourceType resourceType = ResourceType.TEMPLATE;
 	
 	protected DownloadCommand() {
 	}
@@ -122,6 +128,17 @@ public class DownloadCommand extends AbstractDownloadCommand {
 	    this.setSecUrl(secUrl);
 	    this.maxDownloadSizeInBytes = maxDownloadSizeInBytes;
 	}
+	
+	public DownloadCommand(String secUrl, Volume volume, Long maxDownloadSizeInBytes, String checkSum, String url) {
+	    super(volume.getName(), url, ImageFormat.VHD, volume.getAccountId());
+	    //this.hvm = volume.isRequiresHvm();
+	    this.checksum = checkSum;
+	    this.id = volume.getId();
+	    //this.description = volume.get;
+	    this.setSecUrl(secUrl);
+	    this.maxDownloadSizeInBytes = maxDownloadSizeInBytes;
+	    this.resourceType = ResourceType.VOLUME;
+	}
 
 	public DownloadCommand(String secUrl, String url, VirtualMachineTemplate template, String user, String passwd, Long maxDownloadSizeInBytes) {
 	    super(template.getUniqueName(), url, template.getFormat(), template.getAccountId());
@@ -187,4 +204,14 @@ public class DownloadCommand extends AbstractDownloadCommand {
 	public Long getMaxDownloadSizeInBytes() {
 		return maxDownloadSizeInBytes;
 	}
+
+
+	public ResourceType getResourceType() {
+		return resourceType;
+	}
+
+
+	public void setResourceType(ResourceType resourceType) {
+		this.resourceType = resourceType;
+	}
 }
diff --git a/api/src/com/cloud/storage/StorageService.java b/api/src/com/cloud/storage/StorageService.java
index 782775ac7e3..7553f9e84b2 100644
--- a/api/src/com/cloud/storage/StorageService.java
+++ b/api/src/com/cloud/storage/StorageService.java
@@ -21,6 +21,7 @@ import com.cloud.api.commands.CreateVolumeCmd;
 import com.cloud.api.commands.DeletePoolCmd;
 import com.cloud.api.commands.ListVolumesCmd;
 import com.cloud.api.commands.UpdateStoragePoolCmd;
+import com.cloud.api.commands.UploadVolumeCmd;
 import com.cloud.exception.ConcurrentOperationException;
 import com.cloud.exception.InsufficientCapacityException;
 import com.cloud.exception.PermissionDeniedException;
@@ -108,4 +109,13 @@ public interface StorageService{
 
     List<? extends Volume> searchForVolumes(ListVolumesCmd cmd);
 
+    /**
+     * Uploads the volume to secondary storage
+     * 
+     * @param UploadVolumeCmd cmd 
+     *            
+     * @return Volume object
+     */
+    Volume uploadVolume(UploadVolumeCmd cmd)	throws ResourceAllocationException;
+
 }
diff --git a/core/src/com/cloud/storage/VolumeHostVO.java b/core/src/com/cloud/storage/VolumeHostVO.java
new file mode 100755
index 00000000000..6b826c67973
--- /dev/null
+++ b/core/src/com/cloud/storage/VolumeHostVO.java
@@ -0,0 +1,283 @@
+package com.cloud.storage;
+
+import java.util.Date;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.EnumType;
+import javax.persistence.Enumerated;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import javax.persistence.Temporal;
+import javax.persistence.TemporalType;
+
+//import com.cloud.storage.VMVolumeStorageResourceAssoc.Status;
+import com.cloud.storage.VMTemplateStorageResourceAssoc.Status;
+import com.cloud.utils.db.GenericDaoBase;
+
+/**
+ * Join table for storage hosts and volumes
+ * @author Nitin
+ *
+ */
+@Entity
+@Table(name="volume_host_ref")
+public class VolumeHostVO {
+	@Id
+	@GeneratedValue(strategy=GenerationType.IDENTITY)
+	Long id;
+	
+	@Column(name="host_id")
+	private long hostId;
+	
+	@Column(name="volume_id")
+	private long volumeId;
+	
+	@Column(name=GenericDaoBase.CREATED_COLUMN)
+	private Date created = null;
+	
+	@Column(name="last_updated")
+	@Temporal(value=TemporalType.TIMESTAMP)
+	private Date lastUpdated = null;
+	
+	@Column (name="download_pct")
+	private int downloadPercent;
+	
+	@Column (name="size")
+	private long size;
+	
+	@Column (name="physical_size")
+	private long physicalSize;
+	
+	@Column (name="download_state")
+	@Enumerated(EnumType.STRING)
+	private Status downloadState;
+	
+    @Column(name="checksum")
+    private String checksum;
+	
+	@Column (name="local_path")
+	private String localDownloadPath;
+	
+	@Column (name="error_str")
+	private String errorString;
+	
+	@Column (name="job_id")
+	private String jobId;	
+	
+	@Column (name="install_path")
+    private String installPath;
+	
+	@Column (name="url")
+	private String downloadUrl;
+
+	@Column(name="is_copy")
+	private boolean isCopy = false;
+    
+    @Column(name="destroyed")
+    boolean destroyed = false;
+    
+	
+    public String getInstallPath() {
+		return installPath;
+	}
+
+	public long getHostId() {
+		return hostId;
+	}
+
+	public void setHostId(long hostId) {
+		this.hostId = hostId;
+	}
+
+	
+    public long getVolumeId() {
+		return volumeId;
+	}
+
+	
+    public void setVolumeId(long volumeId) {
+		this.volumeId = volumeId;
+	}
+
+	
+    public int getDownloadPercent() {
+		return downloadPercent;
+	}
+
+	
+    public void setDownloadPercent(int downloadPercent) {
+		this.downloadPercent = downloadPercent;
+	}
+
+	
+    public void setDownloadState(Status downloadState) {
+		this.downloadState = downloadState;
+	}
+
+	
+    public long getId() {
+		return id;
+	}
+
+	
+    public Date getCreated() {
+		return created;
+	}
+
+	
+    public Date getLastUpdated() {
+		return lastUpdated;
+	}
+	
+	
+    public void setLastUpdated(Date date) {
+	    lastUpdated = date;
+	}
+	
+	
+    public void setInstallPath(String installPath) {
+	    this.installPath = installPath;
+	}
+
+	
+    public Status getDownloadState() {
+		return downloadState;
+	}
+
+	public String getChecksum() {
+		return checksum;
+	}
+
+	public void setChecksum(String checksum) {
+		this.checksum = checksum;
+	}
+
+	public VolumeHostVO(long hostId, long volumeId) {
+		super();
+		this.hostId = hostId;
+		this.volumeId = volumeId;
+	}
+
+	public VolumeHostVO(long hostId, long volumeId, Date lastUpdated,
+			int downloadPercent, Status downloadState,
+			String localDownloadPath, String errorString, String jobId,
+			String installPath, String downloadUrl, String checksum) {
+		//super();
+		this.hostId = hostId;
+		this.volumeId = volumeId;
+		this.lastUpdated = lastUpdated;
+		this.downloadPercent = downloadPercent;
+		this.downloadState = downloadState;
+		this.localDownloadPath = localDownloadPath;
+		this.errorString = errorString;
+		this.jobId = jobId;
+		this.installPath = installPath;
+		this.setDownloadUrl(downloadUrl);
+		this.checksum = checksum;
+	}
+
+	protected VolumeHostVO() {
+		
+	}
+
+	
+    public void setLocalDownloadPath(String localPath) {
+		this.localDownloadPath = localPath;
+	}
+
+	
+    public String getLocalDownloadPath() {
+		return localDownloadPath;
+	}
+
+	
+    public void setErrorString(String errorString) {
+		this.errorString = errorString;
+	}
+
+	
+    public String getErrorString() {
+		return errorString;
+	}
+
+	
+    public void setJobId(String jobId) {
+		this.jobId = jobId;
+	}
+
+	
+    public String getJobId() {
+		return jobId;
+	}
+
+	
+	public boolean equals(Object obj) {
+		if (obj instanceof VolumeHostVO) {
+			VolumeHostVO other = (VolumeHostVO)obj;
+			return (this.volumeId==other.getVolumeId() && this.hostId==other.getHostId());		   
+		}
+		return false;
+	}
+
+	
+	public int hashCode() {
+		Long tid = new Long(volumeId);
+		Long hid = new Long(hostId);
+		return tid.hashCode()+hid.hashCode();
+	}
+
+    public void setSize(long size) {
+        this.size = size;
+    }
+
+    public long getSize() {
+        return size;
+    }
+	
+    
+    public void setPhysicalSize(long physicalSize) {
+        this.physicalSize = physicalSize;
+    }
+
+    public long getPhysicalSize() {
+        return physicalSize;
+    }
+
+    public void setDestroyed(boolean destroyed) {
+    	this.destroyed = destroyed;
+    }
+
+    public boolean getDestroyed() {
+    	return destroyed;
+    }
+
+	public void setDownloadUrl(String downloadUrl) {
+		this.downloadUrl = downloadUrl;
+	}
+
+	public String getDownloadUrl() {
+		return downloadUrl;
+	}
+
+	public void setCopy(boolean isCopy) {
+		this.isCopy = isCopy;
+	}
+
+	public boolean isCopy() {
+		return isCopy;
+	}
+	
+	
+    public long getVolumeSize() {
+	    return -1;
+	}
+	
+	
+    public String toString() {
+	    return new StringBuilder("VolumeHost[").append(id).append("-").append(volumeId).append("-").append(hostId).append(installPath).append("]").toString();
+	}
+
+}
diff --git a/core/src/com/cloud/storage/template/DownloadManager.java b/core/src/com/cloud/storage/template/DownloadManager.java
index 833297ec829..94e8094b793 100644
--- a/core/src/com/cloud/storage/template/DownloadManager.java
+++ b/core/src/com/cloud/storage/template/DownloadManager.java
@@ -18,6 +18,7 @@ import java.util.Map;
 import com.cloud.agent.api.storage.DownloadAnswer;
 import com.cloud.agent.api.storage.DownloadCommand;
 import com.cloud.agent.api.storage.DownloadCommand.Proxy;
+import com.cloud.agent.api.storage.DownloadCommand.ResourceType;
 import com.cloud.storage.VMTemplateHostVO;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.resource.SecondaryStorageResource;
@@ -38,10 +39,11 @@ public interface DownloadManager extends Manager {
 	 * @param descr	description of the template
 	 * @param user username used for authentication to the server
 	 * @param password password used for authentication to the server
-	 * @param maxDownloadSizeInBytes (optional) max download size for the template, in bytes. 
+	 * @param maxDownloadSizeInBytes (optional) max download size for the template, in bytes.
+	 * @param resourceType signifying the type of resource like template, volume etc. 
 	 * @return job-id that can be used to interrogate the status of the download.
 	 */
-	public String downloadPublicTemplate(long id, String url, String name, ImageFormat format, boolean hvm, Long accountId, String descr, String cksum, String installPathPrefix, String userName, String passwd, long maxDownloadSizeInBytes, Proxy proxy);
+	public String downloadPublicTemplate(long id, String url, String name, ImageFormat format, boolean hvm, Long accountId, String descr, String cksum, String installPathPrefix, String userName, String passwd, long maxDownloadSizeInBytes, Proxy proxy, ResourceType resourceType);
 	
 	
 	/**
@@ -84,11 +86,11 @@ public interface DownloadManager extends Manager {
 	 * @return answer representing status of download.
 	 */
 	public DownloadAnswer handleDownloadCommand(SecondaryStorageResource resource, DownloadCommand cmd);
-	
-	
+		
 	/**
 	/**
 	 * @return list of template info for installed templates
 	 */
 	public Map<String, TemplateInfo> gatherTemplateInfo(String templateDir);
+
 }
\ No newline at end of file
diff --git a/core/src/com/cloud/storage/template/DownloadManagerImpl.java b/core/src/com/cloud/storage/template/DownloadManagerImpl.java
index 156dd0b25a5..bf77ac80d17 100755
--- a/core/src/com/cloud/storage/template/DownloadManagerImpl.java
+++ b/core/src/com/cloud/storage/template/DownloadManagerImpl.java
@@ -43,6 +43,7 @@ import com.cloud.agent.api.Answer;
 import com.cloud.agent.api.storage.DownloadAnswer;
 import com.cloud.agent.api.storage.DownloadCommand;
 import com.cloud.agent.api.storage.DownloadCommand.Proxy;
+import com.cloud.agent.api.storage.DownloadCommand.ResourceType;
 import com.cloud.agent.api.storage.DownloadProgressCommand;
 import com.cloud.agent.api.storage.DownloadProgressCommand.RequestType;
 import com.cloud.exception.InternalErrorException;
@@ -100,8 +101,9 @@ public class DownloadManagerImpl implements DownloadManager {
         private long templatesize;
         private long templatePhysicalSize;
         private long id;
+		private ResourceType resourceType;
 
-        public DownloadJob(TemplateDownloader td, String jobId, long id, String tmpltName, ImageFormat format, boolean hvm, Long accountId, String descr, String cksum, String installPathPrefix) {
+        public DownloadJob(TemplateDownloader td, String jobId, long id, String tmpltName, ImageFormat format, boolean hvm, Long accountId, String descr, String cksum, String installPathPrefix, ResourceType resourceType) {
             super();
             this.td = td;
             this.jobId = jobId;
@@ -114,6 +116,7 @@ public class DownloadManagerImpl implements DownloadManager {
             this.installPathPrefix = installPathPrefix;
             this.templatesize = 0;
             this.id = id;
+            this.resourceType = resourceType;
         }
 
         public TemplateDownloader getTd() {
@@ -156,7 +159,11 @@ public class DownloadManagerImpl implements DownloadManager {
             return id;
         }
 
-        public void setTmpltPath(String tmpltPath) {
+        public ResourceType getResourceType() {
+			return resourceType;
+		}
+
+		public void setTmpltPath(String tmpltPath) {
             this.tmpltPath = tmpltPath;
         }
 
@@ -206,7 +213,9 @@ public class DownloadManagerImpl implements DownloadManager {
 
     public static final Logger s_logger = Logger.getLogger(DownloadManagerImpl.class);
     private String _templateDir;
+    private String _volumeDir;
     private String createTmpltScr;
+    private String createVolScr;
     private Adapters<Processor> processors;
 
     private ExecutorService threadPool;
@@ -308,19 +317,22 @@ public class DownloadManagerImpl implements DownloadManager {
     private String postDownload(String jobId) {
         DownloadJob dnld = jobs.get(jobId);
         TemplateDownloader td = dnld.getTemplateDownloader();
-        String templatePath = null;
+        String templatePath = null;        
         templatePath = dnld.getInstallPathPrefix() + dnld.getAccountId() + File.separator + dnld.getId() + File.separator;// dnld.getTmpltName();
+        ResourceType resourceType = dnld.getResourceType();
 
         _storage.mkdirs(templatePath);
 
         // once template path is set, remove the parent dir so that the template is installed with a relative path
-        String finalTemplatePath = _templateDir + File.separator + dnld.getAccountId() + File.separator + dnld.getId() + File.separator;
+        String finalTemplatePath = resourceType == ResourceType.TEMPLATE ? _templateDir : _volumeDir 
+        	+ File.separator + dnld.getAccountId() + File.separator + dnld.getId() + File.separator;
         dnld.setTmpltPath(finalTemplatePath);
 
         int imgSizeGigs = (int) Math.ceil(_storage.getSize(td.getDownloadLocalPath()) * 1.0d / (1024 * 1024 * 1024));
         imgSizeGigs++; // add one just in case
         long timeout = imgSizeGigs * installTimeoutPerGig;
         Script scr = null;
+        String script = resourceType == ResourceType.TEMPLATE ? createTmpltScr : createVolScr;
         scr = new Script(createTmpltScr, timeout, s_logger);
         scr.add("-s", Integer.toString(imgSizeGigs));
         scr.add("-S", Long.toString(td.getMaxTemplateSizeInBytes()));
@@ -417,7 +429,7 @@ public class DownloadManagerImpl implements DownloadManager {
     }
 
     @Override
-    public String downloadPublicTemplate(long id, String url, String name, ImageFormat format, boolean hvm, Long accountId, String descr, String cksum, String installPathPrefix, String user, String password, long maxTemplateSizeInBytes, Proxy proxy) {
+    public String downloadPublicTemplate(long id, String url, String name, ImageFormat format, boolean hvm, Long accountId, String descr, String cksum, String installPathPrefix, String user, String password, long maxTemplateSizeInBytes, Proxy proxy, ResourceType resourceType) {
         UUID uuid = UUID.randomUUID();
         String jobId = uuid.toString();
         String tmpDir = installPathPrefix + File.separator + accountId + File.separator + id;
@@ -428,8 +440,9 @@ public class DownloadManagerImpl implements DownloadManager {
                 s_logger.warn("Unable to create " + tmpDir);
                 return "Unable to create " + tmpDir;
             }
-
-            File file = _storage.getFile(tmpDir + File.separator + TemplateLocation.Filename);
+            //	TO DO - define constant for volume properties.
+            File file = ResourceType.TEMPLATE == resourceType ? _storage.getFile(tmpDir + File.separator + TemplateLocation.Filename) : 
+            	_storage.getFile(tmpDir + File.separator + "volume.properties");
             if ( file.exists() ) {
                 file.delete();
             }
@@ -448,7 +461,7 @@ public class DownloadManagerImpl implements DownloadManager {
             TemplateDownloader td;
             if ((uri != null) && (uri.getScheme() != null)) {
                 if (uri.getScheme().equalsIgnoreCase("http") || uri.getScheme().equalsIgnoreCase("https")) {
-                    td = new HttpTemplateDownloader(_storage, url, tmpDir, new Completion(jobId), maxTemplateSizeInBytes, user, password, proxy);
+                    td = new HttpTemplateDownloader(_storage, url, tmpDir, new Completion(jobId), maxTemplateSizeInBytes, user, password, proxy, resourceType);
                 } else if (uri.getScheme().equalsIgnoreCase("file")) {
                     td = new LocalTemplateDownloader(_storage, url, tmpDir, maxTemplateSizeInBytes, new Completion(jobId));
                 } else if (uri.getScheme().equalsIgnoreCase("scp")) {
@@ -463,7 +476,7 @@ public class DownloadManagerImpl implements DownloadManager {
             } else {
                 throw new CloudRuntimeException("Unable to download from URL: " + url);
             }
-            DownloadJob dj = new DownloadJob(td, jobId, id, name, format, hvm, accountId, descr, cksum, installPathPrefix);
+            DownloadJob dj = new DownloadJob(td, jobId, id, name, format, hvm, accountId, descr, cksum, installPathPrefix, resourceType);
             jobs.put(jobId, dj);
             threadPool.execute(td);
 
@@ -555,12 +568,13 @@ public class DownloadManagerImpl implements DownloadManager {
 
     @Override
     public DownloadAnswer handleDownloadCommand(SecondaryStorageResource resource, DownloadCommand cmd) {
+    	ResourceType resourceType = cmd.getResourceType();
         if (cmd instanceof DownloadProgressCommand) {
             return handleDownloadProgressCmd( resource, (DownloadProgressCommand) cmd);
         }
 
         if (cmd.getUrl() == null) {
-            return new DownloadAnswer("Template is corrupted on storage due to an invalid url , cannot download", VMTemplateStorageResourceAssoc.Status.DOWNLOAD_ERROR);
+            return new DownloadAnswer(resourceType.toString() + " is corrupted on storage due to an invalid url , cannot download", VMTemplateStorageResourceAssoc.Status.DOWNLOAD_ERROR);
         }
 
         if (cmd.getName() == null) {
@@ -568,7 +582,11 @@ public class DownloadManagerImpl implements DownloadManager {
         }
 
         String installPathPrefix = null;
-        installPathPrefix = resource.getRootDir(cmd) + File.separator + _templateDir;
+        if (ResourceType.TEMPLATE == resourceType){
+        	installPathPrefix = resource.getRootDir(cmd) + File.separator + _templateDir;
+        }else {
+        	installPathPrefix = resource.getRootDir(cmd) + File.separator + _volumeDir;
+        }
 
         String user = null;
         String password = null;
@@ -576,9 +594,9 @@ public class DownloadManagerImpl implements DownloadManager {
             user = cmd.getAuth().getUserName();
             password = new String(cmd.getAuth().getPassword());
         }
-
+        //TO DO - Define Volume max size as well
         long maxDownloadSizeInBytes = (cmd.getMaxDownloadSizeInBytes() == null) ? TemplateDownloader.DEFAULT_MAX_TEMPLATE_SIZE_IN_BYTES : (cmd.getMaxDownloadSizeInBytes());
-        String jobId = downloadPublicTemplate(cmd.getId(), cmd.getUrl(), cmd.getName(), cmd.getFormat(), cmd.isHvm(), cmd.getAccountId(), cmd.getDescription(), cmd.getChecksum(), installPathPrefix, user, password, maxDownloadSizeInBytes, cmd.getProxy());
+        String jobId = downloadPublicTemplate(cmd.getId(), cmd.getUrl(), cmd.getName(), cmd.getFormat(), cmd.isHvm(), cmd.getAccountId(), cmd.getDescription(), cmd.getChecksum(), installPathPrefix, user, password, maxDownloadSizeInBytes, cmd.getProxy(), resourceType);
         sleep();
         if (jobId == null) {
             return new DownloadAnswer("Internal Error", VMTemplateStorageResourceAssoc.Status.DOWNLOAD_ERROR);
@@ -831,6 +849,12 @@ public class DownloadManagerImpl implements DownloadManager {
         }
         s_logger.info("createtmplt.sh found in " + createTmpltScr);
 
+        createVolScr = Script.findScript(scriptsDir, "createvolume.sh");
+        if (createVolScr == null) {
+            throw new ConfigurationException("Unable to find createvolume.sh");
+        }
+        s_logger.info("createvolume.sh found in " + createVolScr);
+
         List<ComponentInfo<Adapter>> processors = new ArrayList<ComponentInfo<Adapter>>();
 
         Processor processor = new VhdProcessor();
@@ -860,6 +884,7 @@ public class DownloadManagerImpl implements DownloadManager {
             _templateDir = TemplateConstants.DEFAULT_TMPLT_ROOT_DIR;
         }
         _templateDir += File.separator + TemplateConstants.DEFAULT_TMPLT_FIRST_LEVEL_DIR;
+        _volumeDir = TemplateConstants.DEFAULT_VOLUME_ROOT_DIR;
         // Add more processors here.
         threadPool = Executors.newFixedThreadPool(numInstallThreads);
         return true;
diff --git a/core/src/com/cloud/storage/template/HttpTemplateDownloader.java b/core/src/com/cloud/storage/template/HttpTemplateDownloader.java
index d93d61a2ef0..fd5acc94fa0 100644
--- a/core/src/com/cloud/storage/template/HttpTemplateDownloader.java
+++ b/core/src/com/cloud/storage/template/HttpTemplateDownloader.java
@@ -39,8 +39,10 @@ import org.apache.commons.httpclient.auth.AuthScope;
 import org.apache.commons.httpclient.methods.GetMethod;
 import org.apache.commons.httpclient.params.HttpMethodParams;
 import org.apache.log4j.Logger;
+import org.jnetpcap.util.resolver.Resolver.ResolverType;
 
 import com.cloud.agent.api.storage.DownloadCommand.Proxy;
+import com.cloud.agent.api.storage.DownloadCommand.ResourceType;
 import com.cloud.storage.StorageLayer;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.utils.Pair;
@@ -70,15 +72,17 @@ public class HttpTemplateDownloader implements TemplateDownloader {
 
 	private String toDir;
 	private long MAX_TEMPLATE_SIZE_IN_BYTES;
-
+	private ResourceType resourceType = ResourceType.TEMPLATE;
 	private final HttpMethodRetryHandler myretryhandler;
 
-	public HttpTemplateDownloader (StorageLayer storageLayer, String downloadUrl, String toDir, DownloadCompleteCallback callback, long maxTemplateSizeInBytes, String user, String password, Proxy proxy) {
+	
+
+	public HttpTemplateDownloader (StorageLayer storageLayer, String downloadUrl, String toDir, DownloadCompleteCallback callback, long maxTemplateSizeInBytes, String user, String password, Proxy proxy, ResourceType resourceType) {
 		this._storage = storageLayer;
 		this.downloadUrl = downloadUrl;
 		this.setToDir(toDir);
 		this.status = TemplateDownloader.Status.NOT_STARTED;
-		
+		this.resourceType = resourceType;
 		this.MAX_TEMPLATE_SIZE_IN_BYTES = maxTemplateSizeInBytes;
 		
 		this.totalBytes = 0;
@@ -428,7 +432,7 @@ public class HttpTemplateDownloader implements TemplateDownloader {
 			// TODO Auto-generated catch block
 			e.printStackTrace();
 		}
-		TemplateDownloader td = new HttpTemplateDownloader(null, url,"/tmp/mysql", null, TemplateDownloader.DEFAULT_MAX_TEMPLATE_SIZE_IN_BYTES, null, null, null);
+		TemplateDownloader td = new HttpTemplateDownloader(null, url,"/tmp/mysql", null, TemplateDownloader.DEFAULT_MAX_TEMPLATE_SIZE_IN_BYTES, null, null, null, null);
 		long bytes = td.download(true, null);
 		if (bytes > 0) {
 			System.out.println("Downloaded  (" + bytes + " bytes)" + " in " + td.getDownloadTime()/1000 + " secs");
@@ -450,4 +454,9 @@ public class HttpTemplateDownloader implements TemplateDownloader {
 		return inited;
 	}
 
+
+	public ResourceType getResourceType() {
+		return resourceType;
+	}
+
 }
diff --git a/core/src/com/cloud/storage/template/TemplateConstants.java b/core/src/com/cloud/storage/template/TemplateConstants.java
index e038e7f2408..ec2aa4cf90c 100644
--- a/core/src/com/cloud/storage/template/TemplateConstants.java
+++ b/core/src/com/cloud/storage/template/TemplateConstants.java
@@ -18,6 +18,7 @@ package com.cloud.storage.template;
  */
 public final class TemplateConstants {
 	public static final String DEFAULT_TMPLT_ROOT_DIR = "template";
+	public static final String DEFAULT_VOLUME_ROOT_DIR = "volume";
 	public static final String DEFAULT_TMPLT_FIRST_LEVEL_DIR = "tmpl/";
 	
 	public static final String DEFAULT_SYSTEM_VM_TEMPLATE_PATH = "template/tmpl/1/";
diff --git a/scripts/storage/secondary/createvolume.sh b/scripts/storage/secondary/createvolume.sh
new file mode 100755
index 00000000000..cc2120b7f75
--- /dev/null
+++ b/scripts/storage/secondary/createvolume.sh
@@ -0,0 +1,229 @@
+#!/usr/bin/env bash
+# Copyright (C) 2011 Citrix Systems, Inc.  All rights reserved
+#     
+# This software is licensed under the GNU General Public License v3 or later.
+# 
+# It is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+# 
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+# 
+
+
+
+ 
+
+# $Id: createtmplt.sh 9132 2010-06-04 20:17:43Z manuel $ $HeadURL: svn://svn.lab.vmops.com/repos/vmdev/java/scripts/storage/secondary/createtmplt.sh $
+# createtmplt.sh -- install a volume
+
+usage() {
+  printf "Usage: %s: -t <volume-fs> -n <volumename> -f <root disk file> -c <md5 cksum> -d <descr> -h  [-u] [-v]\n" $(basename $0) >&2
+}
+
+
+#set -x
+ulimit -f 41943040 #40GiB in blocks
+ulimit -c 0
+
+rollback_if_needed() {
+  if [ $2 -gt 0 ]
+  then
+    printf "$3\n"
+    #back out all changes
+    rm -rf $1
+    exit 2
+fi
+}
+
+verify_cksum() {
+  echo  "$1  $2" | md5sum  -c --status
+  #printf "$1\t$2" | md5sum  -c --status
+  if [ $? -gt 0 ] 
+  then
+    printf "Checksum failed, not proceeding with install\n"
+    exit 3
+  fi
+}
+
+untar() {
+  local ft=$(file $1| awk -F" " '{print $2}')
+  case $ft in
+  USTAR) 
+     printf "tar archives not supported\n"  >&2
+     return 1
+          ;;
+  *) printf "$1"
+     return 0
+	  ;;
+  esac
+
+}
+
+is_compressed() {
+  local ft=$(file $1| awk -F" " '{print $2}')
+  local tmpfile=${1}.tmp
+
+  case $ft in
+  gzip)  ctype="gzip"
+         ;;
+  bzip2)  ctype="bz2"
+         ;;
+  ZIP)  ctype="zip"
+        ;;
+    *) echo "File $1 does not appear to be compressed" >&2
+        return 1
+	;;
+  esac
+  echo "Uncompressing to $tmpfile (type $ctype)...could take a long time" >&2
+  return 0
+}
+
+uncompress() {
+  local ft=$(file $1| awk -F" " '{print $2}')
+  local tmpfile=${1}.tmp
+
+  case $ft in
+  gzip)  gunzip -q -c $1 > $tmpfile
+         ;;
+  bzip2)  bunzip2 -q -c $1 > $tmpfile
+         ;;
+  ZIP)  unzip -q -p $1 | cat > $tmpfile
+        ;;
+    *) printf "$1"
+       return 0
+	;;
+  esac
+
+  if [ $? -gt 0 ] 
+  then
+    printf "Failed to uncompress file (filetype=$ft), exiting "
+    return 1 
+  fi
+ 
+  rm -f $1
+  printf $tmpfile
+
+  return 0
+}
+
+create_from_file() {
+  local tmpltfs=$1
+  local tmpltimg=$2
+  local tmpltname=$3
+
+  [ -n "$verbose" ] && echo "Moving to $tmpltfs/$tmpltname...could take a while" >&2
+  mv $tmpltimg /$tmpltfs/$tmpltname
+
+}
+
+tflag=
+nflag=
+fflag=
+sflag=
+hflag=
+hvm=false
+cleanup=false
+dflag=
+cflag=
+
+while getopts 'vuht:n:f:s:c:d:S:' OPTION
+do
+  case $OPTION in
+  t)	tflag=1
+		tmpltfs="$OPTARG"
+		;;
+  n)	nflag=1
+		tmpltname="$OPTARG"
+		;;
+  f)	fflag=1
+		tmpltimg="$OPTARG"
+		;;
+  s)	sflag=1
+		;;
+  c)	cflag=1
+		cksum="$OPTARG"
+		;;
+  d)	dflag=1
+		descr="$OPTARG"
+		;;
+  S)	Sflag=1
+		size=$OPTARG
+                let "size>>=10"
+		ulimit -f $size
+		;;
+  h)	hflag=1
+		hvm="true"
+		;;
+  u)	cleanup="true"
+		;;
+  v)	verbose="true"
+		;;
+  ?)	usage
+		exit 2
+		;;
+  esac
+done
+
+if [ "$tflag$nflag$fflag$sflag" != "1111" ]
+then
+ usage
+ exit 2
+fi
+
+mkdir -p $tmpltfs
+
+if [ ! -f $tmpltimg ] 
+then
+  printf "root disk file $tmpltimg doesn't exist\n"
+  exit 3
+fi
+
+if [ -n "$cksum" ]
+then
+  verify_cksum $cksum $tmpltimg
+fi
+[ -n "$verbose" ] && is_compressed $tmpltimg
+tmpltimg2=$(uncompress $tmpltimg)
+rollback_if_needed $tmpltfs $? "failed to uncompress $tmpltimg\n"
+
+tmpltimg2=$(untar $tmpltimg2)
+rollback_if_needed $tmpltfs $? "tar archives not supported\n"
+
+if [ ${tmpltname%.vhd} != ${tmpltname} ]
+then
+  if  which  vhd-util &>/dev/null
+  then 
+    vhd-util check -n ${tmpltimg2} > /dev/null
+    rollback_if_needed $tmpltfs $? "vhd check of $tmpltimg2 failed\n"
+    vhd-util set -n ${tmpltimg2} -f "hidden" -v "0" > /dev/null
+    rollback_if_needed $tmpltfs $? "vhd remove $tmpltimg2 hidden failed\n"
+  fi
+fi
+
+imgsize=$(ls -l $tmpltimg2| awk -F" " '{print $5}')
+
+create_from_file $tmpltfs $tmpltimg2 $tmpltname
+
+touch /$tmpltfs/volume.properties
+rollback_if_needed $tmpltfs $? "Failed to create volume.properties file"
+echo -n "" > /$tmpltfs/volume.properties
+
+today=$(date '+%m_%d_%Y')
+echo "filename=$tmpltname" > /$tmpltfs/volume.properties
+echo "description=$descr" >> /$tmpltfs/volume.properties
+echo "checksum=$cksum" >> /$tmpltfs/volume.properties
+echo "hvm=$hvm" >> /$tmpltfs/volume.properties
+echo "size=$imgsize" >> /$tmpltfs/volume.properties
+
+if [ "$cleanup" == "true" ]
+then
+  rm -f $tmpltimg
+fi
+
+exit 0
diff --git a/server/src/com/cloud/configuration/Config.java b/server/src/com/cloud/configuration/Config.java
index b92c9598ccd..fd98789652a 100755
--- a/server/src/com/cloud/configuration/Config.java
+++ b/server/src/com/cloud/configuration/Config.java
@@ -65,6 +65,7 @@ public enum Config {
 	StorageOverprovisioningFactor("Storage", StoragePoolAllocator.class, String.class, "storage.overprovisioning.factor", "2", "Used for storage overprovisioning calculation; available storage will be (actualStorageSize * storage.overprovisioning.factor)", null),
 	StorageStatsInterval("Storage", ManagementServer.class, String.class, "storage.stats.interval", "60000", "The interval (in milliseconds) when storage stats (per host) are retrieved from agents.", null),
 	MaxVolumeSize("Storage", ManagementServer.class, Integer.class, "storage.max.volume.size", "2000", "The maximum size for a volume (in GB).", null),
+	MaxUploadVolumeSize("Storage",  ManagementServer.class, Integer.class, "storage.max.volume.upload.size", "50", "The maximum size for a uploaded volume(in GB).", null),
 	TotalRetries("Storage", AgentManager.class, Integer.class, "total.retries", "4", "The number of times each command sent to a host should be retried in case of failure.", null),
 	StoragePoolMaxWaitSeconds("Storage", ManagementServer.class, Integer.class, "storage.pool.max.waitseconds", "3600", "Timeout (in seconds) to synchronize storage pool operations.", null),
 	StorageTemplateCleanupEnabled("Storage", ManagementServer.class, Boolean.class, "storage.template.cleanup.enabled", "true", "Enable/disable template cleanup activity, only take effect when overall storage cleanup is enabled", null),
diff --git a/server/src/com/cloud/configuration/DefaultComponentLibrary.java b/server/src/com/cloud/configuration/DefaultComponentLibrary.java
index f3a196f648b..adf44645d59 100755
--- a/server/src/com/cloud/configuration/DefaultComponentLibrary.java
+++ b/server/src/com/cloud/configuration/DefaultComponentLibrary.java
@@ -153,6 +153,7 @@ import com.cloud.storage.dao.VMTemplatePoolDaoImpl;
 import com.cloud.storage.dao.VMTemplateSwiftDaoImpl;
 import com.cloud.storage.dao.VMTemplateZoneDaoImpl;
 import com.cloud.storage.dao.VolumeDaoImpl;
+import com.cloud.storage.dao.VolumeHostDaoImpl;
 import com.cloud.storage.download.DownloadMonitorImpl;
 import com.cloud.storage.secondary.SecondaryStorageManagerImpl;
 import com.cloud.storage.snapshot.SnapshotManagerImpl;
@@ -249,6 +250,7 @@ public class DefaultComponentLibrary extends ComponentLibraryBase implements Com
         addDao("ResourceCountDao", ResourceCountDaoImpl.class);
         addDao("UserAccountDao", UserAccountDaoImpl.class);
         addDao("VMTemplateHostDao", VMTemplateHostDaoImpl.class);
+        addDao("VolumeHostDao", VolumeHostDaoImpl.class);
         addDao("VMTemplateSwiftDao", VMTemplateSwiftDaoImpl.class);
         addDao("UploadDao", UploadDaoImpl.class);
         addDao("VMTemplatePoolDao", VMTemplatePoolDaoImpl.class);
diff --git a/server/src/com/cloud/server/ConfigurationServerImpl.java b/server/src/com/cloud/server/ConfigurationServerImpl.java
index 7a686f78986..352289e9779 100755
--- a/server/src/com/cloud/server/ConfigurationServerImpl.java
+++ b/server/src/com/cloud/server/ConfigurationServerImpl.java
@@ -193,9 +193,11 @@ public class ConfigurationServerImpl implements ConfigurationServer {
             createServiceOffering(User.UID_SYSTEM, "Small Instance", 1, 512, 500, "Small Instance", false, false, null);
             createServiceOffering(User.UID_SYSTEM, "Medium Instance", 1, 1024, 1000, "Medium Instance", false, false, null);
             // Save default disk offerings
-            createdefaultDiskOffering(null, "Small", "Small Disk, 5 GB", 5, null);
-            createdefaultDiskOffering(null, "Medium", "Medium Disk, 20 GB", 20, null);
-            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", 100, null);
+            createdefaultDiskOffering(null, "Small", "Small Disk, 5 GB", 5, null, false);
+            createdefaultDiskOffering(null, "Medium", "Medium Disk, 20 GB", 20, null, false);
+            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", 100, null, false);
+            createdefaultDiskOffering(null, "Large", "Large Disk, 100 GB", 100, null, false);
+            createdefaultDiskOffering(null, "Custom", "Custom Disk", 0, null, true);
 
             // Save the mount parent to the configuration table
             String mountParent = getMountParent();
@@ -818,12 +820,12 @@ public class ConfigurationServerImpl implements ConfigurationServer {
         return pod;
     }
 
-    private DiskOfferingVO createdefaultDiskOffering(Long domainId, String name, String description, int numGibibytes, String tags) {
+    private DiskOfferingVO createdefaultDiskOffering(Long domainId, String name, String description, int numGibibytes, String tags, boolean isCustomized) {
         long diskSize = numGibibytes;
         diskSize = diskSize * 1024 * 1024 * 1024;
         tags = cleanupTags(tags);
 
-        DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, diskSize, tags, false);
+        DiskOfferingVO newDiskOffering = new DiskOfferingVO(domainId, name, description, diskSize, tags, isCustomized);
         newDiskOffering.setUniqueName("Cloud.Com-" + name);
         newDiskOffering = _diskOfferingDao.persistDeafultDiskOffering(newDiskOffering);
         return newDiskOffering;
diff --git a/server/src/com/cloud/storage/StorageManagerImpl.java b/server/src/com/cloud/storage/StorageManagerImpl.java
index 6daf1dd1068..8c1694b8ee9 100755
--- a/server/src/com/cloud/storage/StorageManagerImpl.java
+++ b/server/src/com/cloud/storage/StorageManagerImpl.java
@@ -144,6 +144,7 @@ import com.cloud.storage.dao.VMTemplateHostDao;
 import com.cloud.storage.dao.VMTemplatePoolDao;
 import com.cloud.storage.dao.VMTemplateSwiftDao;
 import com.cloud.storage.dao.VolumeDao;
+import com.cloud.storage.download.DownloadMonitor;
 import com.cloud.storage.listener.StoragePoolMonitor;
 import com.cloud.storage.secondary.SecondaryStorageVmManager;
 import com.cloud.storage.snapshot.SnapshotManager;
@@ -307,6 +308,8 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
     protected ResourceManager _resourceMgr;
     @Inject
     protected CheckPointManager _checkPointMgr;
+    @Inject
+    protected DownloadMonitor _downloadMonitor;
 
     @Inject(adapter = StoragePoolAllocator.class)
     protected Adapters<StoragePoolAllocator> _storagePoolAllocators;
@@ -1633,12 +1636,12 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
     /*
      * Just allocate a volume in the database, don't send the createvolume cmd to hypervisor. The volume will be finally
      * created
-     * only when it's attached to a VM.
+     * 
      */
-//    @Override
+    @Override
     @DB
     @ActionEvent(eventType = EventTypes.EVENT_VOLUME_CREATE, eventDescription = "creating volume", create = true)
-    public VolumeVO createVolumeEntry(UploadVolumeCmd cmd) throws ResourceAllocationException{
+    public VolumeVO uploadVolume(UploadVolumeCmd cmd) throws ResourceAllocationException{
     	Account caller = UserContext.current().getCaller();
         long ownerId = cmd.getEntityOwnerId();
         Long zoneId = cmd.getZoneId();
@@ -1646,8 +1649,9 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         String url = cmd.getUrl();
         
     	validateVolume(caller, ownerId, zoneId, volumeName, url, cmd.getFormat());
-    	persistVolume();
-		return null;
+    	VolumeVO volume = persistVolume(caller, ownerId, zoneId, volumeName, url, cmd.getFormat());
+    	_downloadMonitor.downloadVolumeToStorage(volume, zoneId, url, cmd.getChecksum());
+		return volume;
     	
     }
     
@@ -1732,9 +1736,35 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
     	
     }
     
-    private boolean persistVolume() {
+    private VolumeVO persistVolume(Account caller, long ownerId, Long zoneId, String volumeName, String url, String format) {
     	
-		return false;
+        Transaction txn = Transaction.currentTxn();
+        txn.start();
+
+        VolumeVO volume = new VolumeVO(volumeName, zoneId, -1, -1, -1, new Long(-1), null, null, 0, Volume.Type.DATADISK);
+        volume.setPoolId(null);
+        volume.setDataCenterId(zoneId);
+        volume.setPodId(null);
+        volume.setAccountId(ownerId);
+        volume.setDomainId(((caller == null) ? Domain.ROOT_DOMAIN : caller.getDomainId()));
+        long diskOfferingId = _diskOfferingDao.findByUniqueName("Cloud.com-Custom").getId();
+        volume.setDiskOfferingId(diskOfferingId);
+        //volume.setSize(size);
+        volume.setInstanceId(null);
+        volume.setUpdated(new Date());
+        volume.setDomainId((caller == null) ? Domain.ROOT_DOMAIN : caller.getDomainId());
+
+        volume = _volsDao.persist(volume);
+        UsageEventVO usageEvent = new UsageEventVO(EventTypes.EVENT_VOLUME_CREATE, volume.getAccountId(), volume.getDataCenterId(), volume.getId(), volume.getName(), diskOfferingId, null, 0l);
+        _usageEventDao.persist(usageEvent);
+
+        UserContext.current().setEventDetails("Volume Id: " + volume.getId());
+
+        // Increment resource count during allocation; if actual creation fails, decrement it
+        _resourceLimitMgr.incrementResourceCount(volume.getAccountId(), ResourceType.volume);
+
+        txn.commit();
+		return volume;
 	}
     	    
     
diff --git a/server/src/com/cloud/storage/dao/VolumeHostDao.java b/server/src/com/cloud/storage/dao/VolumeHostDao.java
new file mode 100755
index 00000000000..26e13ac43d5
--- /dev/null
+++ b/server/src/com/cloud/storage/dao/VolumeHostDao.java
@@ -0,0 +1,10 @@
+package com.cloud.storage.dao;
+
+import com.cloud.storage.VolumeHostVO;
+import com.cloud.utils.db.GenericDao;
+
+public interface VolumeHostDao extends GenericDao<VolumeHostVO, Long> {
+
+	VolumeHostVO findByHostVolume(long id, long id2);
+
+}
diff --git a/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java b/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java
new file mode 100755
index 00000000000..d95d50196e4
--- /dev/null
+++ b/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java
@@ -0,0 +1,34 @@
+package com.cloud.storage.dao;
+
+
+import javax.ejb.Local;
+
+import com.cloud.storage.VolumeHostVO;
+import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+@Local(value={VolumeHostDao.class})
+public class VolumeHostDaoImpl extends GenericDaoBase<VolumeHostVO, Long> implements VolumeHostDao {
+
+	protected final SearchBuilder<VolumeHostVO> HostVolumeSearch;
+	
+	VolumeHostDaoImpl(){
+		HostVolumeSearch = createSearchBuilder();
+		HostVolumeSearch.and("host_id", HostVolumeSearch.entity().getHostId(), SearchCriteria.Op.EQ);
+		HostVolumeSearch.and("volume_id", HostVolumeSearch.entity().getVolumeId(), SearchCriteria.Op.EQ);
+		HostVolumeSearch.and("destroyed", HostVolumeSearch.entity().getDestroyed(), SearchCriteria.Op.EQ);
+		HostVolumeSearch.done();
+	}
+    
+    
+    
+	@Override
+	public VolumeHostVO findByHostVolume(long hostId, long volumeId) {
+		SearchCriteria<VolumeHostVO> sc = HostVolumeSearch.create();
+	    sc.setParameters("host_id", hostId);
+	    sc.setParameters("volume_id", volumeId);
+        sc.setParameters("destroyed", false);
+        return findOneIncludingRemovedBy(sc);
+	}	
+
+}
diff --git a/server/src/com/cloud/storage/download/DownloadListener.java b/server/src/com/cloud/storage/download/DownloadListener.java
index e13a119226b..654958498f4 100755
--- a/server/src/com/cloud/storage/download/DownloadListener.java
+++ b/server/src/com/cloud/storage/download/DownloadListener.java
@@ -37,12 +37,17 @@ import com.cloud.agent.api.storage.DownloadProgressCommand.RequestType;
 import com.cloud.exception.AgentUnavailableException;
 import com.cloud.exception.ConnectionException;
 import com.cloud.host.HostVO;
+
 import com.cloud.storage.Storage;
 import com.cloud.storage.VMTemplateHostVO;
+import com.cloud.storage.VolumeHostVO;
+import com.cloud.storage.VolumeVO;
 import com.cloud.storage.VMTemplateStorageResourceAssoc.Status;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.dao.VMTemplateDao;
 import com.cloud.storage.dao.VMTemplateHostDao;
+import com.cloud.storage.dao.VolumeDao;
+import com.cloud.storage.dao.VolumeHostDao;
 import com.cloud.storage.download.DownloadState.DownloadEvent;
 import com.cloud.utils.exception.CloudRuntimeException;
 
@@ -98,9 +103,12 @@ public class DownloadListener implements Listener {
 	private HostVO sserver;
 	private HostVO ssAgent;
 	private VMTemplateVO template;
+	private VolumeVO volume;
 	
 	private boolean downloadActive = true;
 
+	private VolumeHostDao volumeHostDao;
+	private VolumeDao _volumeDao;	
 	private VMTemplateHostDao vmTemplateHostDao;
 	private VMTemplateDao _vmTemplateDao;
 
@@ -119,6 +127,7 @@ public class DownloadListener implements Listener {
 	
 	private final Map<String,  DownloadState> stateMap = new HashMap<String, DownloadState>();
 	private Long templateHostId;
+	private Long volumeHostId;
 	
 	public DownloadListener(HostVO ssAgent, HostVO host, VMTemplateVO template, Timer _timer, VMTemplateHostDao dao, Long templHostId, DownloadMonitorImpl downloadMonitor, DownloadCommand cmd, VMTemplateDao templateDao) {
 	    this.ssAgent = ssAgent;
@@ -137,6 +146,24 @@ public class DownloadListener implements Listener {
 		updateDatabase(Status.NOT_DOWNLOADED, "");
 	}
 	
+	public DownloadListener(HostVO ssAgent, HostVO host, VolumeVO volume, Timer _timer, VolumeHostDao dao, Long volHostId, DownloadMonitorImpl downloadMonitor, DownloadCommand cmd, VolumeDao volumeDao) {
+	    this.ssAgent = ssAgent;
+        this.sserver = host;
+		this.volume = volume;
+		this.volumeHostDao = dao;
+		this.downloadMonitor = downloadMonitor;
+		this.cmd = cmd;
+		this.volumeHostId = volHostId;
+		initStateMachine();
+		this.currState=getState(Status.NOT_DOWNLOADED.toString());
+		this.timer = _timer;
+		this.timeoutTask = new TimeoutTask(this);
+		this.timer.schedule(timeoutTask, 3*STATUS_POLL_INTERVAL);
+		this._volumeDao = volumeDao;
+		updateDatabase(Status.NOT_DOWNLOADED, "");
+	}
+	
+	
 	public void setCurrState(VMTemplateHostVO.Status currState) {
 		this.currState = getState(currState.toString());
 	}
@@ -181,15 +208,27 @@ public class DownloadListener implements Listener {
 	}
 
 	public synchronized void updateDatabase(Status state, String errorString) {
-	    VMTemplateHostVO vo = vmTemplateHostDao.createForUpdate();
-		vo.setDownloadState(state);
-		vo.setLastUpdated(new Date());
-		vo.setErrorString(errorString);
-		vmTemplateHostDao.update(getTemplateHostId(), vo);
+		if (template != null){
+		    VMTemplateHostVO vo = vmTemplateHostDao.createForUpdate();
+			vo.setDownloadState(state);
+			vo.setLastUpdated(new Date());
+			vo.setErrorString(errorString);
+			vmTemplateHostDao.update(getTemplateHostId(), vo);
+		}else {
+		    VolumeHostVO vo = volumeHostDao.createForUpdate();
+			vo.setDownloadState(state);
+			vo.setLastUpdated(new Date());
+			vo.setErrorString(errorString);
+			volumeHostDao.update(getVolumeHostId(), vo);
+		}
 	}
 	
 	public void log(String message, Level level) {
-		s_logger.log(level, message + ", template=" + template.getName() + " at host " + sserver.getName());
+		if (template != null){
+			s_logger.log(level, message + ", template=" + template.getName() + " at host " + sserver.getName());
+		}else {
+			s_logger.log(level, message + ", volume=" + volume.getName() + " at host " + sserver.getName());
+		}
 	}
 
 	private Long getTemplateHostId() {
@@ -199,11 +238,21 @@ public class DownloadListener implements Listener {
 		}
 		return templateHostId;
 	}
+	
+	private Long getVolumeHostId() {
+		if (volumeHostId == null){
+			VolumeHostVO volHost = volumeHostDao.findByHostVolume(sserver.getId(), volume.getId());
+			volumeHostId = volHost.getId();
+		}
+		return volumeHostId;
+	}
 
 	public DownloadListener(DownloadMonitorImpl monitor) {
 	    downloadMonitor = monitor;
 	}
+
 	
+
 	@Override
 	public boolean isRecurring() {
 		return false;
@@ -247,23 +296,44 @@ public class DownloadListener implements Listener {
 	}
 
 	public synchronized void updateDatabase(DownloadAnswer answer) {
-        VMTemplateHostVO updateBuilder = vmTemplateHostDao.createForUpdate();
-		updateBuilder.setDownloadPercent(answer.getDownloadPct());
-		updateBuilder.setDownloadState(answer.getDownloadStatus());
-		updateBuilder.setLastUpdated(new Date());
-		updateBuilder.setErrorString(answer.getErrorString());
-		updateBuilder.setJobId(answer.getJobId());
-		updateBuilder.setLocalDownloadPath(answer.getDownloadPath());
-		updateBuilder.setInstallPath(answer.getInstallPath());
-		updateBuilder.setSize(answer.getTemplateSize());
-		updateBuilder.setPhysicalSize(answer.getTemplatePhySicalSize());
-		
-		vmTemplateHostDao.update(getTemplateHostId(), updateBuilder);
-		
-		if (answer.getCheckSum() != null) {
-			VMTemplateVO templateDaoBuilder = _vmTemplateDao.createForUpdate();
-			templateDaoBuilder.setChecksum(answer.getCheckSum());
-			_vmTemplateDao.update(template.getId(), templateDaoBuilder);
+		if (template != null){
+	        VMTemplateHostVO updateBuilder = vmTemplateHostDao.createForUpdate();
+			updateBuilder.setDownloadPercent(answer.getDownloadPct());
+			updateBuilder.setDownloadState(answer.getDownloadStatus());
+			updateBuilder.setLastUpdated(new Date());
+			updateBuilder.setErrorString(answer.getErrorString());
+			updateBuilder.setJobId(answer.getJobId());
+			updateBuilder.setLocalDownloadPath(answer.getDownloadPath());
+			updateBuilder.setInstallPath(answer.getInstallPath());
+			updateBuilder.setSize(answer.getTemplateSize());
+			updateBuilder.setPhysicalSize(answer.getTemplatePhySicalSize());
+			
+			vmTemplateHostDao.update(getTemplateHostId(), updateBuilder);
+			
+			if (answer.getCheckSum() != null) {
+				VMTemplateVO templateDaoBuilder = _vmTemplateDao.createForUpdate();
+				templateDaoBuilder.setChecksum(answer.getCheckSum());
+				_vmTemplateDao.update(template.getId(), templateDaoBuilder);
+			}
+		} else {
+	        VolumeHostVO updateBuilder = volumeHostDao.createForUpdate();
+			updateBuilder.setDownloadPercent(answer.getDownloadPct());
+			updateBuilder.setDownloadState(answer.getDownloadStatus());
+			updateBuilder.setLastUpdated(new Date());
+			updateBuilder.setErrorString(answer.getErrorString());
+			updateBuilder.setJobId(answer.getJobId());
+			updateBuilder.setLocalDownloadPath(answer.getDownloadPath());
+			updateBuilder.setInstallPath(answer.getInstallPath());
+			updateBuilder.setSize(answer.getTemplateSize());
+			updateBuilder.setPhysicalSize(answer.getTemplatePhySicalSize());
+			
+			volumeHostDao.update(getVolumeHostId(), updateBuilder);
+			
+			/*if (answer.getCheckSum() != null) {
+				VMTemplateVO templateDaoBuilder = _vmTemplateDao.createForUpdate();
+				templateDaoBuilder.setChecksum(answer.getCheckSum());
+				_vmTemplateDao.update(template.getId(), templateDaoBuilder);
+			}*/
 		}
  	}
 
@@ -361,7 +431,11 @@ public class DownloadListener implements Listener {
 
 	public void setDownloadInactive(Status reason) {
 		downloadActive=false;
-		downloadMonitor.handleDownloadEvent(sserver, template, reason);
+		if (template != null){
+			downloadMonitor.handleDownloadEvent(sserver, template, reason);
+		}else {
+			downloadMonitor.handleDownloadEvent(sserver, volume, reason);
+		}
 	}
 
 	public void cancelTimeoutTask() {
diff --git a/server/src/com/cloud/storage/download/DownloadMonitor.java b/server/src/com/cloud/storage/download/DownloadMonitor.java
index cae6be2c1ff..982776b2d20 100644
--- a/server/src/com/cloud/storage/download/DownloadMonitor.java
+++ b/server/src/com/cloud/storage/download/DownloadMonitor.java
@@ -17,6 +17,7 @@ import java.util.Map;
 import com.cloud.exception.StorageUnavailableException;
 import com.cloud.host.HostVO;
 import com.cloud.storage.VMTemplateVO;
+import com.cloud.storage.VolumeVO;
 import com.cloud.storage.template.TemplateInfo;
 import com.cloud.utils.component.Manager;
 
@@ -42,4 +43,6 @@ public interface DownloadMonitor extends Manager{
 
     void addSystemVMTemplatesToHost(HostVO host, Map<String, TemplateInfo> templateInfos);
 
+	boolean downloadVolumeToStorage(VolumeVO volume, Long zoneId, String url, String checkSum);
+
 }
\ No newline at end of file
diff --git a/server/src/com/cloud/storage/download/DownloadMonitorImpl.java b/server/src/com/cloud/storage/download/DownloadMonitorImpl.java
index f5022132b30..1a76a2f0cf7 100755
--- a/server/src/com/cloud/storage/download/DownloadMonitorImpl.java
+++ b/server/src/com/cloud/storage/download/DownloadMonitorImpl.java
@@ -60,6 +60,8 @@ import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.SwiftVO;
 import com.cloud.storage.VMTemplateHostVO;
 import com.cloud.storage.VMTemplateStorageResourceAssoc;
+import com.cloud.storage.VolumeHostVO;
+import com.cloud.storage.VolumeVO;
 import com.cloud.storage.VMTemplateStorageResourceAssoc.Status;
 import com.cloud.storage.VMTemplateVO;
 import com.cloud.storage.VMTemplateZoneVO;
@@ -70,6 +72,8 @@ import com.cloud.storage.dao.VMTemplateHostDao;
 import com.cloud.storage.dao.VMTemplatePoolDao;
 import com.cloud.storage.dao.VMTemplateSwiftDao;
 import com.cloud.storage.dao.VMTemplateZoneDao;
+import com.cloud.storage.dao.VolumeDao;
+import com.cloud.storage.dao.VolumeHostDao;
 import com.cloud.storage.secondary.SecondaryStorageVmManager;
 import com.cloud.storage.swift.SwiftManager;
 import com.cloud.storage.template.TemplateConstants;
@@ -88,6 +92,8 @@ import com.cloud.vm.UserVmManager;
 import com.cloud.vm.VirtualMachine.State;
 import com.cloud.vm.dao.SecondaryStorageVmDao;
 
+import edu.emory.mathcs.backport.java.util.Collections;
+
 
 
 /**
@@ -111,6 +117,10 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
     @Inject
     SecondaryStorageVmDao _secStorageVmDao;
     @Inject
+    VolumeDao _volumeDao;
+    @Inject
+    VolumeHostDao _volumeHostDao;
+    @Inject
     AlertManager _alertMgr;
     @Inject
     protected SwiftManager _swiftMgr;
@@ -151,6 +161,7 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
 	Timer _timer;
 
 	final Map<VMTemplateHostVO, DownloadListener> _listenerMap = new ConcurrentHashMap<VMTemplateHostVO, DownloadListener>();
+	final Map<VolumeHostVO, DownloadListener> _listenerVolumeMap = new ConcurrentHashMap<VolumeHostVO, DownloadListener>();
 
 
 	public void send(Long hostId, Command cmd, Listener listener) throws AgentUnavailableException {
@@ -395,6 +406,67 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
 	    }
 	    return true;
 	}
+	
+	@Override
+	public boolean downloadVolumeToStorage(VolumeVO volume, Long zoneId, String url, String checkSum) {
+                                
+	    List<HostVO> ssHosts = _ssvmMgr.listAllTypesSecondaryStorageHostsInOneZone(zoneId);
+	    Collections.shuffle(ssHosts);
+	    HostVO ssHost = ssHosts.get(0);
+	    downloadVolumeToStorage(volume, ssHost, url, checkSum);
+	    return true;
+	}
+	
+	private void downloadVolumeToStorage(VolumeVO volume, HostVO sserver, String url, String checkSum) {
+		boolean downloadJobExists = false;
+        VolumeHostVO volumeHost = null;
+
+        volumeHost = _volumeHostDao.findByHostVolume(sserver.getId(), volume.getId());
+        if (volumeHost == null) {
+            volumeHost = new VolumeHostVO(sserver.getId(), volume.getId(), new Date(), 0, VMTemplateStorageResourceAssoc.Status.NOT_DOWNLOADED, null, null, "jobid0000", null, url, checkSum);
+            _volumeHostDao.persist(volumeHost);
+        } else if ((volumeHost.getJobId() != null) && (volumeHost.getJobId().length() > 2)) {
+            downloadJobExists = true;
+        }
+                
+        Long maxVolumeSizeInBytes = getMaxVolumeSizeInBytes();
+        String secUrl = sserver.getStorageUrl();
+		if(volumeHost != null) {
+		    start();
+			DownloadCommand dcmd = new DownloadCommand(secUrl, volume, maxVolumeSizeInBytes, checkSum, url);
+			dcmd.setProxy(getHttpProxy());
+	        if (downloadJobExists) {
+	            dcmd = new DownloadProgressCommand(dcmd, volumeHost.getJobId(), RequestType.GET_OR_RESTART);
+	        }
+			
+			HostVO ssvm = _ssvmMgr.pickSsvmHost(sserver);
+			if( ssvm == null ) {
+	             s_logger.warn("There is no secondary storage VM for secondary storage host " + sserver.getName());
+	             return;
+			}
+			DownloadListener dl = new DownloadListener(ssvm, sserver, volume, _timer, _volumeHostDao, volumeHost.getId(), this, dcmd, _volumeDao);
+			
+			if (downloadJobExists) {
+				dl.setCurrState(volumeHost.getDownloadState());
+	 		}
+            DownloadListener old = null;
+            synchronized (_listenerVolumeMap) {
+                old = _listenerVolumeMap.put(volumeHost, dl);
+            }
+            if( old != null ) {
+                old.abandon();
+            }
+
+			try {
+	            send(ssvm.getId(), dcmd, dl);
+            } catch (AgentUnavailableException e) {
+				s_logger.warn("Unable to start /resume download of template " + volume.getName() + " to " + sserver.getName(), e);
+				dl.setDisconnected();
+				dl.scheduleStatusCheck(RequestType.GET_OR_RESTART);
+            }
+		}
+	}
+
 
 	private void initiateTemplateDownload(Long templateId, HostVO ssHost) {
 		VMTemplateVO template = _templateDao.findById(templateId);
@@ -438,6 +510,37 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
         }
         txn.commit();
 	}
+
+	@DB
+	public void handleDownloadEvent(HostVO host, VolumeVO volume, Status dnldStatus) {
+		if ((dnldStatus == VMTemplateStorageResourceAssoc.Status.DOWNLOADED) || (dnldStatus==Status.ABANDONED)){
+			VolumeHostVO volumeHost = new VolumeHostVO(host.getId(), volume.getId());
+	        synchronized (_listenerVolumeMap) {
+	        	_listenerVolumeMap.remove(volumeHost);
+	        }
+		}
+		
+		VolumeHostVO volumeHost = _volumeHostDao.findByHostVolume(host.getId(), volume.getId());
+		
+		Transaction txn = Transaction.currentTxn();
+        txn.start();		
+
+        if (dnldStatus == Status.DOWNLOADED) {
+            long size = -1;
+            if(volumeHost!=null){
+                size = volumeHost.getPhysicalSize();
+            }
+            else{
+                s_logger.warn("Failed to get size for volume" + volume.getName());
+            }
+            String eventType = EventTypes.EVENT_VOLUME_CREATE;            
+            if(volume.getAccountId() != Account.ACCOUNT_ID_SYSTEM){
+               UsageEventVO usageEvent = new UsageEventVO(eventType, volume.getAccountId(), host.getDataCenterId(), volume.getId(), volume.getName(), null, 0l , size);
+               _usageEventDao.persist(usageEvent);
+            }
+        }
+        txn.commit();		
+	}
 	
 	@Override
     public void handleSysTemplateDownload(HostVO host) {
@@ -752,6 +855,14 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
 		}
 	}
 	
+	private Long getMaxVolumeSizeInBytes() {
+		try {
+			return Long.parseLong(_configDao.getValue("storage.max.volume.upload.size")) * 1024L * 1024L * 1024L;
+		} catch (NumberFormatException e) {
+			return null;
+		}
+	}
+	
 	private Proxy getHttpProxy() {
 		if (_proxy == null) {
 			return null;
@@ -763,7 +874,7 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
 		} catch (URISyntaxException e) {
 			return null;
 		}
-	}
+	}	
 	
 }
 	
diff --git a/server/src/com/cloud/vm/UserVmManagerImpl.java b/server/src/com/cloud/vm/UserVmManagerImpl.java
index 8436c746edb..8c5c55e1435 100755
--- a/server/src/com/cloud/vm/UserVmManagerImpl.java
+++ b/server/src/com/cloud/vm/UserVmManagerImpl.java
@@ -506,15 +506,6 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
             throw new InvalidParameterValueException("Please specify a valid data volume.");
         }
 
-        // Check that the volume is stored on shared storage
-        if (!(Volume.State.Allocated.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()))) {
-            throw new InvalidParameterValueException("Volume state must be in Allocated or Ready state");
-        }
-
         // Check that the volume is not currently attached to any VM
         if (volume.getInstanceId() != null) {
             throw new InvalidParameterValueException("Please specify a volume that is not attached to any VM.");
@@ -556,6 +547,16 @@ public class UserVmManagerImpl implements UserVmManager, UserVmService, Manager
 
         //permission check
         _accountMgr.checkAccess(caller, null, true, volume, vm);
+        
+
+        // Check that the volume is stored on shared storage
+        if (!(Volume.State.Allocated.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()))) {
+            throw new InvalidParameterValueException("Volume state must be in Allocated or Ready state");
+        }
 
         VolumeVO rootVolumeOfVm = null;
         List<VolumeVO> rootVolumesOfVm = _volsDao.findByInstanceAndType(vmId, Volume.Type.ROOT);
diff --git a/setup/db/create-schema.sql b/setup/db/create-schema.sql
index e7353507e33..a5f49ec6a8c 100755
--- a/setup/db/create-schema.sql
+++ b/setup/db/create-schema.sql
@@ -38,6 +38,7 @@ DROP TABLE IF EXISTS `cloud`.`pricing`;
 DROP TABLE IF EXISTS `cloud`.`sequence`;
 DROP TABLE IF EXISTS `cloud`.`user_vm`;
 DROP TABLE IF EXISTS `cloud`.`template_host_ref`;
+DROP TABLE IF EXISTS `cloud`.`volume_host_ref`;
 DROP TABLE IF EXISTS `cloud`.`upload`;
 DROP TABLE IF EXISTS `cloud`.`template_zone_ref`;
 DROP TABLE IF EXISTS `cloud`.`dc_vnet_alloc`;
@@ -1125,6 +1126,30 @@ CREATE TABLE  `cloud`.`template_host_ref` (
   INDEX `i_template_host_ref__template_id`(`template_id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
 
+CREATE TABLE  `cloud`.`volume_host_ref` (
+  `id` bigint unsigned NOT NULL auto_increment,
+  `host_id` bigint unsigned NOT NULL,
+  `volume_id` bigint unsigned NOT NULL,
+  `created` DATETIME NOT NULL,
+  `last_updated` DATETIME,
+  `job_id` varchar(255),
+  `download_pct` int(10) unsigned,
+  `size` bigint unsigned,
+  `physical_size` bigint unsigned DEFAULT 0,
+  `download_state` varchar(255),
+  `checksum` varchar(255) COMMENT 'checksum for the data disk',
+  `error_str` varchar(255),
+  `local_path` varchar(255),
+  `install_path` varchar(255),
+  `url` varchar(255),
+  `destroyed` tinyint(1) COMMENT 'indicates whether the volume_host entry was destroyed by the user or not',
+  PRIMARY KEY  (`id`),
+  CONSTRAINT `fk_volume_host_ref__host_id` FOREIGN KEY `fk_volume_host_ref__host_id` (`host_id`) REFERENCES `host` (`id`) ON DELETE CASCADE,
+  INDEX `i_volume_host_ref__host_id`(`host_id`),
+  CONSTRAINT `fk_volume_host_ref__volume_id` FOREIGN KEY `fk_volume_host_ref__volume_id` (`volume_id`) REFERENCES `volumes` (`id`),
+  INDEX `i_volume_host_ref__volume_id`(`volume_id`)
+) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
+
 CREATE TABLE  `cloud`.`template_swift_ref` (
   `id` bigint unsigned NOT NULL auto_increment,
   `swift_id` bigint unsigned NOT NULL,