fix CLOUDSTACK-3591 add usage recording for VM snapshots

Conflicts:

	plugins/hypervisors/xen/src/com/cloud/hypervisor/xen/resource/CitrixResourceBase.java
	server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java
	setup/db/db/schema-410to420.sql

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

@@ -41,6 +41,7 @@ public class VolumeTO implements InternalIdentity {
     private Long bytesWriteRate;
     private Long iopsReadRate;
     private Long iopsWriteRate;
+    private Long chainSize;
 
     public VolumeTO(long id, Volume.Type type, StoragePoolType poolType, String poolUuid, String name, String mountPoint, String path, long size, String chainInfo) {
         this.id = id;
@@ -77,6 +78,7 @@ public class VolumeTO implements InternalIdentity {
         this.storagePoolUuid = pool.getUuid();
         this.mountPoint = volume.getFolder();
         this.chainInfo = volume.getChainInfo();
+        this.chainSize = volume.getVmSnapshotChainSize();
         if (volume.getDeviceId() != null)
             this.deviceId = volume.getDeviceId();
     }
@@ -170,4 +172,11 @@ public class VolumeTO implements InternalIdentity {
         return iopsWriteRate;
     }
 
+    public Long getChainSize() {
+        return chainSize;
+    }
+
+    public void setChainSize(Long chainSize) {
+        this.chainSize = chainSize;
+    }
 }

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

@@ -184,4 +184,5 @@ public interface Volume extends ControlledEntity, Identity, InternalIdentity, Ba
 	 */
 	void setReservationId(String reserv);
 	Storage.ImageFormat getFormat();
+	Long getVmSnapshotChainSize();
 }

api/src/org/apache/cloudstack/usage/UsageTypes.java

@@ -40,6 +40,7 @@ public class UsageTypes {
     public static final int VM_DISK_IO_WRITE = 22;
     public static final int VM_DISK_BYTES_READ = 23;
     public static final int VM_DISK_BYTES_WRITE = 24;
+    public static final int VM_SNAPSHOT = 25;
     
     public static List<UsageTypeResponse> listUsageTypes(){
     	List<UsageTypeResponse> responseList = new ArrayList<UsageTypeResponse>();
@@ -61,6 +62,7 @@ public class UsageTypes {
         responseList.add(new UsageTypeResponse(VM_DISK_IO_WRITE, "VM Disk usage(I/O Write)"));
         responseList.add(new UsageTypeResponse(VM_DISK_BYTES_READ, "VM Disk usage(Bytes Read)"));
         responseList.add(new UsageTypeResponse(VM_DISK_BYTES_WRITE, "VM Disk usage(Bytes Write)"));
+        responseList.add(new UsageTypeResponse(VM_SNAPSHOT, "VM Snapshot storage usage"));
     	return responseList;
     }
 }

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

@@ -150,6 +150,9 @@ public class VolumeVO implements Volume {
     @Column(name = "iscsi_name")
     private String _iScsiName;
 
+    @Column(name = "vm_snapshot_chain_size")
+    private Long vmSnapshotChainSize;
+    
     @Transient
     // @Column(name="reservation")
     String reservationId;
@@ -550,4 +553,12 @@ public class VolumeVO implements Volume {
     public void setFormat(Storage.ImageFormat format) {
         this.format = format;
     }
+    
+    public void setVmSnapshotChainSize(Long vmSnapshotChainSize){
+        this.vmSnapshotChainSize = vmSnapshotChainSize;
+    }
+
+    public Long getVmSnapshotChainSize(){
+        return this.vmSnapshotChainSize;
+    }
 }

engine/schema/src/com/cloud/usage/UsageVMSnapshotVO.java

@@ -0,0 +1,122 @@
+// 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.usage;
+
+import java.util.Date;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Table;
+import javax.persistence.Temporal;
+import javax.persistence.TemporalType;
+
+import org.apache.cloudstack.api.InternalIdentity;
+
+@Entity
+@Table(name="usage_vmsnapshot")
+public class UsageVMSnapshotVO implements InternalIdentity {
+    
+    @Column(name="id") // volumeId
+    private long id;
+    
+    @Column(name="zone_id")
+    private long zoneId;
+    
+    @Column(name="account_id")
+    private long accountId;
+
+    @Column(name="domain_id")
+    private long domainId;
+
+    @Column(name="vm_id")
+    private long vmId;
+    
+    @Column(name="disk_offering_id")
+    private Long diskOfferingId;
+    
+    @Column(name="size")
+    private long size;
+
+    @Column(name="created")
+    @Temporal(value=TemporalType.TIMESTAMP)
+    private Date created = null;
+    
+    @Column(name="processed")
+    @Temporal(value=TemporalType.TIMESTAMP)
+    private Date processed;
+
+    protected UsageVMSnapshotVO() {
+    }
+
+    public UsageVMSnapshotVO(long id, long zoneId, long accountId, long domainId, 
+            long vmId, Long diskOfferingId, long size, Date created, Date processed) {
+        this.zoneId = zoneId;
+        this.accountId = accountId;
+        this.domainId = domainId;
+        this.diskOfferingId = diskOfferingId;
+        this.id = id;
+        this.size = size;
+        this.created = created;
+        this.vmId = vmId;
+        this.processed = processed;
+    }
+
+    public long getZoneId() {
+        return zoneId;
+    }
+    
+    public long getAccountId() {
+        return accountId;
+    }
+
+    public long getDomainId() {
+        return domainId;
+    }
+
+    public Long getDiskOfferingId() {
+        return diskOfferingId;
+    }
+    
+    public long getSize() {
+        return size;
+    }
+
+    public Date getProcessed() {
+        return processed;
+    }
+
+    public void setProcessed(Date processed) {
+        this.processed = processed;
+    }
+
+    public Date getCreated() {
+        return created;
+    }
+
+    public void setCreated(Date created) {
+        this.created = created;
+    }
+
+    public long getVmId() {
+        return vmId;
+    }
+    
+    public long getId(){
+        return this.id;
+    }
+
+}

engine/schema/src/com/cloud/usage/dao/UsageVMSnapshotDao.java

@@ -0,0 +1,29 @@
+// 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.usage.dao;
+
+import java.util.Date;
+import java.util.List;
+
+import com.cloud.usage.UsageVMSnapshotVO;
+import com.cloud.utils.db.GenericDao;
+
+public interface UsageVMSnapshotDao extends GenericDao<UsageVMSnapshotVO, Long> {
+    public void update(UsageVMSnapshotVO usage);
+    public List<UsageVMSnapshotVO> getUsageRecords(Long accountId, Long domainId, Date startDate, Date endDate);
+    UsageVMSnapshotVO getPreviousUsageRecord(UsageVMSnapshotVO rec);
+}

engine/schema/src/com/cloud/usage/dao/UsageVMSnapshotDaoImpl.java

@@ -0,0 +1,182 @@
+// 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.usage.dao;
+
+import com.cloud.usage.UsageVMSnapshotVO;
+import com.cloud.utils.DateUtil;
+import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.Transaction;
+import org.apache.log4j.Logger;
+import org.springframework.stereotype.Component;
+
+import javax.ejb.Local;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.TimeZone;
+@Component
+@Local(value={UsageVMSnapshotDao.class})
+public class UsageVMSnapshotDaoImpl extends GenericDaoBase<UsageVMSnapshotVO, Long> implements UsageVMSnapshotDao{
+    public static final Logger s_logger = Logger.getLogger(UsageVMSnapshotDaoImpl.class.getName());
+    protected static final String GET_USAGE_RECORDS_BY_ACCOUNT = 
+            "SELECT id, zone_id, account_id, domain_id, vm_id, disk_offering_id, size, created, processed " +
+            " FROM usage_vmsnapshot" +
+            " WHERE account_id = ? " +
+            " AND ( (created BETWEEN ? AND ?) OR " +
+            "      (created < ? AND processed is NULL) ) ORDER BY created asc";
+    protected static final String UPDATE_DELETED = 
+            "UPDATE usage_vmsnapshot SET processed = ? WHERE account_id = ? AND id = ? and vm_id = ?  and created = ?";
+    
+    protected static final String PREVIOUS_QUERY = 
+            "SELECT id, zone_id, account_id, domain_id, vm_id, disk_offering_id,size, created, processed " +
+            "FROM usage_vmsnapshot " +
+            "WHERE account_id = ? AND id = ? AND vm_id = ? AND created < ? AND processed IS NULL " +
+            "ORDER BY created desc limit 1";
+    
+    public void update(UsageVMSnapshotVO usage) {
+        Transaction txn = Transaction.open(Transaction.USAGE_DB);
+        PreparedStatement pstmt = null;
+        try {
+            txn.start();
+            pstmt = txn.prepareAutoCloseStatement(UPDATE_DELETED);
+            pstmt.setString(1, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), usage.getProcessed()));
+            pstmt.setLong(2, usage.getAccountId());
+            pstmt.setLong(3, usage.getId());
+            pstmt.setLong(4, usage.getVmId());
+            pstmt.setString(5, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), usage.getCreated()));
+            pstmt.executeUpdate();
+            txn.commit();
+        } catch (Exception e) {
+            txn.rollback();
+            s_logger.warn("Error updating UsageVMSnapshotVO", e);
+        } finally {
+            txn.close();
+        }
+    }
+    
+    public List<UsageVMSnapshotVO> getUsageRecords(Long accountId, Long domainId, 
+            Date startDate, Date endDate) {
+        List<UsageVMSnapshotVO> usageRecords = new ArrayList<UsageVMSnapshotVO>();
+
+        String sql = GET_USAGE_RECORDS_BY_ACCOUNT;
+        Transaction txn = Transaction.open(Transaction.USAGE_DB);
+        PreparedStatement pstmt = null;
+
+        try {
+            int i = 1;
+            pstmt = txn.prepareAutoCloseStatement(sql);
+            pstmt.setLong(i++, accountId);
+            pstmt.setString(i++, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), startDate));
+            pstmt.setString(i++, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), endDate));
+            pstmt.setString(i++, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), startDate));
+
+            ResultSet rs = pstmt.executeQuery();
+            while (rs.next()) {
+                //id, zone_id, account_id, domain_iVMSnapshotVOd, vm_id, disk_offering_id, size, created, processed
+                Long vId = Long.valueOf(rs.getLong(1));
+                Long zoneId = Long.valueOf(rs.getLong(2));
+                Long acctId = Long.valueOf(rs.getLong(3));
+                Long dId = Long.valueOf(rs.getLong(4));
+                Long vmId = Long.valueOf(rs.getLong(5));
+                Long doId = Long.valueOf(rs.getLong(6));
+                if(doId == 0){
+                    doId = null;
+                }
+                Long size = Long.valueOf(rs.getLong(7));
+                Date createdDate = null;
+                Date processDate = null;
+                String createdTS = rs.getString(8);
+                String processed = rs.getString(9);
+                
+
+                if (createdTS != null) {
+                    createdDate = DateUtil.parseDateString(s_gmtTimeZone, createdTS);
+                }
+                if (processed != null) {
+                    processDate = DateUtil.parseDateString(s_gmtTimeZone, processed);
+                }
+                usageRecords.add(new UsageVMSnapshotVO(vId, zoneId, acctId, dId, vmId, 
+                        doId, size, createdDate, processDate));
+            }
+        } catch (Exception e) {
+            txn.rollback();
+            s_logger.warn("Error getting usage records", e);
+        } finally {
+            txn.close();
+        }
+
+        return usageRecords;
+    }
+
+    @Override
+    public UsageVMSnapshotVO getPreviousUsageRecord(UsageVMSnapshotVO rec) {
+        List<UsageVMSnapshotVO> usageRecords = new ArrayList<UsageVMSnapshotVO>();
+
+        String sql = PREVIOUS_QUERY;
+        Transaction txn = Transaction.open(Transaction.USAGE_DB);
+        PreparedStatement pstmt = null;
+        try {
+            int i = 1;
+            pstmt = txn.prepareAutoCloseStatement(sql);
+            pstmt.setLong(i++, rec.getAccountId());
+            pstmt.setLong(i++, rec.getId());
+            pstmt.setLong(i++, rec.getVmId());
+            pstmt.setString(i++, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), rec.getCreated()));
+
+            ResultSet rs = pstmt.executeQuery();
+            while (rs.next()) {
+                //id, zone_id, account_id, domain_iVMSnapshotVOd, vm_id, disk_offering_id, size, created, processed
+                Long vId = Long.valueOf(rs.getLong(1));
+                Long zoneId = Long.valueOf(rs.getLong(2));
+                Long acctId = Long.valueOf(rs.getLong(3));
+                Long dId = Long.valueOf(rs.getLong(4));
+                Long vmId = Long.valueOf(rs.getLong(5));
+                Long doId = Long.valueOf(rs.getLong(6));
+                if(doId == 0){
+                    doId = null;
+                }
+                Long size = Long.valueOf(rs.getLong(7));
+                Date createdDate = null;
+                Date processDate = null;
+                String createdTS = rs.getString(8);
+                String processed = rs.getString(9);
+                
+
+                if (createdTS != null) {
+                    createdDate = DateUtil.parseDateString(s_gmtTimeZone, createdTS);
+                }
+                if (processed != null) {
+                    processDate = DateUtil.parseDateString(s_gmtTimeZone, processed);
+                }
+                usageRecords.add(new UsageVMSnapshotVO(vId, zoneId, acctId, dId, vmId, 
+                        doId, size, createdDate, processDate));
+            }
+        } catch (Exception e) {
+            txn.rollback();
+            s_logger.warn("Error getting usage records", e);
+        } finally {
+            txn.close();
+        }
+
+        if(usageRecords.size() > 0)
+            return usageRecords.get(0);
+        return null;
+    }
+}

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

