diff --git a/api/src/com/cloud/api/commands/UploadVolumeCmd.java b/api/src/com/cloud/api/commands/UploadVolumeCmd.java
index 51588695e9a..fedff6ccc7b 100755
--- a/api/src/com/cloud/api/commands/UploadVolumeCmd.java
+++ b/api/src/com/cloud/api/commands/UploadVolumeCmd.java
@@ -30,9 +30,6 @@ public class UploadVolumeCmd extends BaseCmd {
     //////////////// API parameters /////////////////////
     /////////////////////////////////////////////////////
 
-    @Parameter(name=ApiConstants.DISPLAY_TEXT, type=CommandType.STRING, required=true, description="the display text of the volume. This is usually used for display purposes.", length=4096)
-    private String displayText;
-
     @Parameter(name=ApiConstants.FORMAT, type=CommandType.STRING, required=true, description="the format for the volume. Possible values include QCOW2, OVA, and VHD.")
     private String format;
 
@@ -63,10 +60,6 @@ public class UploadVolumeCmd extends BaseCmd {
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
 
-    public String getDisplayText() {
-        return displayText;
-    }
-
     public String getFormat() {
         return format;
     }
diff --git a/api/src/com/cloud/storage/Volume.java b/api/src/com/cloud/storage/Volume.java
index 73e69d53026..9b5f9afe3f6 100755
--- a/api/src/com/cloud/storage/Volume.java
+++ b/api/src/com/cloud/storage/Volume.java
@@ -62,7 +62,9 @@ public interface Volume extends ControlledEntity, BasedOn, StateObject<Volume.St
             s_fsm.addTransition(Allocated, Event.UploadRequested, Uploading);
             s_fsm.addTransition(Uploading, Event.UploadSucceeded, Uploaded);
             s_fsm.addTransition(Uploading, Event.OperationFailed, UploadError);
+            s_fsm.addTransition(Uploaded, Event.UploadSucceeded, Uploaded);
             s_fsm.addTransition(Uploaded, Event.CopyRequested, Creating);
+            s_fsm.addTransition(Uploaded, Event.DestroyRequested, Destroy);
             s_fsm.addTransition(Creating, Event.CopySucceeded, Ready);
             s_fsm.addTransition(Creating, Event.CopyFailed, Uploaded);
             s_fsm.addTransition(Ready, Event.DestroyRequested, Destroy);
diff --git a/core/src/com/cloud/storage/resource/NfsSecondaryStorageResource.java b/core/src/com/cloud/storage/resource/NfsSecondaryStorageResource.java
index 3108067abd6..03ce87ad75e 100755
--- a/core/src/com/cloud/storage/resource/NfsSecondaryStorageResource.java
+++ b/core/src/com/cloud/storage/resource/NfsSecondaryStorageResource.java
@@ -72,6 +72,7 @@ import com.cloud.agent.api.storage.ListVolumeCommand;
 import com.cloud.agent.api.storage.UploadCommand;
 import com.cloud.agent.api.storage.ssCommand;
 import com.cloud.agent.api.to.SwiftTO;
+import com.cloud.api.commands.DeleteVolumeCmd;
 import com.cloud.exception.InternalErrorException;
 import com.cloud.host.Host;
 import com.cloud.host.Host.Type;
@@ -144,7 +145,9 @@ public class NfsSecondaryStorageResource extends ServerResourceBase implements S
             return new CheckHealthAnswer((CheckHealthCommand)cmd, true);
         } else if (cmd instanceof DeleteTemplateCommand) {
         	return execute((DeleteTemplateCommand) cmd);
-        } else if (cmd instanceof ReadyCommand) {
+        } else if (cmd instanceof DeleteVolumeCommand) {
+        	return execute((DeleteVolumeCommand) cmd);
+        }else if (cmd instanceof ReadyCommand) {
             return new ReadyAnswer((ReadyCommand)cmd);
         } else if (cmd instanceof SecStorageFirewallCfgCommand){
         	return execute((SecStorageFirewallCfgCommand)cmd);
diff --git a/server/src/com/cloud/configuration/Config.java b/server/src/com/cloud/configuration/Config.java
index b8e567311aa..0c0d73eb8da 100755
--- a/server/src/com/cloud/configuration/Config.java
+++ b/server/src/com/cloud/configuration/Config.java
@@ -64,7 +64,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),
+	MaxUploadVolumeSize("Storage",  ManagementServer.class, Integer.class, "storage.max.volume.upload.size", "500", "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/storage/StorageManagerImpl.java b/server/src/com/cloud/storage/StorageManagerImpl.java
index 7c6f9eaa4a0..cfe716c1819 100755
--- a/server/src/com/cloud/storage/StorageManagerImpl.java
+++ b/server/src/com/cloud/storage/StorageManagerImpl.java
@@ -60,6 +60,7 @@ import com.cloud.agent.api.storage.CopyVolumeCommand;
 import com.cloud.agent.api.storage.CreateAnswer;
 import com.cloud.agent.api.storage.CreateCommand;
 import com.cloud.agent.api.storage.DeleteTemplateCommand;
+import com.cloud.agent.api.storage.DeleteVolumeCommand;
 import com.cloud.agent.api.storage.DestroyCommand;
 import com.cloud.agent.api.to.StorageFilerTO;
 import com.cloud.agent.api.to.VolumeTO;
@@ -751,6 +752,8 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         volume.setPoolId(destPool.getId());        
         volume.setPodId(destPool.getPodId());
         stateTransitTo(volume, Event.CopySucceeded); 
+        UsageEventVO usageEvent = new UsageEventVO(EventTypes.EVENT_VOLUME_CREATE, volume.getAccountId(), volume.getDataCenterId(), volume.getId(), volume.getName(), volume.getDiskOfferingId(), null, volume.getSize());
+        _usageEventDao.persist(usageEvent);
         _volumeHostDao.remove(volumeHostVO.getId());
     	txn.commit();
 		return volume;
@@ -1691,7 +1694,7 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
      */
     @Override
     @DB
-    @ActionEvent(eventType = EventTypes.EVENT_VOLUME_CREATE, eventDescription = "creating volume", create = true)
+    @ActionEvent(eventType = EventTypes.EVENT_VOLUME_UPLOAD, eventDescription = "creating volume", create = true)
     public VolumeVO uploadVolume(UploadVolumeCmd cmd) throws ResourceAllocationException{
     	Account caller = UserContext.current().getCaller();
         long ownerId = cmd.getEntityOwnerId();
@@ -1806,8 +1809,6 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         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);
         try {
 			stateTransitTo(volume, Event.UploadRequested);
 		} catch (NoTransitionException e) {
@@ -2353,7 +2354,39 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
                     s_logger.warn("problem cleaning up snapshots in secondary storage " + secondaryStorageHost, e2);
                 }
             }
+            
+            //CleanUp volumes on Secondary Storage.
+            for (HostVO secondaryStorageHost : secondaryStorageHosts) {
+                try {
+                    long hostId = secondaryStorageHost.getId();
+                    List<VolumeHostVO> destroyedVolumeHostVOs = _volumeHostDao.listDestroyed(hostId);
+                    s_logger.debug("Secondary storage garbage collector found " + destroyedVolumeHostVOs.size() + " templates to cleanup on secondary storage host: "
+                            + secondaryStorageHost.getName());
+                    for (VolumeHostVO destroyedVolumeHostVO : destroyedVolumeHostVOs) {
+                        if (s_logger.isDebugEnabled()) {
+                            s_logger.debug("Deleting volume host: " + destroyedVolumeHostVO);
+                        }
 
+                        String installPath = destroyedVolumeHostVO.getInstallPath();
+
+                        if (installPath != null) {
+                            Answer answer = _agentMgr.sendToSecStorage(secondaryStorageHost, new DeleteVolumeCommand(secondaryStorageHost.getStorageUrl(), destroyedVolumeHostVO.getInstallPath()));
+
+                            if (answer == null || !answer.getResult()) {
+                                s_logger.debug("Failed to delete " + destroyedVolumeHostVO + " due to " + ((answer == null) ? "answer is null" : answer.getDetails()));
+                            } else {
+                                _volumeHostDao.remove(destroyedVolumeHostVO.getId());
+                                s_logger.debug("Deleted volume at: " + destroyedVolumeHostVO.getInstallPath());
+                            }
+                        } else {
+                            _volumeHostDao.remove(destroyedVolumeHostVO.getId());
+                        }
+                    }
+                
+                }catch (Exception e2) {
+                    s_logger.warn("problem cleaning up volumes in secondary storage " + secondaryStorageHost, e2);
+                }
+            }
         } catch (Exception e3) {
             s_logger.warn("problem cleaning up secondary storage ", e3);
         }