@@ -613,4 +613,9 @@ public class VolumeObject implements VolumeInfo {
         }
         return true;
     }
+
+    @Override
+    public Long getVmSnapshotChainSize() {
+        return this.volumeVO.getVmSnapshotChainSize();
+    }
 }

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

@@ -51,14 +51,18 @@ import com.cloud.agent.api.RevertToVMSnapshotCommand;
 import com.cloud.hypervisor.vmware.mo.CustomFieldConstants;
 import com.cloud.hypervisor.vmware.mo.DatacenterMO;
 import com.cloud.hypervisor.vmware.mo.DatastoreMO;
+import com.cloud.hypervisor.vmware.mo.HostDatastoreBrowserMO;
 import com.cloud.hypervisor.vmware.mo.HostMO;
 import com.cloud.hypervisor.vmware.mo.HypervisorHostHelper;
+import com.cloud.hypervisor.vmware.mo.SnapshotDescriptor;
+import com.cloud.hypervisor.vmware.mo.SnapshotDescriptor.SnapshotInfo;
 import com.cloud.hypervisor.vmware.mo.TaskMO;
 import com.cloud.hypervisor.vmware.mo.VirtualMachineMO;
 import com.cloud.hypervisor.vmware.mo.VmwareHypervisorHost;
 import com.cloud.hypervisor.vmware.util.VmwareContext;
 import com.cloud.hypervisor.vmware.util.VmwareHelper;
 import com.cloud.storage.JavaStorageLayer;
+import com.cloud.storage.Volume;
 import com.cloud.storage.Storage.ImageFormat;
 import com.cloud.storage.StorageLayer;
 import com.cloud.storage.template.VmdkProcessor;
@@ -70,6 +74,10 @@ import com.cloud.utils.script.Script;
 import com.cloud.utils.exception.CloudRuntimeException;
 import com.cloud.vm.VirtualMachine;
 import com.cloud.vm.snapshot.VMSnapshot;
+import com.vmware.vim25.FileInfo;
+import com.vmware.vim25.FileQueryFlags;
+import com.vmware.vim25.HostDatastoreBrowserSearchResults;
+import com.vmware.vim25.HostDatastoreBrowserSearchSpec;
 import com.vmware.vim25.ManagedObjectReference;
 import com.vmware.vim25.TaskEvent;
 import com.vmware.vim25.TaskInfo;
@@ -1209,6 +1217,55 @@ public class VmwareStorageManagerImpl implements VmwareStorageManager {
         return "snapshots/" + accountId + "/" + volumeId;
     }
 
+    private long getVMSnapshotChainSize(VmwareContext context, VmwareHypervisorHost hyperHost, 
+            String fileName, String poolUuid, String exceptFileName) 
+            throws Exception{
+        long size = 0;
+        ManagedObjectReference morDs = HypervisorHostHelper.findDatastoreWithBackwardsCompatibility(hyperHost, poolUuid);
+        DatastoreMO dsMo = new DatastoreMO(context, morDs);
+        HostDatastoreBrowserMO browserMo = dsMo.getHostDatastoreBrowserMO();
+        String datastorePath = "[" + dsMo.getName() + "]";
+        HostDatastoreBrowserSearchSpec searchSpec = new HostDatastoreBrowserSearchSpec();
+        FileQueryFlags fqf = new FileQueryFlags();
+        fqf.setFileSize(true);
+        fqf.setFileOwner(true);
+        fqf.setModification(true);
+        searchSpec.setDetails(fqf);
+        searchSpec.setSearchCaseInsensitive(false);
+        searchSpec.getMatchPattern().add(fileName);
+        ArrayList<HostDatastoreBrowserSearchResults> results = browserMo.
+                searchDatastoreSubFolders(datastorePath, searchSpec);
+        for(HostDatastoreBrowserSearchResults result : results){
+            if (result != null) {
+                List<FileInfo> info = result.getFile();
+                for (FileInfo fi : info) {
+                    if(exceptFileName != null && fi.getPath().contains(exceptFileName))
+                        continue;
+                    else
+                        size = size + fi.getFileSize();
+                }
+            }
+        }
+        return size;
+    }
+
+    private String extractSnapshotBaseFileName(String input) {
+        if(input == null)
+            return null;
+        String result  = input;
+        if (result.endsWith(".vmdk")){ // get rid of vmdk file extension
+            result = result.substring(0, result.length() - (".vmdk").length());
+        }
+        if(result.split("-").length == 1) // e.g 4da6dcbd412c47b59f96c7ff6dbd7216.vmdk
+            return result;
+        if(result.split("-").length > 2) // e.g ROOT-5-4.vmdk, ROOT-5-4-000001.vmdk
+            return result.split("-")[0] + "-" + result.split("-")[1];
+        if(result.split("-").length == 2) // e.g 4da6dcbd412c47b59f96c7ff6dbd7216-000001.vmdk
+            return result.split("-")[0];
+        else
+            return result;
+    }
+    
     @Override
     public CreateVMSnapshotAnswer execute(VmwareHostService hostService, CreateVMSnapshotCommand cmd) {
         List<VolumeTO> volumeTOs = cmd.getVolumeTOs();
@@ -1250,24 +1307,30 @@ public class VmwareStorageManagerImpl implements VmwareStorageManager {
                 // find VM disk file path after creating snapshot
                 VirtualDisk[] vdisks = vmMo.getAllDiskDevice();
                 for (int i = 0; i < vdisks.length; i ++){
-                    @SuppressWarnings("deprecation")
                     List<Pair<String, ManagedObjectReference>> vmdkFiles = vmMo.getDiskDatastorePathChain(vdisks[i], false);
                     for(Pair<String, ManagedObjectReference> fileItem : vmdkFiles) {
                         String vmdkName = fileItem.first().split(" ")[1];
-                        if ( vmdkName.endsWith(".vmdk")){
+                        if (vmdkName.endsWith(".vmdk")){
                             vmdkName = vmdkName.substring(0, vmdkName.length() - (".vmdk").length());
                         }
-                        String[] s = vmdkName.split("-");
+                        String baseName = extractSnapshotBaseFileName(vmdkName);
-                        mapNewDisk.put(s[0], vmdkName);
+                        mapNewDisk.put(baseName, vmdkName);
                     }
                 }
-
-                // update volume path using maps
                 for (VolumeTO volumeTO : volumeTOs) {
-                    String parentUUID = volumeTO.getPath();
+                    String baseName = extractSnapshotBaseFileName(volumeTO.getPath());
-                    String[] s = parentUUID.split("-");
+                    String newPath = mapNewDisk.get(baseName);
-                    String key = s[0];
+                    // get volume's chain size for this VM snapshot, exclude current volume vdisk
-                    volumeTO.setPath(mapNewDisk.get(key));
+                    long size = getVMSnapshotChainSize(context,hyperHost,baseName + "*.vmdk",
+                            volumeTO.getPoolUuid(), newPath);
+                    
+                    if(volumeTO.getType()== Volume.Type.ROOT){
+                        // add memory snapshot size
+                        size = size + getVMSnapshotChainSize(context,hyperHost,cmd.getVmName()+"*.vmsn",volumeTO.getPoolUuid(),null);
+                    }
+                    
+                    volumeTO.setChainSize(size);
+                    volumeTO.setPath(newPath);
                 }
                 return new CreateVMSnapshotAnswer(cmd, cmd.getTarget(), volumeTOs);
             }
@@ -1322,16 +1385,21 @@ public class VmwareStorageManagerImpl implements VmwareStorageManager {
                         if (vmdkName.endsWith(".vmdk")) {
                             vmdkName = vmdkName.substring(0, vmdkName.length() - (".vmdk").length());
                         }
-                        String[] s = vmdkName.split("-");
+                        String baseName = extractSnapshotBaseFileName(vmdkName);
-                        mapNewDisk.put(s[0], vmdkName);
+                        mapNewDisk.put(baseName, vmdkName);
                     }
                 }
                 for (VolumeTO volumeTo : listVolumeTo) {
-                    String key = null;
+                    String baseName = extractSnapshotBaseFileName(volumeTo.getPath());
-                    String parentUUID = volumeTo.getPath();
+                    String newPath = mapNewDisk.get(baseName);
-                    String[] s = parentUUID.split("-");
+                    long size = getVMSnapshotChainSize(context,hyperHost,
-                    key = s[0];
+                            baseName + "*.vmdk", volumeTo.getPoolUuid(), newPath);
-                    volumeTo.setPath(mapNewDisk.get(key));
+                    if(volumeTo.getType()== Volume.Type.ROOT){
+                        // add memory snapshot size
+                        size = size + getVMSnapshotChainSize(context,hyperHost,cmd.getVmName()+"*.vmsn",volumeTo.getPoolUuid(),null);
+                    }                    
+                    volumeTo.setChainSize(size);
+                    volumeTo.setPath(newPath);
                 }
                 return new DeleteVMSnapshotAnswer(cmd, listVolumeTo);
             }

plugins/hypervisors/xen/src/com/cloud/hypervisor/xen/resource/CitrixResourceBase.java

@@ -6668,6 +6668,60 @@ public abstract class CitrixResourceBase implements ServerResource, HypervisorRe
 
     }
 
+    private long getVMSnapshotChainSize(Connection conn, VolumeTO volumeTo, String vmName) 
+            throws BadServerResponse, XenAPIException, XmlRpcException {
+        Set<VDI> allvolumeVDIs = VDI.getByNameLabel(conn, volumeTo.getName());
+        long size = 0;
+        for (VDI vdi : allvolumeVDIs) {
+            try {
+                if (vdi.getIsASnapshot(conn)
+                        && vdi.getSmConfig(conn).get("vhd-parent") != null) {
+                    String parentUuid = vdi.getSmConfig(conn).get("vhd-parent");
+                    VDI parentVDI = VDI.getByUuid(conn, parentUuid);
+                    // add size of snapshot vdi node, usually this only contains meta data
+                    size = size + vdi.getPhysicalUtilisation(conn);
+                    // add size of snapshot vdi parent, this contains data
+                    if (parentVDI != null)
+                        size = size
+                                + parentVDI.getPhysicalUtilisation(conn)
+                                        .longValue();
+                }
+            } catch (Exception e) {
+                s_logger.debug("Exception occurs when calculate "
+                        + "snapshot capacity for volumes: " + e.getMessage());
+                continue;
+            }
+        }
+        if (volumeTo.getType() == Volume.Type.ROOT) {
+            Map<VM, VM.Record> allVMs = VM.getAllRecords(conn);
+            // add size of memory snapshot vdi
+            if (allVMs.size() > 0) {
+                for (VM vmr : allVMs.keySet()) {
+                    try {
+                        String vName = vmr.getNameLabel(conn);
+                        if (vName != null && vName.contains(vmName)
+                                && vmr.getIsASnapshot(conn)) {
+
+                            VDI memoryVDI = vmr.getSuspendVDI(conn);
+                            size = size
+                                    + memoryVDI.getParent(conn)
+                                            .getPhysicalUtilisation(conn);
+                            size = size
+                                    + memoryVDI.getPhysicalUtilisation(conn);
+                        }
+                    } catch (Exception e) {
+                        s_logger.debug("Exception occurs when calculate "
+                                + "snapshot capacity for memory: "
+                                + e.getMessage());
+                        continue;
+                    }
+                }
+            }
+        }
+        return size;
+    }
+    
+
     protected Answer execute(final CreateVMSnapshotCommand cmd) {
         String vmName = cmd.getVmName();
         String vmSnapshotName = cmd.getTarget().getSnapshotName();
@@ -6745,7 +6799,17 @@ public abstract class CitrixResourceBase implements ServerResource, HypervisorRe
             // extract VM snapshot ref from result
             String ref = result.substring("<value>".length(), result.length() - "</value>".length());
             vmSnapshot = Types.toVM(ref);
+            try {
+                Thread.sleep(5000);
+            } catch (final InterruptedException ex) {
 
+            }
+            // calculate used capacity for this VM snapshot
+            for (VolumeTO volumeTo : cmd.getVolumeTOs()){
+                long size = getVMSnapshotChainSize(conn,volumeTo,cmd.getVmName());
+                volumeTo.setChainSize(size);
+            }
+            
             success = true;
             return new CreateVMSnapshotAnswer(cmd, cmd.getTarget(), cmd.getVolumeTOs());
         } catch (Exception e) {
@@ -6862,6 +6926,18 @@ public abstract class CitrixResourceBase implements ServerResource, HypervisorRe
             for (VDI vdi : vdiList) {
                 vdi.destroy(conn);
             }
+
+            try {
+                Thread.sleep(5000);
+            } catch (final InterruptedException ex) {
+
+            }
+            // re-calculate used capacify for this VM snapshot
+            for (VolumeTO volumeTo : cmd.getVolumeTOs()){
+                long size = getVMSnapshotChainSize(conn,volumeTo,cmd.getVmName());
+                volumeTo.setChainSize(size);
+            }
+            
             return new DeleteVMSnapshotAnswer(cmd, cmd.getVolumeTOs());
         } catch (Exception e) {
             s_logger.warn("Catch Exception: " + e.getClass().toString()

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

@@ -3413,6 +3413,13 @@ public class ApiResponseHelper implements ResponseGenerator {
 			//Security Group Id
             SecurityGroupVO sg = _entityMgr.findById(SecurityGroupVO.class, usageRecord.getUsageId().toString());
 			usageRecResponse.setUsageId(sg.getUuid());
+		} else if(usageRecord.getUsageType() == UsageTypes.VM_SNAPSHOT){
+		    VMInstanceVO vm = _entityMgr.findById(VMInstanceVO.class, usageRecord.getVmInstanceId().toString());
+		    usageRecResponse.setVmName(vm.getInstanceName());
+		    usageRecResponse.setUsageId(vm.getUuid());
+	        usageRecResponse.setSize(usageRecord.getSize());
+	        if(usageRecord.getOfferingId() != null)
+	            usageRecResponse.setOfferingId(usageRecord.getOfferingId().toString());
 		}
 
 		if (usageRecord.getRawUsage() != null) {

server/src/com/cloud/capacity/CapacityManagerImpl.java

@@ -479,24 +479,9 @@ public class CapacityManagerImpl extends ManagerBase implements CapacityManager,
         for (VolumeVO volume : volumes) {
             if(volume.getInstanceId() == null)
                 continue;
-            Long vmId = volume.getInstanceId();
+            Long chainSize = volume.getVmSnapshotChainSize();
-            UserVm vm = _userVMDao.findById(vmId);
+            if(chainSize != null)
-            if(vm == null)
+                totalSize += chainSize;
-                continue;
-            ServiceOffering offering = _offeringsDao.findById(vm.getServiceOfferingId());
-            List<VMSnapshotVO> vmSnapshots =  _vmSnapshotDao.findByVm(vmId);
-            long pathCount = 0;
-            long memorySnapshotSize = 0;
-            for (VMSnapshotVO vmSnapshotVO : vmSnapshots) {
-                if(_vmSnapshotDao.listByParent(vmSnapshotVO.getId()).size() == 0)
-                    pathCount++;
-                if(vmSnapshotVO.getType() == VMSnapshot.Type.DiskAndMemory)
-                    memorySnapshotSize += (offering.getRamSize() * 1024L * 1024L);
-            }
-            if(pathCount <= 1)
-                totalSize = totalSize + memorySnapshotSize;
-            else
-                totalSize = totalSize + volume.getSize() * (pathCount - 1) + memorySnapshotSize;
         }
         return totalSize;
     }

server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java

@@ -50,6 +50,7 @@ import com.cloud.agent.api.to.VolumeTO;
 import com.cloud.configuration.dao.ConfigurationDao;
 import com.cloud.event.ActionEvent;
 import com.cloud.event.EventTypes;
+import com.cloud.event.UsageEventUtils;
 import com.cloud.exception.AgentUnavailableException;
 import com.cloud.exception.ConcurrentOperationException;
 import com.cloud.exception.InsufficientCapacityException;
@@ -64,11 +65,14 @@ import com.cloud.hypervisor.Hypervisor.HypervisorType;
 import com.cloud.hypervisor.HypervisorGuruManager;
 import com.cloud.hypervisor.dao.HypervisorCapabilitiesDao;
 import com.cloud.projects.Project.ListProjectResourcesCriteria;
+import com.cloud.service.dao.ServiceOfferingDao;
+import com.cloud.storage.DiskOfferingVO;
 import com.cloud.storage.GuestOSVO;
 import com.cloud.storage.Snapshot;
 import com.cloud.storage.SnapshotVO;
 import com.cloud.storage.StoragePool;
 import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.DiskOfferingDao;
 import com.cloud.storage.dao.GuestOSDao;
 import com.cloud.storage.dao.SnapshotDao;
 import com.cloud.storage.dao.VolumeDao;
@@ -121,6 +125,8 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
     @Inject DataStoreManager dataStoreMgr;
     @Inject ConfigurationDao _configDao;
     @Inject HypervisorCapabilitiesDao _hypervisorCapabilitiesDao;
+    @Inject DiskOfferingDao _diskOfferingDao;
+    @Inject ServiceOfferingDao _serviceOfferingDao;
     int _vmSnapshotMax;
     int _wait;
     StateMachine2<VMSnapshot.State, VMSnapshot.Event, VMSnapshot> _vmSnapshottateMachine ;
@@ -249,7 +255,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
 
         // check hypervisor capabilities
         if(!_hypervisorCapabilitiesDao.isVmSnapshotEnabled(userVmVo.getHypervisorType(), "default"))
-        	throw new InvalidParameterValueException("VM snapshot is not enabled for hypervisor type: " + userVmVo.getHypervisorType());
+            throw new InvalidParameterValueException("VM snapshot is not enabled for hypervisor type: " + userVmVo.getHypervisorType());
 
         // parameter length check
         if(vsDisplayName != null && vsDisplayName.length()>255)
@@ -351,8 +357,8 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
     }
 
     protected VMSnapshot createVmSnapshotInternal(UserVmVO userVm, VMSnapshotVO vmSnapshot, Long hostId) {
+        CreateVMSnapshotAnswer answer = null;
         try {
-            CreateVMSnapshotAnswer answer = null;
             GuestOSVO guestOS = _guestOSDao.findById(userVm.getGuestOSId());
             
             // prepare snapshotVolumeTos
@@ -403,9 +409,37 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
                 s_logger.warn("Create vm snapshot " + vmSnapshot.getName() + " failed for vm: " + userVm.getInstanceName());
                 _vmSnapshotDao.remove(vmSnapshot.getId());
             }
+            if(vmSnapshot.getState() == VMSnapshot.State.Ready && answer != null){
+                for (VolumeTO volumeTo : answer.getVolumeTOs()){
+                    publishUsageEvent(EventTypes.EVENT_VM_SNAPSHOT_CREATE,vmSnapshot,userVm,volumeTo);
+                }
+            }
         }
     }
 
+    private void publishUsageEvent(String type, VMSnapshot vmSnapshot, UserVm userVm, VolumeTO volumeTo){
+        VolumeVO volume = _volumeDao.findById(volumeTo.getId());
+        Long diskOfferingId = volume.getDiskOfferingId();
+        Long offeringId = null;
+        if (diskOfferingId != null) {
+            DiskOfferingVO offering = _diskOfferingDao.findById(diskOfferingId);
+            if (offering != null
+                    && (offering.getType() == DiskOfferingVO.Type.Disk)) {
+                offeringId = offering.getId();
+            }
+        }
+        UsageEventUtils.publishUsageEvent(
+                type,
+                vmSnapshot.getAccountId(),
+                userVm.getDataCenterId(),
+                userVm.getId(),
+                vmSnapshot.getName(),
+                offeringId,     
+                volume.getId(), // save volume's id into templateId field
+                volumeTo.getChainSize(),
+                VMSnapshot.class.getName(), vmSnapshot.getUuid());       
+    }
+    
     protected List<VolumeTO> getVolumeTOList(Long vmId) {
         List<VolumeTO> volumeTOs = new ArrayList<VolumeTO>();
         List<VolumeVO> volumeVos = _volumeDao.findByInstance(vmId);
@@ -532,6 +566,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
             if (volume.getPath() != null) {
                 VolumeVO volumeVO = _volumeDao.findById(volume.getId());
                 volumeVO.setPath(volume.getPath());
+                volumeVO.setVmSnapshotChainSize(volume.getChainSize());
                 _volumeDao.persist(volumeVO);
             }
         }
@@ -590,7 +625,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
     @DB
     protected boolean deleteSnapshotInternal(VMSnapshotVO vmSnapshot) {
         UserVmVO userVm = _userVMDao.findById(vmSnapshot.getVmId());
-        
+        DeleteVMSnapshotAnswer answer = null;
         try {
             vmSnapshotStateTransitTo(vmSnapshot,VMSnapshot.Event.ExpungeRequested);
             Long hostId = pickRunningHost(vmSnapshot.getVmId());
@@ -606,7 +641,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
             GuestOSVO guestOS = _guestOSDao.findById(userVm.getGuestOSId());
             DeleteVMSnapshotCommand deleteSnapshotCommand = new DeleteVMSnapshotCommand(vmInstanceName, vmSnapshotTO, volumeTOs,guestOS.getDisplayName());
             
-            DeleteVMSnapshotAnswer answer = (DeleteVMSnapshotAnswer) sendToPool(hostId, deleteSnapshotCommand);
+            answer = (DeleteVMSnapshotAnswer) sendToPool(hostId, deleteSnapshotCommand);
            
             if (answer != null && answer.getResult()) {
                 processAnswer(vmSnapshot, userVm, answer, hostId);
@@ -620,6 +655,12 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
             String msg = "Delete vm snapshot " + vmSnapshot.getName() + " of vm " + userVm.getInstanceName() + " failed due to " + e.getMessage();
             s_logger.error(msg , e);
             throw new CloudRuntimeException(e.getMessage());
+        } finally{
+            if(answer != null && answer.getResult()){
+                for (VolumeTO volumeTo : answer.getVolumeTOs()){
+                    publishUsageEvent(EventTypes.EVENT_VM_SNAPSHOT_DELETE,vmSnapshot,userVm,volumeTo);
+                }
+            }
         }
     }
 
@@ -682,7 +723,7 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
                     _itMgr.advanceStop(userVm.getUuid(), true);
                 } catch (Exception e) {
                     s_logger.error("Stop VM " + userVm.getInstanceName() + " before reverting failed due to " + e.getMessage());
-    			    throw new CloudRuntimeException(e.getMessage());
+                    throw new CloudRuntimeException(e.getMessage());
                 }
             }
             hostId = pickRunningHost(userVm.getId());