@@ -3330,6 +3363,22 @@ public class StorageManagerImpl implements StorageManager, Manager, ClusterManag
         if (s_logger.isDebugEnabled()) {
             s_logger.debug("Expunging " + vol);
         }
+        
+        //Find out if the volume is present on secondary storage
+        VolumeHostVO volumeHost = _volumeHostDao.findByVolumeId(vol.getId());
+        if(volumeHost != null){
+        	HostVO ssHost = _hostDao.findById(volumeHost.getHostId());
+        	DeleteVolumeCommand dtCommand = new DeleteVolumeCommand(ssHost.getStorageUrl(), volumeHost.getInstallPath());            
+        	Answer answer = _agentMgr.sendToSecStorage(ssHost, dtCommand);
+             if (answer == null || !answer.getResult()) {
+                 s_logger.debug("Failed to delete " + volumeHost + " due to " + ((answer == null) ? "answer is null" : answer.getDetails()));
+                 return;
+             }
+            _volumeHostDao.remove(volumeHost.getId());
+            _volumeDao.remove(vol.getId());
+            return;             
+        }
+        
         String vmName = null;
         if (vol.getVolumeType() == Type.ROOT && vol.getInstanceId() != null) {
             VirtualMachine vm = _vmInstanceDao.findByIdIncludingRemoved(vol.getInstanceId());
diff --git a/server/src/com/cloud/storage/dao/VolumeHostDao.java b/server/src/com/cloud/storage/dao/VolumeHostDao.java
index 96efd126595..f4d667ad021 100755
--- a/server/src/com/cloud/storage/dao/VolumeHostDao.java
+++ b/server/src/com/cloud/storage/dao/VolumeHostDao.java
@@ -14,4 +14,6 @@ public interface VolumeHostDao extends GenericDao<VolumeHostVO, Long> {
 
 	List<VolumeHostVO> listBySecStorage(long sserverId);
 
+	List<VolumeHostVO> listDestroyed(long hostId);
+
 }
diff --git a/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java b/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java
index fb1ad0afbad..5c205fd0561 100755
--- a/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java
+++ b/server/src/com/cloud/storage/dao/VolumeHostDaoImpl.java
@@ -6,6 +6,7 @@ import java.util.List;
 import javax.ejb.Local;
 
 import com.cloud.host.HostVO;
+import com.cloud.storage.VMTemplateHostVO;
 import com.cloud.storage.VolumeHostVO;
 import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.SearchBuilder;
@@ -16,6 +17,7 @@ public class VolumeHostDaoImpl extends GenericDaoBase<VolumeHostVO, Long> implem
 	protected final SearchBuilder<VolumeHostVO> HostVolumeSearch;
 	protected final SearchBuilder<VolumeHostVO> VolumeSearch;
 	protected final SearchBuilder<VolumeHostVO> HostSearch;
+	protected final SearchBuilder<VolumeHostVO> HostDestroyedSearch;
 	
 	VolumeHostDaoImpl(){
 		HostVolumeSearch = createSearchBuilder();
@@ -33,6 +35,11 @@ public class VolumeHostDaoImpl extends GenericDaoBase<VolumeHostVO, Long> implem
 		VolumeSearch.and("volume_id", VolumeSearch.entity().getVolumeId(), SearchCriteria.Op.EQ);
 		VolumeSearch.and("destroyed", VolumeSearch.entity().getDestroyed(), SearchCriteria.Op.EQ);
 		VolumeSearch.done();
+		
+		HostDestroyedSearch = createSearchBuilder();
+		HostDestroyedSearch.and("host_id", HostDestroyedSearch.entity().getHostId(), SearchCriteria.Op.EQ);
+		HostDestroyedSearch.and("destroyed", HostDestroyedSearch.entity().getDestroyed(), SearchCriteria.Op.EQ);
+		HostDestroyedSearch.done();	
 	}
     
     
@@ -63,5 +70,13 @@ public class VolumeHostDaoImpl extends GenericDaoBase<VolumeHostVO, Long> implem
 	    sc.setParameters("destroyed", false);
 	    return listAll();
 	}