@@ -712,8 +753,8 @@ public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotMana
             List<VolumeTO> volumeTOs = getVolumeTOList(userVm.getId());
             String vmInstanceName = userVm.getInstanceName();
             VMSnapshotTO parent = getSnapshotWithParents(snapshot).getParent();
-            VMSnapshotTO vmSnapshotTO = new VMSnapshotTO(snapshot.getId(), snapshot.getName(), snapshot.getType(),
+            VMSnapshotTO vmSnapshotTO = new VMSnapshotTO(snapshot.getId(), snapshot.getName(), snapshot.getType(), 
-            		snapshot.getCreated().getTime(), snapshot.getDescription(), snapshot.getCurrent(), parent);
+                    snapshot.getCreated().getTime(), snapshot.getDescription(), snapshot.getCurrent(), parent);
             
             GuestOSVO guestOS = _guestOSDao.findById(userVm.getGuestOSId());
             RevertToVMSnapshotCommand revertToSnapshotCommand = new RevertToVMSnapshotCommand(vmInstanceName, vmSnapshotTO, volumeTOs, guestOS.getDisplayName());

server/src/com/cloud/vm/snapshot/VMSnapshotManagerImpl.java.orig

@@ -0,0 +1,913 @@
+// 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.vm.snapshot;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.ejb.Local;
+import javax.inject.Inject;
+import javax.naming.ConfigurationException;
+
+import org.apache.log4j.Logger;
+import org.springframework.stereotype.Component;
+
+import org.apache.cloudstack.api.command.user.vmsnapshot.ListVMSnapshotCmd;
+import org.apache.cloudstack.context.CallContext;
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
+
+import com.cloud.agent.AgentManager;
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.Command;
+import com.cloud.agent.api.CreateVMSnapshotAnswer;
+import com.cloud.agent.api.CreateVMSnapshotCommand;
+import com.cloud.agent.api.DeleteVMSnapshotAnswer;
+import com.cloud.agent.api.DeleteVMSnapshotCommand;
+import com.cloud.agent.api.RevertToVMSnapshotAnswer;
+import com.cloud.agent.api.RevertToVMSnapshotCommand;
+import com.cloud.agent.api.VMSnapshotTO;
+import com.cloud.agent.api.to.VolumeTO;
+import com.cloud.configuration.dao.ConfigurationDao;
+import com.cloud.event.ActionEvent;
+import com.cloud.event.EventTypes;
+import com.cloud.event.UsageEventUtils;
+import com.cloud.exception.AgentUnavailableException;
+import com.cloud.exception.ConcurrentOperationException;
+import com.cloud.exception.InsufficientCapacityException;
+import com.cloud.exception.InvalidParameterValueException;
+import com.cloud.exception.OperationTimedoutException;
+import com.cloud.exception.ResourceAllocationException;
+import com.cloud.exception.ResourceUnavailableException;
+import com.cloud.host.Host;
+import com.cloud.host.HostVO;
+import com.cloud.host.dao.HostDao;
+import com.cloud.hypervisor.Hypervisor.HypervisorType;
+import com.cloud.hypervisor.HypervisorGuruManager;
+import com.cloud.hypervisor.dao.HypervisorCapabilitiesDao;
+import com.cloud.projects.Project.ListProjectResourcesCriteria;
+import com.cloud.service.dao.ServiceOfferingDao;
+import com.cloud.storage.DiskOfferingVO;
+import com.cloud.storage.GuestOSVO;
+import com.cloud.storage.Snapshot;
+import com.cloud.storage.SnapshotVO;
+import com.cloud.storage.StoragePool;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.DiskOfferingDao;
+import com.cloud.storage.dao.GuestOSDao;
+import com.cloud.storage.dao.SnapshotDao;
+import com.cloud.storage.dao.VolumeDao;
+import com.cloud.user.Account;
+import com.cloud.user.AccountManager;
+import com.cloud.user.dao.AccountDao;
+import com.cloud.user.dao.UserDao;
+import com.cloud.uservm.UserVm;
+import com.cloud.utils.DateUtil;
+import com.cloud.utils.NumbersUtil;
+import com.cloud.utils.Ternary;
+import com.cloud.utils.component.ManagerBase;
+import com.cloud.utils.db.DB;
+import com.cloud.utils.db.Filter;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+import com.cloud.utils.db.Transaction;
+import com.cloud.utils.exception.CloudRuntimeException;
+import com.cloud.utils.fsm.NoTransitionException;
+import com.cloud.utils.fsm.StateMachine2;
+import com.cloud.vm.UserVmVO;
+import com.cloud.vm.VMInstanceVO;
+import com.cloud.vm.VirtualMachine;
+import com.cloud.vm.VirtualMachine.State;
+import com.cloud.vm.VirtualMachineManager;
+import com.cloud.vm.VirtualMachineProfile;
+import com.cloud.vm.dao.UserVmDao;
+import com.cloud.vm.dao.VMInstanceDao;
+import com.cloud.vm.snapshot.dao.VMSnapshotDao;
+
+@Component
+@Local(value = { VMSnapshotManager.class, VMSnapshotService.class })
+public class VMSnapshotManagerImpl extends ManagerBase implements VMSnapshotManager, VMSnapshotService {
+    private static final Logger s_logger = Logger.getLogger(VMSnapshotManagerImpl.class);
+    String _name;
+    @Inject VMSnapshotDao _vmSnapshotDao;
+    @Inject VolumeDao _volumeDao;
+    @Inject AccountDao _accountDao;
+    @Inject VMInstanceDao _vmInstanceDao;
+    @Inject UserVmDao _userVMDao;
+    @Inject HostDao _hostDao;
+    @Inject UserDao _userDao;
+    @Inject AgentManager _agentMgr;
+    @Inject HypervisorGuruManager _hvGuruMgr;
+    @Inject AccountManager _accountMgr;
+    @Inject GuestOSDao _guestOSDao;
+    @Inject PrimaryDataStoreDao _storagePoolDao;
+    @Inject SnapshotDao _snapshotDao;
+    @Inject VirtualMachineManager _itMgr;
+    @Inject DataStoreManager dataStoreMgr;
+    @Inject ConfigurationDao _configDao;
+    @Inject HypervisorCapabilitiesDao _hypervisorCapabilitiesDao;
+    @Inject DiskOfferingDao _diskOfferingDao;
+    @Inject ServiceOfferingDao _serviceOfferingDao;
+    int _vmSnapshotMax;
+    int _wait;
+    StateMachine2<VMSnapshot.State, VMSnapshot.Event, VMSnapshot> _vmSnapshottateMachine ;
+
+    @Override
+    public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
+        _name = name;
+        if (_configDao == null) {
+            throw new ConfigurationException(
+                    "Unable to get the configuration dao.");
+        }
+
+        _vmSnapshotMax = NumbersUtil.parseInt(_configDao.getValue("vmsnapshot.max"), VMSNAPSHOTMAX);
+        
+        String value = _configDao.getValue("vmsnapshot.create.wait");
+        _wait = NumbersUtil.parseInt(value, 1800);
+
+        _vmSnapshottateMachine   = VMSnapshot.State.getStateMachine();
+        return true;
+    }
+
+    @Override
+    public boolean start() {
+        return true;
+    }
+
+    @Override
+    public boolean stop() {
+        return true;
+    }
+
+    @Override
+    public List<VMSnapshotVO> listVMSnapshots(ListVMSnapshotCmd cmd) {
+        Account caller = getCaller();
+        List<Long> permittedAccounts = new ArrayList<Long>();
+
+        boolean listAll = cmd.listAll();
+        Long id = cmd.getId();
+        Long vmId = cmd.getVmId();
+        
+        String state = cmd.getState();
+        String keyword = cmd.getKeyword();
+        String name = cmd.getVmSnapshotName();
+        String accountName = cmd.getAccountName();
+
+        Ternary<Long, Boolean, ListProjectResourcesCriteria> domainIdRecursiveListProject = new Ternary<Long, Boolean, ListProjectResourcesCriteria>(
+                cmd.getDomainId(), cmd.isRecursive(), null);
+        _accountMgr.buildACLSearchParameters(caller, id, cmd.getAccountName(), cmd.getProjectId(), permittedAccounts, domainIdRecursiveListProject, listAll,
+                false);
+        Long domainId = domainIdRecursiveListProject.first();
+        Boolean isRecursive = domainIdRecursiveListProject.second();
+        ListProjectResourcesCriteria listProjectResourcesCriteria = domainIdRecursiveListProject.third();
+
+        Filter searchFilter = new Filter(VMSnapshotVO.class, "created", false, cmd.getStartIndex(), cmd.getPageSizeVal());
+        SearchBuilder<VMSnapshotVO> sb = _vmSnapshotDao.createSearchBuilder();
+        _accountMgr.buildACLSearchBuilder(sb, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
+
+        sb.and("vm_id", sb.entity().getVmId(), SearchCriteria.Op.EQ);
+        sb.and("domain_id", sb.entity().getDomainId(), SearchCriteria.Op.EQ);
+        sb.and("status", sb.entity().getState(), SearchCriteria.Op.IN);
+        sb.and("state", sb.entity().getState(), SearchCriteria.Op.EQ);
+        sb.and("id", sb.entity().getId(), SearchCriteria.Op.EQ);
+        sb.and("display_name", sb.entity().getDisplayName(), SearchCriteria.Op.EQ);
+        sb.and("account_id", sb.entity().getAccountId(), SearchCriteria.Op.EQ);
+        sb.done();
+
+        SearchCriteria<VMSnapshotVO> sc = sb.create();
+        _accountMgr.buildACLSearchCriteria(sc, domainId, isRecursive, permittedAccounts, listProjectResourcesCriteria);
+
+        if (accountName != null && cmd.getDomainId() != null) {
+            Account account = _accountMgr.getActiveAccountByName(accountName, cmd.getDomainId());
+            sc.setParameters("account_id", account.getId());
+        }
+
+        if (vmId != null) {
+            sc.setParameters("vm_id", vmId);
+        }
+
+        if (domainId != null) {
+            sc.setParameters("domain_id", domainId);
+        }
+
+        if (state == null) {
+            VMSnapshot.State[] status = { VMSnapshot.State.Ready, VMSnapshot.State.Creating, VMSnapshot.State.Allocated,
+                    VMSnapshot.State.Error, VMSnapshot.State.Expunging, VMSnapshot.State.Reverting };
+            sc.setParameters("status", (Object[]) status);
+        } else {
+            sc.setParameters("state", state);
+        }
+
+        if (name != null) {
+            sc.setParameters("display_name", name);
+        }
+
+        if (keyword != null) {
+            SearchCriteria<VMSnapshotVO> ssc = _vmSnapshotDao.createSearchCriteria();
+            ssc.addOr("name", SearchCriteria.Op.LIKE, "%" + keyword + "%");
+            ssc.addOr("display_name", SearchCriteria.Op.LIKE, "%" + keyword + "%");
+            ssc.addOr("description", SearchCriteria.Op.LIKE, "%" + keyword + "%");
+            sc.addAnd("name", SearchCriteria.Op.SC, ssc);
+        }
+
+        if (id != null) {
+            sc.setParameters("id", id);
+        }
+
+        return _vmSnapshotDao.search(sc, searchFilter);
+    
+    }
+
+    protected Account getCaller(){
+        return CallContext.current().getCallingAccount();
+    }
+    
+    @Override
+    public VMSnapshot allocVMSnapshot(Long vmId, String vsDisplayName, String vsDescription, Boolean snapshotMemory)
+            throws ResourceAllocationException {
+
+        Account caller = getCaller();
+
+        // check if VM exists
+        UserVmVO userVmVo = _userVMDao.findById(vmId);
+        if (userVmVo == null) {
+            throw new InvalidParameterValueException("Creating VM snapshot failed due to VM:" + vmId + " is a system VM or does not exist");
+        }
+
+        // check hypervisor capabilities
+        if(!_hypervisorCapabilitiesDao.isVmSnapshotEnabled(userVmVo.getHypervisorType(), "default"))
+            throw new InvalidParameterValueException("VM snapshot is not enabled for hypervisor type: " + userVmVo.getHypervisorType());
+
+        // parameter length check
+        if(vsDisplayName != null && vsDisplayName.length()>255)
+            throw new InvalidParameterValueException("Creating VM snapshot failed due to length of VM snapshot vsDisplayName should not exceed 255");
+        if(vsDescription != null && vsDescription.length()>255)
+            throw new InvalidParameterValueException("Creating VM snapshot failed due to length of VM snapshot vsDescription should not exceed 255");
+        
+        // VM snapshot display name must be unique for a VM
+        String timeString = DateUtil.getDateDisplayString(DateUtil.GMT_TIMEZONE, new Date(), DateUtil.YYYYMMDD_FORMAT);
+        String vmSnapshotName = userVmVo.getInstanceName() + "_VS_" + timeString;
+        if (vsDisplayName == null) {
+            vsDisplayName = vmSnapshotName;
+        }
+        if(_vmSnapshotDao.findByName(vmId,vsDisplayName) != null){
+            throw new InvalidParameterValueException("Creating VM snapshot failed due to VM snapshot with name" + vsDisplayName + "  already exists");
+        }
+        
+        // check VM state
+        if (userVmVo.getState() != VirtualMachine.State.Running && userVmVo.getState() != VirtualMachine.State.Stopped) {
+            throw new InvalidParameterValueException("Creating vm snapshot failed due to VM:" + vmId + " is not in the running or Stopped state");
+        }
+        
+        if(snapshotMemory && userVmVo.getState() == VirtualMachine.State.Stopped){
+            throw new InvalidParameterValueException("Can not snapshot memory when VM is in stopped state");
+        }
+        
+        // for KVM, only allow snapshot with memory when VM is in running state
+        if(userVmVo.getHypervisorType() == HypervisorType.KVM && userVmVo.getState() == State.Running && !snapshotMemory){
+            throw new InvalidParameterValueException("KVM VM does not allow to take a disk-only snapshot when VM is in running state");
+        }
+        
+        // check access
+        _accountMgr.checkAccess(caller, null, true, userVmVo);
+
+        // check max snapshot limit for per VM
+        if (_vmSnapshotDao.findByVm(vmId).size() >= _vmSnapshotMax) {
+            throw new CloudRuntimeException("Creating vm snapshot failed due to a VM can just have : " + _vmSnapshotMax
+                    + " VM snapshots. Please delete old ones");
+        }
+
+        // check if there are active volume snapshots tasks
+        List<VolumeVO> listVolumes = _volumeDao.findByInstance(vmId);
+        for (VolumeVO volume : listVolumes) {
+            List<SnapshotVO> activeSnapshots = _snapshotDao.listByInstanceId(volume.getInstanceId(), Snapshot.State.Creating,
+                    Snapshot.State.CreatedOnPrimary, Snapshot.State.BackingUp);
+            if (activeSnapshots.size() > 0) {
+                throw new CloudRuntimeException(
+                        "There is other active volume snapshot tasks on the instance to which the volume is attached, please try again later.");
+            }
+        }
+        
+        // check if there are other active VM snapshot tasks
+        if (hasActiveVMSnapshotTasks(vmId)) {
+            throw new CloudRuntimeException("There is other active vm snapshot tasks on the instance, please try again later");
+        }
+        
+        VMSnapshot.Type vmSnapshotType = VMSnapshot.Type.Disk;
+        if(snapshotMemory && userVmVo.getState() == VirtualMachine.State.Running)
+            vmSnapshotType = VMSnapshot.Type.DiskAndMemory;
+        
+        try {
+            VMSnapshotVO vmSnapshotVo = new VMSnapshotVO(userVmVo.getAccountId(), userVmVo.getDomainId(), vmId, vsDescription, vmSnapshotName,
+                    vsDisplayName, userVmVo.getServiceOfferingId(), vmSnapshotType, null);
+            VMSnapshot vmSnapshot = _vmSnapshotDao.persist(vmSnapshotVo);
+            if (vmSnapshot == null) {
+                throw new CloudRuntimeException("Failed to create snapshot for vm: " + vmId);
+            }
+            return vmSnapshot;
+        } catch (Exception e) {
+            String msg = e.getMessage();
+            s_logger.error("Create vm snapshot record failed for vm: " + vmId + " due to: " + msg);
+        }
+        return null;
+    }
+
+    @Override
+    public String getName() {
+        return _name;
+    }
+
+    @Override
+    @ActionEvent(eventType = EventTypes.EVENT_VM_SNAPSHOT_CREATE, eventDescription = "creating VM snapshot", async = true)
+    public VMSnapshot creatVMSnapshot(Long vmId, Long vmSnapshotId) {
+        UserVmVO userVm = _userVMDao.findById(vmId);
+        if (userVm == null) {
+            throw new InvalidParameterValueException("Create vm to snapshot failed due to vm: " + vmId + " is not found");
+        }
+        VMSnapshotVO vmSnapshot = _vmSnapshotDao.findById(vmSnapshotId);
+        if(vmSnapshot == null){
+            throw new CloudRuntimeException("VM snapshot id: " + vmSnapshotId + " can not be found");
+        }
+        Long hostId = pickRunningHost(vmId);
+        try {
+            vmSnapshotStateTransitTo(vmSnapshot, VMSnapshot.Event.CreateRequested);
+        } catch (NoTransitionException e) {
+            throw new CloudRuntimeException(e.getMessage());
+        }
+        return createVmSnapshotInternal(userVm, vmSnapshot, hostId);
+    }
+
+    protected VMSnapshot createVmSnapshotInternal(UserVmVO userVm, VMSnapshotVO vmSnapshot, Long hostId) {
+<<<<<<< HEAD
+        try {
+            CreateVMSnapshotAnswer answer = null;
+=======
+        CreateVMSnapshotAnswer answer = null;
+        try {
+>>>>>>> acd2396... fix CLOUDSTACK-3591 add usage recording for VM snapshots
+            GuestOSVO guestOS = _guestOSDao.findById(userVm.getGuestOSId());
+            
+            // prepare snapshotVolumeTos
+            List<VolumeTO> volumeTOs = getVolumeTOList(userVm.getId());
+            
+            // prepare target snapshotTO and its parent snapshot (current snapshot)
+            VMSnapshotTO current = null;
+            VMSnapshotVO currentSnapshot = _vmSnapshotDao.findCurrentSnapshotByVmId(userVm.getId());
+            if (currentSnapshot != null)
+                current = getSnapshotWithParents(currentSnapshot);
+            VMSnapshotTO target = new VMSnapshotTO(vmSnapshot.getId(),  vmSnapshot.getName(), vmSnapshot.getType(), null, vmSnapshot.getDescription(), false,
+                    current);
+            if (current == null)
+                vmSnapshot.setParent(null);
+            else
+                vmSnapshot.setParent(current.getId());
+
+            CreateVMSnapshotCommand ccmd = new CreateVMSnapshotCommand(userVm.getInstanceName(),target ,volumeTOs, guestOS.getDisplayName(),userVm.getState());
+            ccmd.setWait(_wait);
+            
+            answer = (CreateVMSnapshotAnswer) sendToPool(hostId, ccmd);
+            if (answer != null && answer.getResult()) {
+                processAnswer(vmSnapshot, userVm, answer, hostId);
+                s_logger.debug("Create vm snapshot " + vmSnapshot.getName() + " succeeded for vm: " + userVm.getInstanceName());
+            }else{
+                
+                String errMsg = "Creating VM snapshot: " + vmSnapshot.getName() + " failed";
+                if(answer != null && answer.getDetails() != null)
+                    errMsg = errMsg + " due to " + answer.getDetails();
+                s_logger.error(errMsg);
+                vmSnapshotStateTransitTo(vmSnapshot, VMSnapshot.Event.OperationFailed);
+                throw new CloudRuntimeException(errMsg);
+            }
+            return vmSnapshot;
+        } catch (Exception e) {
+            if(e instanceof AgentUnavailableException){
+                try {
+                    vmSnapshotStateTransitTo(vmSnapshot, VMSnapshot.Event.OperationFailed);
+                } catch (NoTransitionException e1) {
+                    s_logger.error("Cannot set vm snapshot state due to: " + e1.getMessage());
+                }
+            }
+            String msg = e.getMessage();
+            s_logger.error("Create vm snapshot " + vmSnapshot.getName() + " failed for vm: " + userVm.getInstanceName() + " due to " + msg);
+            throw new CloudRuntimeException(msg);
+        } finally{
+            if(vmSnapshot.getState() == VMSnapshot.State.Allocated){
+                s_logger.warn("Create vm snapshot " + vmSnapshot.getName() + " failed for vm: " + userVm.getInstanceName());
+                _vmSnapshotDao.remove(vmSnapshot.getId());
+            }
+            if(vmSnapshot.getState() == VMSnapshot.State.Ready && answer != null){
+                for (VolumeTO volumeTo : answer.getVolumeTOs()){
+                    publishUsageEvent(EventTypes.EVENT_VM_SNAPSHOT_CREATE,vmSnapshot,userVm,volumeTo);
+                }
+            }
+        }
+    }
+
+    private void publishUsageEvent(String type, VMSnapshot vmSnapshot, UserVm userVm, VolumeTO volumeTo){
+        VolumeVO volume = _volumeDao.findById(volumeTo.getId());
+        Long diskOfferingId = volume.getDiskOfferingId();
+        Long offeringId = null;
+        if (diskOfferingId != null) {
+            DiskOfferingVO offering = _diskOfferingDao.findById(diskOfferingId);
+            if (offering != null
+                    && (offering.getType() == DiskOfferingVO.Type.Disk)) {
+                offeringId = offering.getId();
+            }
+        }
+        UsageEventUtils.publishUsageEvent(
+                type,
+                vmSnapshot.getAccountId(),
+                userVm.getDataCenterId(),
+                userVm.getId(),
+                vmSnapshot.getName(),
+                offeringId,     
+                volume.getId(), // save volume's id into templateId field
+                volumeTo.getChainSize(),
+                VMSnapshot.class.getName(), vmSnapshot.getUuid());       
+    }
+    
+    protected List<VolumeTO> getVolumeTOList(Long vmId) {
+        List<VolumeTO> volumeTOs = new ArrayList<VolumeTO>();
+        List<VolumeVO> volumeVos = _volumeDao.findByInstance(vmId);
+        
+        for (VolumeVO volume : volumeVos) {
+            StoragePool pool = (StoragePool)dataStoreMgr.getPrimaryDataStore(volume.getPoolId());
+            VolumeTO volumeTO = new VolumeTO(volume, pool);
+            volumeTOs.add(volumeTO);
+        }
+        return volumeTOs;
+    }
+
+    // get snapshot and its parents recursively
+    private VMSnapshotTO getSnapshotWithParents(VMSnapshotVO snapshot) {
+        Map<Long, VMSnapshotVO> snapshotMap = new HashMap<Long, VMSnapshotVO>();
+        List<VMSnapshotVO> allSnapshots = _vmSnapshotDao.findByVm(snapshot.getVmId());
+        for (VMSnapshotVO vmSnapshotVO : allSnapshots) {
+            snapshotMap.put(vmSnapshotVO.getId(), vmSnapshotVO);
+        }
+
+        VMSnapshotTO currentTO = convert2VMSnapshotTO(snapshot);
+        VMSnapshotTO result = currentTO;
+        VMSnapshotVO current = snapshot;
+        while (current.getParent() != null) {
+            VMSnapshotVO parent = snapshotMap.get(current.getParent());
+            currentTO.setParent(convert2VMSnapshotTO(parent));
+            current = snapshotMap.get(current.getParent());
+            currentTO = currentTO.getParent();
+        }
+        return result;
+    }
+
+    private VMSnapshotTO convert2VMSnapshotTO(VMSnapshotVO vo) {
+        return new VMSnapshotTO(vo.getId(), vo.getName(),  vo.getType(), vo.getCreated().getTime(), vo.getDescription(),
+                vo.getCurrent(), null);
+    }
+
+    protected boolean vmSnapshotStateTransitTo(VMSnapshotVO vsnp, VMSnapshot.Event event) throws NoTransitionException {
+        return _vmSnapshottateMachine.transitTo(vsnp, event, null, _vmSnapshotDao);
+    }
+    
+    @DB
+    protected void processAnswer(VMSnapshotVO vmSnapshot, UserVmVO  userVm, Answer as, Long hostId) {
+        final Transaction txn = Transaction.currentTxn();
+        try {
+            txn.start();
+            if (as instanceof CreateVMSnapshotAnswer) {
+                CreateVMSnapshotAnswer answer = (CreateVMSnapshotAnswer) as;
+                finalizeCreate(vmSnapshot, answer.getVolumeTOs());
+                vmSnapshotStateTransitTo(vmSnapshot, VMSnapshot.Event.OperationSucceeded);
+            } else if (as instanceof RevertToVMSnapshotAnswer) {
+                RevertToVMSnapshotAnswer answer = (RevertToVMSnapshotAnswer) as;
+                finalizeRevert(vmSnapshot, answer.getVolumeTOs());
+                vmSnapshotStateTransitTo(vmSnapshot, VMSnapshot.Event.OperationSucceeded);
+            } else if (as instanceof DeleteVMSnapshotAnswer) {
+                DeleteVMSnapshotAnswer answer = (DeleteVMSnapshotAnswer) as;
+                finalizeDelete(vmSnapshot, answer.getVolumeTOs());
+                _vmSnapshotDao.remove(vmSnapshot.getId());
+            }
+            txn.commit();
+        } catch (Exception e) {
+            String errMsg = "Error while process answer: " + as.getClass() + " due to " + e.getMessage();
+            s_logger.error(errMsg, e);
+            txn.rollback();
+            throw new CloudRuntimeException(errMsg);
+        } finally {
+            txn.close();
+        }
+    }
+
+    protected void finalizeDelete(VMSnapshotVO vmSnapshot, List<VolumeTO> VolumeTOs) {
+        // update volumes path
+        updateVolumePath(VolumeTOs);
+        
+        // update children's parent snapshots
+        List<VMSnapshotVO> children= _vmSnapshotDao.listByParent(vmSnapshot.getId());
+        for (VMSnapshotVO child : children) {
+            child.setParent(vmSnapshot.getParent());
+            _vmSnapshotDao.persist(child);
+        }
+        
+        // update current snapshot
+        VMSnapshotVO current = _vmSnapshotDao.findCurrentSnapshotByVmId(vmSnapshot.getVmId());
+        if(current != null && current.getId() == vmSnapshot.getId() && vmSnapshot.getParent() != null){
+            VMSnapshotVO parent = _vmSnapshotDao.findById(vmSnapshot.getParent());
+            parent.setCurrent(true);
+            _vmSnapshotDao.persist(parent);
+        }
+        vmSnapshot.setCurrent(false);
+        _vmSnapshotDao.persist(vmSnapshot);
+    }
+
+    protected void finalizeCreate(VMSnapshotVO vmSnapshot, List<VolumeTO> VolumeTOs) {
+        // update volumes path
+        updateVolumePath(VolumeTOs);
+        
+        vmSnapshot.setCurrent(true);
+        
+        // change current snapshot
+        if (vmSnapshot.getParent() != null) {
+            VMSnapshotVO previousCurrent = _vmSnapshotDao.findById(vmSnapshot.getParent());
+            previousCurrent.setCurrent(false);
+            _vmSnapshotDao.persist(previousCurrent);
+        }
+        _vmSnapshotDao.persist(vmSnapshot);
+    }
+
+    protected void finalizeRevert(VMSnapshotVO vmSnapshot, List<VolumeTO> volumeToList) {
+        // update volumes path
+        updateVolumePath(volumeToList);
+
+        // update current snapshot, current snapshot is the one reverted to
+        VMSnapshotVO previousCurrent = _vmSnapshotDao.findCurrentSnapshotByVmId(vmSnapshot.getVmId());
+        if(previousCurrent != null){
+            previousCurrent.setCurrent(false);
+            _vmSnapshotDao.persist(previousCurrent);
+        }
+        vmSnapshot.setCurrent(true);
+        _vmSnapshotDao.persist(vmSnapshot);
+    }
+
+    private void updateVolumePath(List<VolumeTO> volumeTOs) {
+        for (VolumeTO volume : volumeTOs) {
+            if (volume.getPath() != null) {
+                VolumeVO volumeVO = _volumeDao.findById(volume.getId());
+                volumeVO.setPath(volume.getPath());
+                volumeVO.setVmSnapshotChainSize(volume.getChainSize());
+                _volumeDao.persist(volumeVO);
+            }
+        }
+    }
+    
+    public VMSnapshotManagerImpl() {
+        
+    }
+    
+    protected Answer sendToPool(Long hostId, Command cmd) throws AgentUnavailableException, OperationTimedoutException {
+        long targetHostId = _hvGuruMgr.getGuruProcessedCommandTargetHost(hostId, cmd);
+        Answer answer = _agentMgr.send(targetHostId, cmd);
+        return answer;
+    }
+    
+    @Override
+    public boolean hasActiveVMSnapshotTasks(Long vmId){
+        List<VMSnapshotVO> activeVMSnapshots = _vmSnapshotDao.listByInstanceId(vmId,
+                VMSnapshot.State.Creating, VMSnapshot.State.Expunging,VMSnapshot.State.Reverting,VMSnapshot.State.Allocated);
+        return activeVMSnapshots.size() > 0;
+    }
+
+    @Override
+    @ActionEvent(eventType = EventTypes.EVENT_VM_SNAPSHOT_DELETE, eventDescription = "delete vm snapshots", async=true)
+    public boolean deleteVMSnapshot(Long vmSnapshotId) {
+        Account caller = getCaller();
+
+        VMSnapshotVO vmSnapshot = _vmSnapshotDao.findById(vmSnapshotId);
+        if (vmSnapshot == null) {
+            throw new InvalidParameterValueException("unable to find the vm snapshot with id " + vmSnapshotId);
+        }
+
+        _accountMgr.checkAccess(caller, null, true, vmSnapshot);
+        
+        // check VM snapshot states, only allow to delete vm snapshots in created and error state
+        if (VMSnapshot.State.Ready != vmSnapshot.getState() && VMSnapshot.State.Expunging != vmSnapshot.getState() && VMSnapshot.State.Error != vmSnapshot.getState()) {
+            throw new InvalidParameterValueException("Can't delete the vm snapshotshot " + vmSnapshotId + " due to it is not in Created or Error, or Expunging State");
+        }
+        
+        // check if there are other active VM snapshot tasks
+        if (hasActiveVMSnapshotTasks(vmSnapshot.getVmId())) {
+            List<VMSnapshotVO> expungingSnapshots = _vmSnapshotDao.listByInstanceId(vmSnapshot.getVmId(), VMSnapshot.State.Expunging);
+            if(expungingSnapshots.size() > 0 && expungingSnapshots.get(0).getId() == vmSnapshot.getId())
+                s_logger.debug("Target VM snapshot already in expunging state, go on deleting it: " + vmSnapshot.getDisplayName());
+            else
+                throw new InvalidParameterValueException("There is other active vm snapshot tasks on the instance, please try again later");
+        }
+
+        if(vmSnapshot.getState() == VMSnapshot.State.Allocated){
+            return _vmSnapshotDao.remove(vmSnapshot.getId());
+        }else{
+            return deleteSnapshotInternal(vmSnapshot);
+        }
+    }
+
+    @DB
+    protected boolean deleteSnapshotInternal(VMSnapshotVO vmSnapshot) {
+        UserVmVO userVm = _userVMDao.findById(vmSnapshot.getVmId());
+        DeleteVMSnapshotAnswer answer = null;
+        try {
+            vmSnapshotStateTransitTo(vmSnapshot,VMSnapshot.Event.ExpungeRequested);
+            Long hostId = pickRunningHost(vmSnapshot.getVmId());
+
+            // prepare snapshotVolumeTos
+            List<VolumeTO> volumeTOs = getVolumeTOList(vmSnapshot.getVmId());
+            
+            // prepare DeleteVMSnapshotCommand
+            String vmInstanceName = userVm.getInstanceName();
+            VMSnapshotTO parent = getSnapshotWithParents(vmSnapshot).getParent();
+            VMSnapshotTO vmSnapshotTO = new VMSnapshotTO(vmSnapshot.getId(), vmSnapshot.getName(), vmSnapshot.getType(),
+                    vmSnapshot.getCreated().getTime(), vmSnapshot.getDescription(), vmSnapshot.getCurrent(), parent);
+            GuestOSVO guestOS = _guestOSDao.findById(userVm.getGuestOSId());
+            DeleteVMSnapshotCommand deleteSnapshotCommand = new DeleteVMSnapshotCommand(vmInstanceName, vmSnapshotTO, volumeTOs,guestOS.getDisplayName());
+            
+            answer = (DeleteVMSnapshotAnswer) sendToPool(hostId, deleteSnapshotCommand);
+           
+            if (answer != null && answer.getResult()) {
+                processAnswer(vmSnapshot, userVm, answer, hostId);
+                s_logger.debug("Delete VM snapshot " + vmSnapshot.getName() + " succeeded for vm: " + userVm.getInstanceName());
+                return true;
+            } else {
+                s_logger.error("Delete vm snapshot " + vmSnapshot.getName() + " of vm " + userVm.getInstanceName() + " failed due to " + answer.getDetails());
+                return false;
+            }
+        } catch (Exception e) {
+            String msg = "Delete vm snapshot " + vmSnapshot.getName() + " of vm " + userVm.getInstanceName() + " failed due to " + e.getMessage();
+            s_logger.error(msg , e);
+            throw new CloudRuntimeException(e.getMessage());
+        } finally{
+            if(answer != null && answer.getResult()){
+                for (VolumeTO volumeTo : answer.getVolumeTOs()){
+                    publishUsageEvent(EventTypes.EVENT_VM_SNAPSHOT_DELETE,vmSnapshot,userVm,volumeTo);
+                }
+            }
+        }
+    }
+
+    @Override
+    @ActionEvent(eventType = EventTypes.EVENT_VM_SNAPSHOT_REVERT, eventDescription = "revert to VM snapshot", async = true)
+    public UserVm revertToSnapshot(Long vmSnapshotId) throws InsufficientCapacityException, ResourceUnavailableException, ConcurrentOperationException {
+
+        // check if VM snapshot exists in DB
+        VMSnapshotVO vmSnapshotVo = _vmSnapshotDao.findById(vmSnapshotId);
+        if (vmSnapshotVo == null) {
+            throw new InvalidParameterValueException(
+                    "unable to find the vm snapshot with id " + vmSnapshotId);
+        }
+        Long vmId = vmSnapshotVo.getVmId();
+        UserVmVO userVm = _userVMDao.findById(vmId);
+        // check if VM exists
+        if (userVm == null) {
+            throw new InvalidParameterValueException("Revert vm to snapshot: "
+                    + vmSnapshotId + " failed due to vm: " + vmId
+                    + " is not found");
+        }
+        
+        // check if there are other active VM snapshot tasks
+        if (hasActiveVMSnapshotTasks(vmId)) {
+            throw new InvalidParameterValueException("There is other active vm snapshot tasks on the instance, please try again later");
+        }
+        
+        Account caller = getCaller();
+        _accountMgr.checkAccess(caller, null, true, vmSnapshotVo);
+
+        // VM should be in running or stopped states
+        if (userVm.getState() != VirtualMachine.State.Running
+                && userVm.getState() != VirtualMachine.State.Stopped) {
+            throw new InvalidParameterValueException(
+                    "VM Snapshot reverting failed due to vm is not in the state of Running or Stopped.");
+        }
+        
+        // if snapshot is not created, error out
+        if (vmSnapshotVo.getState() != VMSnapshot.State.Ready) {
+            throw new InvalidParameterValueException(
+                    "VM Snapshot reverting failed due to vm snapshot is not in the state of Created.");
+        }
+
+        UserVmVO vm = null;
+        Long hostId = null;
+        
+        // start or stop VM first, if revert from stopped state to running state, or from running to stopped
+        if(userVm.getState() == VirtualMachine.State.Stopped && vmSnapshotVo.getType() == VMSnapshot.Type.DiskAndMemory){
+            try {
+<<<<<<< HEAD
+                _itMgr.advanceStart(userVm.getUuid(), new HashMap<VirtualMachineProfile.Param, Object>());
+                vm = _userVMDao.findById(userVm.getId());
+        	    hostId = vm.getHostId();
+        	} catch (Exception e) {
+        	    s_logger.error("Start VM " + userVm.getInstanceName() + " before reverting failed due to " + e.getMessage());
+        	    throw new CloudRuntimeException(e.getMessage());
+        	}
+        }else {
+            if(userVm.getState() == VirtualMachine.State.Running && vmSnapshotVo.getType() == VMSnapshot.Type.Disk){
+                try {
+                    _itMgr.advanceStop(userVm.getUuid(), true);
+=======
+                vm = _itMgr.advanceStart(userVm, new HashMap<VirtualMachineProfile.Param, Object>(), callerUser, owner);
+                hostId = vm.getHostId();
+            } catch (Exception e) {
+                s_logger.error("Start VM " + userVm.getInstanceName() + " before reverting failed due to " + e.getMessage());
+                throw new CloudRuntimeException(e.getMessage());
+            }
+        }else {
+            if(userVm.getState() == VirtualMachine.State.Running && vmSnapshotVo.getType() == VMSnapshot.Type.Disk){
+                try {
+                    _itMgr.advanceStop(userVm, true, callerUser, owner);
+>>>>>>> acd2396... fix CLOUDSTACK-3591 add usage recording for VM snapshots
+                } catch (Exception e) {
+                    s_logger.error("Stop VM " + userVm.getInstanceName() + " before reverting failed due to " + e.getMessage());
+                    throw new CloudRuntimeException(e.getMessage());
+                }
+            }
+            hostId = pickRunningHost(userVm.getId());
+        }
+        
+        if(hostId == null)
+            throw new CloudRuntimeException("Can not find any host to revert snapshot " + vmSnapshotVo.getName());
+        
+        // check if there are other active VM snapshot tasks
+        if (hasActiveVMSnapshotTasks(userVm.getId())) {
+            throw new InvalidParameterValueException("There is other active vm snapshot tasks on the instance, please try again later");
+        }
+        
+        userVm = _userVMDao.findById(userVm.getId());
+        try {
+            vmSnapshotStateTransitTo(vmSnapshotVo, VMSnapshot.Event.RevertRequested);
+        } catch (NoTransitionException e) {
+            throw new CloudRuntimeException(e.getMessage());
+        }
+        return revertInternal(userVm, vmSnapshotVo, hostId);
+    }
+
+    private UserVm revertInternal(UserVmVO userVm, VMSnapshotVO vmSnapshotVo, Long hostId) {
+        try {
+            VMSnapshotVO snapshot = _vmSnapshotDao.findById(vmSnapshotVo.getId());
+            // prepare RevertToSnapshotCommand
+            List<VolumeTO> volumeTOs = getVolumeTOList(userVm.getId());
+            String vmInstanceName = userVm.getInstanceName();
+            VMSnapshotTO parent = getSnapshotWithParents(snapshot).getParent();
+<<<<<<< HEAD
+            VMSnapshotTO vmSnapshotTO = new VMSnapshotTO(snapshot.getId(), snapshot.getName(), snapshot.getType(),
+            		snapshot.getCreated().getTime(), snapshot.getDescription(), snapshot.getCurrent(), parent);
+=======
+            VMSnapshotTO vmSnapshotTO = new VMSnapshotTO(snapshot.getId(), snapshot.getName(), snapshot.getType(), 
+                    snapshot.getCreated().getTime(), snapshot.getDescription(), snapshot.getCurrent(), parent);
+>>>>>>> acd2396... fix CLOUDSTACK-3591 add usage recording for VM snapshots
+            
+            GuestOSVO guestOS = _guestOSDao.findById(userVm.getGuestOSId());
+            RevertToVMSnapshotCommand revertToSnapshotCommand = new RevertToVMSnapshotCommand(vmInstanceName, vmSnapshotTO, volumeTOs, guestOS.getDisplayName());
+           
+            RevertToVMSnapshotAnswer answer = (RevertToVMSnapshotAnswer) sendToPool(hostId, revertToSnapshotCommand);
+            if (answer != null && answer.getResult()) {
+                processAnswer(vmSnapshotVo, userVm, answer, hostId);
+                s_logger.debug("RevertTo " + vmSnapshotVo.getName() + " succeeded for vm: " + userVm.getInstanceName());
+            } else {
+                String errMsg = "Revert VM: " + userVm.getInstanceName() + " to snapshot: "+ vmSnapshotVo.getName() + " failed";
+                if(answer != null && answer.getDetails() != null)
+                    errMsg = errMsg + " due to " + answer.getDetails();
+                s_logger.error(errMsg);
+                // agent report revert operation fails
+                vmSnapshotStateTransitTo(vmSnapshotVo, VMSnapshot.Event.OperationFailed);
+                throw new CloudRuntimeException(errMsg);
+            }
+        } catch (Exception e) {
+            if(e instanceof AgentUnavailableException){
+                try {
+                    vmSnapshotStateTransitTo(vmSnapshotVo, VMSnapshot.Event.OperationFailed);
+                } catch (NoTransitionException e1) {
+                    s_logger.error("Cannot set vm snapshot state due to: " + e1.getMessage());
+                }
+            }
+            // for other exceptions, do not change VM snapshot state, leave it for snapshotSync
+            String errMsg = "revert vm: " + userVm.getInstanceName() + " to snapshot " + vmSnapshotVo.getName() + " failed due to " + e.getMessage();
+            s_logger.error(errMsg);
+            throw new CloudRuntimeException(e.getMessage());
+        }
+        return userVm;
+    }
+
+
+    @Override
+    public VMSnapshot getVMSnapshotById(Long id) {
+        VMSnapshotVO vmSnapshot = _vmSnapshotDao.findById(id);
+        return vmSnapshot;
+    }
+
+    protected Long pickRunningHost(Long vmId) {
+        UserVmVO vm = _userVMDao.findById(vmId);
+        // use VM's host if VM is running
+        if(vm.getState() == State.Running)
+            return vm.getHostId();
+        
+        // check if lastHostId is available
+        if(vm.getLastHostId() != null){
+           HostVO lastHost =  _hostDao.findById(vm.getLastHostId());
+           if(lastHost.getStatus() == com.cloud.host.Status.Up && !lastHost.isInMaintenanceStates())
+               return lastHost.getId();
+        }
+        
+        List<VolumeVO> listVolumes = _volumeDao.findByInstance(vmId);
+        if (listVolumes == null || listVolumes.size() == 0) {
+            throw new InvalidParameterValueException("vmInstance has no volumes");
+        }
+        VolumeVO volume = listVolumes.get(0);
+        Long poolId = volume.getPoolId();
+        if (poolId == null) {
+            throw new InvalidParameterValueException("pool id is not found");
+        }
+        StoragePoolVO storagePool = _storagePoolDao.findById(poolId);
+        if (storagePool == null) {
+            throw new InvalidParameterValueException("storage pool is not found");
+        }
+        List<HostVO> listHost = _hostDao.listAllUpAndEnabledNonHAHosts(Host.Type.Routing, storagePool.getClusterId(), storagePool.getPodId(),
+                storagePool.getDataCenterId(), null);
+        if (listHost == null || listHost.size() == 0) {
+            throw new InvalidParameterValueException("no host in up state is found");
+        }
+        return listHost.get(0).getId();
+    }
+
+    @Override
+    public VirtualMachine getVMBySnapshotId(Long id) {
+        VMSnapshotVO vmSnapshot = _vmSnapshotDao.findById(id);
+        if(vmSnapshot == null){
+            throw new InvalidParameterValueException("unable to find the vm snapshot with id " + id);
+        }
+        Long vmId = vmSnapshot.getVmId();
+        UserVmVO vm = _userVMDao.findById(vmId);
+        return vm;
+    }
+
+    @Override
+    public boolean deleteAllVMSnapshots(long vmId, VMSnapshot.Type type) {
+        boolean result = true;
+        List<VMSnapshotVO> listVmSnapshots = _vmSnapshotDao.findByVm(vmId);
+        if (listVmSnapshots == null || listVmSnapshots.isEmpty()) {
+            return true;
+        }
+        for (VMSnapshotVO snapshot : listVmSnapshots) {
+            VMSnapshotVO target = _vmSnapshotDao.findById(snapshot.getId());
+            if(type != null && target.getType() != type)
+                continue;
+            if (!deleteSnapshotInternal(target)) {
+                result = false;
+                break;
+            }
+        }
+        return result;
+    }
+
+    @Override
+    public boolean syncVMSnapshot(VMInstanceVO vm, Long hostId) {
+        try{
+            
+            UserVmVO userVm = _userVMDao.findById(vm.getId());
+            if(userVm == null)
+                return false;
+            
+            List<VMSnapshotVO> vmSnapshotsInExpungingStates = _vmSnapshotDao.listByInstanceId(vm.getId(), VMSnapshot.State.Expunging, VMSnapshot.State.Reverting, VMSnapshot.State.Creating);
+            for (VMSnapshotVO vmSnapshotVO : vmSnapshotsInExpungingStates) {
+                if(vmSnapshotVO.getState() == VMSnapshot.State.Expunging){
+                    return deleteSnapshotInternal(vmSnapshotVO);
+                }else if(vmSnapshotVO.getState() == VMSnapshot.State.Creating){
+                    return createVmSnapshotInternal(userVm, vmSnapshotVO, hostId) != null;
+                }else if(vmSnapshotVO.getState() == VMSnapshot.State.Reverting){
+                    return revertInternal(userVm, vmSnapshotVO, hostId) != null;
+                }
+            }
+        }catch (Exception e) {
+            s_logger.error(e.getMessage(),e);
+            if(_vmSnapshotDao.listByInstanceId(vm.getId(), VMSnapshot.State.Expunging).size() == 0)
+                return true;
+            else
+                return false;
+        }
+        return false;
+    }
+ 
+}

setup/db/db/schema-410to420.sql

@@ -2202,3 +2202,21 @@ CREATE VIEW `cloud`.`account_vmstats_view` AS
     where
         vm_type = 'User'
     group by account_id , state;
+INSERT IGNORE INTO `cloud`.`configuration` VALUES ('Network', 'DEFAULT', 'management-server', 'network.loadbalancer.haproxy.max.conn', '4096', 'Load Balancer(haproxy) maximum number of concurrent connections(global max)');
+
+
+DROP TABLE IF EXISTS `cloud_usage`.`usage_vmsnapshot`;
+CREATE TABLE `cloud_usage`.`usage_vmsnapshot` (
+  `id` bigint(20) unsigned NOT NULL,
+  `zone_id` bigint(20) unsigned NOT NULL,
+  `account_id` bigint(20) unsigned NOT NULL,
+  `domain_id` bigint(20) unsigned NOT NULL,
+  `vm_id` bigint(20) unsigned NOT NULL,
+  `disk_offering_id` bigint(20) unsigned,
+  `size` bigint(20),
+  `created` datetime NOT NULL,
+  `processed` datetime,
+  INDEX `i_usage_vmsnapshot` (`account_id`,`id`,`vm_id`,`created`)
+) ENGINE=InnoDB CHARSET=utf8;
+
+ALTER TABLE volumes ADD COLUMN vm_snapshot_chain_size bigint(20) unsigned;

usage/src/com/cloud/usage/UsageManagerImpl.java

@@ -53,6 +53,7 @@ import com.cloud.usage.dao.UsagePortForwardingRuleDao;
 import com.cloud.usage.dao.UsageSecurityGroupDao;
 import com.cloud.usage.dao.UsageStorageDao;
 import com.cloud.usage.dao.UsageVMInstanceDao;
+import com.cloud.usage.dao.UsageVMSnapshotDao;
 import com.cloud.usage.dao.UsageVPNUserDao;
 import com.cloud.usage.dao.UsageVmDiskDao;
 import com.cloud.usage.dao.UsageVolumeDao;
@@ -64,6 +65,7 @@ import com.cloud.usage.parser.PortForwardingUsageParser;
 import com.cloud.usage.parser.SecurityGroupUsageParser;
 import com.cloud.usage.parser.StorageUsageParser;
 import com.cloud.usage.parser.VMInstanceUsageParser;
+import com.cloud.usage.parser.VMSnapshotUsageParser;
 import com.cloud.usage.parser.VPNUserUsageParser;
 import com.cloud.usage.parser.VmDiskUsageParser;
 import com.cloud.usage.parser.VolumeUsageParser;
@@ -116,7 +118,8 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
     @Inject protected AlertManager _alertMgr;
     @Inject protected UsageEventDao _usageEventDao;
     @Inject ConfigurationDao _configDao;
-
+    @Inject private UsageVMSnapshotDao m_usageVMSnapshotDao;
+    
     private String m_version = null;
     private final Calendar m_jobExecTime = Calendar.getInstance();
     private int m_aggregationDuration = 0;
@@ -236,7 +239,7 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
         }
 
         // use the configured exec time and aggregation duration for scheduling the job
-        m_scheduledFuture = m_executor.scheduleAtFixedRate(this, m_jobExecTime.getTimeInMillis() - System.currentTimeMillis(), m_aggregationDuration * 60 * 1000, TimeUnit.MILLISECONDS);
+        m_scheduledFuture = m_executor.scheduleAtFixedRate(this, m_jobExecTime.getTimeInMillis()  - System.currentTimeMillis(), m_aggregationDuration * 60 * 1000, TimeUnit.MILLISECONDS);
 
         m_heartbeat = m_heartbeatExecutor.scheduleAtFixedRate(new Heartbeat(), /* start in 15 seconds...*/15*1000, /* check database every minute*/60*1000, TimeUnit.MILLISECONDS);
         
@@ -857,6 +860,12 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
                 s_logger.debug("VPN user usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");
             }
         }
+        parsed = VMSnapshotUsageParser.parse(account, currentStartDate, currentEndDate);
+        if (s_logger.isDebugEnabled()) {
+            if (!parsed) {
+                s_logger.debug("VM Snapshot usage successfully parsed? " + parsed + " (for account: " + account.getAccountName() + ", id: " + account.getId() + ")");
+            }
+        }
         return parsed;
     }
 
@@ -884,6 +893,8 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
             createVPNUserEvent(event);
         } else if (isSecurityGroupEvent(eventType)) {
             createSecurityGroupEvent(event);
+        } else if (isVmSnapshotEvent(eventType)){
+            createVMSnapshotEvent(event);
         }
     }
 
@@ -951,7 +962,12 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
         return (eventType.equals(EventTypes.EVENT_SECURITY_GROUP_ASSIGN) ||
                 eventType.equals(EventTypes.EVENT_SECURITY_GROUP_REMOVE));
     }
-    
+
+    private boolean isVmSnapshotEvent(String eventType){
+        if (eventType == null) return false;
+        return (eventType.equals(EventTypes.EVENT_VM_SNAPSHOT_CREATE) ||
+                eventType.equals(EventTypes.EVENT_VM_SNAPSHOT_DELETE));
+    }
     private void createVMHelperEvent(UsageEventVO event) {
 
         // One record for handling VM.START and VM.STOP
@@ -1618,7 +1634,22 @@ public class UsageManagerImpl extends ManagerBase implements UsageManager, Runna
             }
         }
     }