+	
+	@Override
+	public List<VolumeHostVO> listDestroyed(long hostId){
+		SearchCriteria<VolumeHostVO> sc = HostDestroyedSearch.create();
+		sc.setParameters("host_id", hostId);
+		sc.setParameters("destroyed", true);
+		return listIncludingRemovedBy(sc);
+	}
 
 }
diff --git a/server/src/com/cloud/storage/download/DownloadMonitorImpl.java b/server/src/com/cloud/storage/download/DownloadMonitorImpl.java
index c880f8f5392..2287179bf48 100755
--- a/server/src/com/cloud/storage/download/DownloadMonitorImpl.java
+++ b/server/src/com/cloud/storage/download/DownloadMonitorImpl.java
@@ -439,12 +439,7 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
             downloadJobExists = true;
         }
         
-        try {
-			_storageMgr.stateTransitTo(volume, Event.UploadRequested);
-		} catch (NoTransitionException e1) {
-			// TODO Auto-generated catch block
-			e1.printStackTrace();
-		}
+
         Long maxVolumeSizeInBytes = getMaxVolumeSizeInBytes();
         String secUrl = sserver.getStorageUrl();
 		if(volumeHost != null) {
@@ -478,7 +473,7 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
 			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);
+				s_logger.warn("Unable to start /resume download of volume " + volume.getName() + " to " + sserver.getName(), e);
 				dl.setDisconnected();
 				dl.scheduleStatusCheck(RequestType.GET_OR_RESTART);
             }
@@ -551,7 +546,7 @@ public class DownloadMonitorImpl implements  DownloadMonitor {
             else{
                 s_logger.warn("Failed to get size for volume" + volume.getName());
             }
-            String eventType = EventTypes.EVENT_VOLUME_CREATE;            
+            String eventType = EventTypes.EVENT_VOLUME_UPLOAD;            
             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);