-    
+
+    private void createVMSnapshotEvent(UsageEventVO event){
+        Long vmId = event.getResourceId();
+        Long volumeId = event.getTemplateId();
+        Long offeringId = event.getOfferingId();
+        Long zoneId = event.getZoneId();
+        Long accountId = event.getAccountId();
+        long size = event.getSize();
+        Date created = event.getCreateDate();
+        Account acct = m_accountDao.findByIdIncludingRemoved(event.getAccountId());
+        Long domainId = acct.getDomainId();
+        UsageVMSnapshotVO vsVO  = new UsageVMSnapshotVO(volumeId,zoneId,accountId,
+                domainId,vmId,offeringId, size, created, null);
+        m_usageVMSnapshotDao.persist(vsVO);
+    }
+
     private class Heartbeat implements Runnable {
         public void run() {
             Transaction usageTxn = Transaction.open(Transaction.USAGE_DB);

usage/src/com/cloud/usage/parser/VMSnapshotUsageParser.java

@@ -0,0 +1,153 @@
+// 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
+// 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.usage.parser;
+
+import java.text.DecimalFormat;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import javax.annotation.PostConstruct;
+import javax.inject.Inject;
+
+import org.apache.cloudstack.usage.UsageTypes;
+import org.apache.log4j.Logger;
+import org.springframework.stereotype.Component;
+
+import com.cloud.usage.UsageVMSnapshotVO;
+import com.cloud.usage.UsageVO;
+import com.cloud.usage.dao.UsageDao;
+import com.cloud.usage.dao.UsageVMSnapshotDao;
+import com.cloud.user.AccountVO;
+
+@Component
+public class VMSnapshotUsageParser {
+    public static final Logger s_logger = Logger.getLogger(VMSnapshotUsageParser.class.getName());
+    
+    private static UsageDao m_usageDao;
+    private static UsageVMSnapshotDao m_usageVMSnapshotDao;
+
+    @Inject private UsageDao _usageDao;
+    @Inject private UsageVMSnapshotDao _usageVMSnapshotDao;
+    
+    @PostConstruct
+    void init() {
+        m_usageDao = _usageDao;
+        m_usageVMSnapshotDao = _usageVMSnapshotDao;
+    }
+    
+    public static boolean parse(AccountVO account, Date startDate, Date endDate) {
+        if (s_logger.isDebugEnabled()) {
+            s_logger.debug("Parsing all VmSnapshot volume usage events for account: " + account.getId());
+        }
+        if ((endDate == null) || endDate.after(new Date())) {
+            endDate = new Date();
+        }
+
+        List<UsageVMSnapshotVO> usageUsageVMSnapshots = 
+                m_usageVMSnapshotDao.getUsageRecords(account.getId(), account.getDomainId(), startDate, endDate);
+        
+        if(usageUsageVMSnapshots.isEmpty()){
+            s_logger.debug("No VM snapshot usage events for this period");
+            return true;
+        }
+
+        Map<String, UsageVMSnapshotVO> unprocessedUsage = new HashMap<String, UsageVMSnapshotVO>();
+        for (UsageVMSnapshotVO usageRec : usageUsageVMSnapshots) {
+            long zoneId = usageRec.getZoneId();
+            Long volId = usageRec.getId();
+            long vmId = usageRec.getVmId();
+            String key = vmId+":"+volId;
+            if(usageRec.getCreated().before(startDate)){
+                unprocessedUsage.put(key, usageRec);
+                continue;
+            }
+            UsageVMSnapshotVO previousEvent = m_usageVMSnapshotDao.
+                    getPreviousUsageRecord(usageRec);
+            if(previousEvent == null || previousEvent.getSize() == 0){
+                unprocessedUsage.put(key, usageRec);
+                continue;
+            }
+           
+            Date previousCreated = previousEvent.getCreated();
+            if (previousCreated.before(startDate)) {
+                previousCreated = startDate;
+            }
+
+            Date createDate = usageRec.getCreated();
+            long duration = (createDate.getTime() - previousCreated.getTime()) + 1;
+            
+            createUsageRecord(UsageTypes.VM_SNAPSHOT, duration, previousCreated, createDate, 
+                    account, volId, zoneId, previousEvent.getDiskOfferingId(), 
+                    vmId, previousEvent.getSize());
+            previousEvent.setProcessed(new Date());
+            m_usageVMSnapshotDao.update(previousEvent);
+            
+            if(usageRec.getSize() == 0){
+                usageRec.setProcessed(new Date());
+                m_usageVMSnapshotDao.update(usageRec);
+            }
+            else
+                unprocessedUsage.put(key, usageRec);
+        }
+
+        for (String key : unprocessedUsage.keySet()){
+            UsageVMSnapshotVO usageRec = unprocessedUsage.get(key);
+            Date created = usageRec.getCreated();
+            if (created.before(startDate)) {
+                created = startDate;
+            }
+            long duration = (endDate.getTime() - created.getTime()) + 1;
+            createUsageRecord(UsageTypes.VM_SNAPSHOT, duration, created, endDate, 
+                    account, usageRec.getId(), usageRec.getZoneId(), usageRec.getDiskOfferingId(), 
+                    usageRec.getVmId(), usageRec.getSize());
+        }
+
+        return true;
+    }
+
+    private static void createUsageRecord(int type, long runningTime, Date startDate, Date endDate, AccountVO account, long volId, long zoneId, Long doId, Long vmId, long size) {
+        // Our smallest increment is hourly for now
+        if (s_logger.isDebugEnabled()) {
+            s_logger.debug("Total running time " + runningTime + "ms");
+        }
+
+        float usage = runningTime / 1000f / 60f / 60f;
+
+        DecimalFormat dFormat = new DecimalFormat("#.######");
+        String usageDisplay = dFormat.format(usage);
+
+        if (s_logger.isDebugEnabled()) {
+            s_logger.debug("Creating VMSnapshot Volume usage record for vol: " + volId + ", usage: " + usageDisplay + ", startDate: " + startDate + ", endDate: " + endDate + ", for account: " + account.getId());
+        }
+
+        // Create the usage record
+        String usageDesc = "VMSnapshot Usage: " + "VM Id: " + vmId + " Volume Id: "+volId+" ";
+
+        if(doId != null){
+            usageDesc += " DiskOffering: " +doId ;
+        } 
+        
+        usageDesc += " Size: " + size;
+        
+        UsageVO usageRecord = new UsageVO(zoneId, account.getId(), account.getDomainId(), usageDesc, usageDisplay + " Hrs", type,
+                new Double(usage), vmId, null, doId, null, volId, size, startDate, endDate);
+        m_usageDao.persist(usageRecord);
+    }
+
+}