Fixes VM snapshot usage records that are attributed to the old snapshot

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

@@ -37,7 +37,7 @@
 public class UsageVMSnapshotDaoImpl extends GenericDaoBase<UsageVMSnapshotVO, Long> implements UsageVMSnapshotDao {
     protected static final String GET_USAGE_RECORDS_BY_ACCOUNT = "SELECT volume_id, zone_id, account_id, domain_id, vm_id, disk_offering_id, size, created, processed, vm_snapshot_id "
         + " FROM usage_vmsnapshot" + " WHERE account_id = ? " + " AND ( (created BETWEEN ? AND ?) OR "
-        + "      (created < ? AND processed is NULL) ) ORDER BY created asc";
+        + "      (created < ? AND (processed IS NULL OR processed >= ?)) ) ORDER BY created asc";
     protected static final String UPDATE_DELETED = "UPDATE usage_vmsnapshot SET processed = ? WHERE account_id = ? AND volume_id = ? and vm_id = ?  and created = ?";
 
     protected static final String PREVIOUS_QUERY = "SELECT volume_id, zone_id, account_id, domain_id, vm_id, disk_offering_id,size, created, processed, vm_snapshot_id "
@@ -80,6 +80,7 @@ public List<UsageVMSnapshotVO> getUsageRecords(Long accountId, Long domainId, Da
             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));
+            pstmt.setString(i++, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), startDate));
 
             ResultSet rs = pstmt.executeQuery();
             while (rs.next()) {

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

@@ -50,6 +50,7 @@
 import org.apache.cloudstack.usage.UsageTypes;
 import org.apache.cloudstack.utils.usage.UsageUtils;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Component;
 
 import com.cloud.alert.AlertManager;
@@ -1942,12 +1943,7 @@ protected void createUsageVMSnapshot(UsageEventVO event) {
         Date created = event.getCreateDate();
         long size = (event.getSize() == null) ? 0 : event.getSize();
 
-        UsageEventDetailsVO detailVO = _usageEventDetailsDao.findDetail(event.getId(), UsageEventVO.DynamicParameters.vmSnapshotId.name());
-        Long vmSnapshotId = null;
-        if (detailVO != null) {
-            String snapId = detailVO.getValue();
-            vmSnapshotId = Long.valueOf(snapId);
-        }
+        Long vmSnapshotId = getVmSnapshotIdFromEvent(event);
         logger.debug("Creating usage Instance Snapshot for Instance ID [{}] assigned to Account [{}] domain [{}], zone [{}], and created at [{}].",
                 vmId, accountId, domainId, zoneId, DateUtil.displayDateInTimezone(usageAggregationTimeZone, event.getCreateDate()));
         UsageVMSnapshotVO vsVO = new UsageVMSnapshotVO(volumeId, zoneId, accountId, domainId, vmId, offeringId, size, created, null);
@@ -1965,7 +1961,14 @@ protected void deleteUsageVMSnapshot(UsageEventVO event) {
         Long diskOfferingId = event.getOfferingId();
         long vmId = event.getResourceId();
         long zoneId = event.getZoneId();
-        List<UsageVMSnapshotVO> usageVMSnapshots = findUsageVMSnapshots(accountId, zoneId, domainId, vmId, diskOfferingId);
+        Long vmSnapshotId = getVmSnapshotIdFromEvent(event);
+        List<UsageVMSnapshotVO> usageVMSnapshots;
+        if (vmSnapshotId != null) {
+            // Only close the entry of the Instance Snapshot (and volume) being deleted; other Instance Snapshots of the same Instance keep accruing usage.
+            usageVMSnapshots = findUsageVMSnapshots(accountId, zoneId, domainId, vmId, event.getTemplateId(), vmSnapshotId);
+        } else {
+            usageVMSnapshots = findUsageVMSnapshots(accountId, zoneId, domainId, vmId, diskOfferingId);
+        }
         if (CollectionUtils.isEmpty(usageVMSnapshots)){
             logger.warn(String.format("No usage entry for Instance Snapshot for Instance ID [%s] assigned to Account [%s] domain [%s] and zone [%s] was found.",
                     vmId, accountId, domainId, zoneId));
@@ -1982,6 +1985,28 @@ protected void deleteUsageVMSnapshot(UsageEventVO event) {
         }
     }
 
+    protected Long getVmSnapshotIdFromEvent(UsageEventVO event) {
+        UsageEventDetailsVO detailVO = _usageEventDetailsDao.findDetail(event.getId(), UsageEventVO.DynamicParameters.vmSnapshotId.name());
+        if (detailVO == null || StringUtils.isBlank(detailVO.getValue())) {
+            return null;
+        }
+        return Long.valueOf(detailVO.getValue());
+    }
+
+    protected List<UsageVMSnapshotVO> findUsageVMSnapshots(long accountId, long zoneId, long domainId, long vmId, Long volumeId, long vmSnapshotId) {
+        SearchCriteria<UsageVMSnapshotVO> sc = _usageVMSnapshotDao.createSearchCriteria();
+        sc.addAnd("zoneId", SearchCriteria.Op.EQ, zoneId);
+        sc.addAnd("accountId", SearchCriteria.Op.EQ, accountId);
+        sc.addAnd("domainId", SearchCriteria.Op.EQ, domainId);
+        sc.addAnd("vmId", SearchCriteria.Op.EQ, vmId);
+        if (volumeId != null) {
+            sc.addAnd("volumeId", SearchCriteria.Op.EQ, volumeId);
+        }
+        sc.addAnd("vmSnapshotId", SearchCriteria.Op.EQ, vmSnapshotId);
+        sc.addAnd("processed", SearchCriteria.Op.NULL);
+        return _usageVMSnapshotDao.search(sc, null);
+    }
+
     protected List<UsageVMSnapshotVO> findUsageVMSnapshots(long accountId, long zoneId, long domainId, long vmId, Long diskOfferingId) {
         SearchCriteria<UsageVMSnapshotVO> sc = _usageVMSnapshotDao.createSearchCriteria();
         sc.addAnd("zoneId", SearchCriteria.Op.EQ, zoneId);

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

@@ -18,9 +18,7 @@
 
 import java.text.DecimalFormat;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 import javax.inject.Inject;
 
@@ -60,51 +58,29 @@ protected boolean parse(AccountVO account, Date startDate, Date endDate) {
             return true;
         }
 
-        Map<String, UsageVMSnapshotVO> unprocessedUsage = new HashMap<String, UsageVMSnapshotVO>();
         for (UsageVMSnapshotVO usageRec : usageUsageVMSnapshots) {
-            long zoneId = usageRec.getZoneId();
-            Long volId = usageRec.getVolumeId();
-            long vmId = usageRec.getVmId();
-            String key = vmId + ":" + volId;
-            if (usageRec.getCreated().before(startDate)) {
-                unprocessedUsage.put(key, usageRec);
-                continue;
-            }
-            UsageVMSnapshotVO previousEvent = usageVMSnapshotDao.getPreviousUsageRecord(usageRec);
-            if (previousEvent == null || previousEvent.getSize() == 0) {
-                unprocessedUsage.put(key, usageRec);
-                continue;
+            // Every usage_vmsnapshot row represents one volume of one Instance Snapshot. Instance Snapshots of the same
+            // Instance/volume coexist, so each row accrues usage independently, from its creation until it is deleted
+            // ("processed" is set by the delete event) or until the end of the aggregation period.
+            Date effectiveStart = usageRec.getCreated();
+            if (effectiveStart.before(startDate)) {
+                effectiveStart = startDate;
             }
 
-            Date previousCreated = previousEvent.getCreated();
-            if (previousCreated.before(startDate)) {
-                previousCreated = startDate;
+            Date effectiveEnd = endDate;
+            Date deleted = usageRec.getProcessed();
+            if (deleted != null && deleted.before(endDate)) {
+                effectiveEnd = deleted;
             }
 
-            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(), usageRec.getVmSnapshotId());
-            previousEvent.setProcessed(new Date());
-            usageVMSnapshotDao.update(previousEvent);
-
-            if (usageRec.getSize() == 0) {
-                usageRec.setProcessed(new Date());
-                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;
+            if (effectiveEnd.before(effectiveStart)) {
+                logger.debug("Skipping Instance Snapshot usage row [{}] as it was deleted before the period [{} - {}].", usageRec, startDate, endDate);
+                continue;
             }
-            long duration = (endDate.getTime() - created.getTime()) + 1;
-            createUsageRecord(UsageTypes.VM_SNAPSHOT, duration, created, endDate, account, usageRec.getVolumeId(), usageRec.getZoneId(), usageRec.getDiskOfferingId(),
-                usageRec.getVmId(), usageRec.getSize(), usageRec.getVmSnapshotId());
+
+            long duration = (effectiveEnd.getTime() - effectiveStart.getTime()) + 1;
+            createUsageRecord(UsageTypes.VM_SNAPSHOT, duration, effectiveStart, effectiveEnd, account, usageRec.getVolumeId(), usageRec.getZoneId(),
+                    usageRec.getDiskOfferingId(), usageRec.getVmId(), usageRec.getSize(), usageRec.getVmSnapshotId());
         }
 
         return true;

usage/src/test/java/com/cloud/usage/UsageManagerImplTest.java

@@ -32,6 +32,7 @@
 import org.mockito.Spy;
 
 import com.cloud.event.EventTypes;
+import com.cloud.event.UsageEventDetailsVO;
 import com.cloud.event.UsageEventVO;
 import com.cloud.usage.dao.UsageVPNUserDao;
 import com.cloud.user.AccountVO;
@@ -217,6 +218,27 @@ public void deleteUsageVMSnapshotMultipleSnapshotsFoundTest() {
         Mockito.verify(usageVMSnapshotDaoMock, Mockito.times(3)).update(Mockito.any(UsageVMSnapshotVO.class));
     }
 
+    @Test
+    public void deleteUsageVMSnapshotOnlyClosesTheDeletedSnapshotTest() {
+        long vmSnapshotId = 5L;
+        long volumeId = 7L;
+        UsageEventDetailsVO detail = Mockito.mock(UsageEventDetailsVO.class);
+        Mockito.doReturn(String.valueOf(vmSnapshotId)).when(detail).getValue();
+        Mockito.doReturn(detail).when(usageEventDetailsDao).findDetail(Mockito.anyLong(), Mockito.eq(UsageEventVO.DynamicParameters.vmSnapshotId.name()));
+        Mockito.doReturn(volumeId).when(usageEventVOMock).getTemplateId();
+
+        List<UsageVMSnapshotVO> vmSnapshotsMock = new ArrayList<UsageVMSnapshotVO>();
+        vmSnapshotsMock.add(vmSnapshotMock);
+        Mockito.doReturn(vmSnapshotsMock).when(usageManagerImpl).findUsageVMSnapshots(Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong(),
+                Mockito.eq(volumeId), Mockito.eq(vmSnapshotId));
+        Mockito.doNothing().when(usageVMSnapshotDaoMock).update(Mockito.any(UsageVMSnapshotVO.class));
+
+        usageManagerImpl.deleteUsageVMSnapshot(usageEventVOMock);
+
+        Mockito.verify(usageManagerImpl, Mockito.never()).findUsageVMSnapshots(Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong(), Mockito.anyLong(), Mockito.any());
+        Mockito.verify(usageVMSnapshotDaoMock, Mockito.times(1)).update(vmSnapshotMock);
+    }
+
     @Test
     public void handleVMSnapshotEventTestCreateVMSnapshot() {
         Mockito.when(usageEventVOMock.getType()).thenReturn(EventTypes.EVENT_VM_SNAPSHOT_CREATE);

usage/src/test/java/com/cloud/usage/parser/VMSnapshotUsageParserTest.java

@@ -0,0 +1,171 @@
+// 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.parser;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.cloudstack.usage.UsageTypes;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+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;
+
+@RunWith(MockitoJUnitRunner.class)
+public class VMSnapshotUsageParserTest {
+
+    private static final long HOUR = 60L * 60L * 1000L;
+    private static final long ACCOUNT_ID = 2L;
+    private static final long DOMAIN_ID = 1L;
+    private static final long ZONE_ID = 1L;
+    private static final long VM_ID = 10L;
+    private static final long VOLUME_ID = 100L;
+
+    @Mock
+    private UsageVMSnapshotDao usageVMSnapshotDao;
+
+    @Mock
+    private UsageDao usageDao;
+
+    @Mock
+    private AccountVO account;
+
+    @InjectMocks
+    private VMSnapshotUsageParser parser = new VMSnapshotUsageParser();
+
+    private final Date startDate = new Date(1_700_000_000_000L);
+    private final Date endDate = new Date(startDate.getTime() + 24 * HOUR - 1);
+
+    @Before
+    public void setUp() {
+        parser.usageDao = usageDao;
+        Mockito.when(account.getId()).thenReturn(ACCOUNT_ID);
+        Mockito.when(account.getDomainId()).thenReturn(DOMAIN_ID);
+    }
+
+    private UsageVMSnapshotVO row(long vmSnapshotId, long size, Date created, Date processed) {
+        UsageVMSnapshotVO vo = new UsageVMSnapshotVO(VOLUME_ID, ZONE_ID, ACCOUNT_ID, DOMAIN_ID, VM_ID, null, size, created, processed);
+        vo.setVmSnapshotId(vmSnapshotId);
+        return vo;
+    }
+
+    private Map<Long, UsageVO> parseAndCapture(List<UsageVMSnapshotVO> rows, int expectedRecords) {
+        Mockito.when(usageVMSnapshotDao.getUsageRecords(ACCOUNT_ID, DOMAIN_ID, startDate, endDate)).thenReturn(rows);
+
+        Assert.assertTrue(parser.parse(account, startDate, endDate));
+
+        ArgumentCaptor<UsageVO> captor = ArgumentCaptor.forClass(UsageVO.class);
+        Mockito.verify(usageDao, Mockito.times(expectedRecords)).persist(captor.capture());
+        Map<Long, UsageVO> bySnapshot = new HashMap<>();
+        for (UsageVO usage : captor.getAllValues()) {
+            Assert.assertEquals(UsageTypes.VM_SNAPSHOT, usage.getUsageType());
+            Assert.assertNull("Snapshot billed more than once in the same period", bySnapshot.put(usage.getUsageId(), usage));
+        }
+        // The parser must never consume ("process") rows; that column is reserved for the delete event.
+        Mockito.verify(usageVMSnapshotDao, Mockito.never()).update(Mockito.any(UsageVMSnapshotVO.class));
+        return bySnapshot;
+    }
+
+    private static double hours(UsageVO usage) {
+        return usage.getRawUsage();
+    }
+
+    @Test
+    public void twoSnapshotsCreatedInSamePeriodAreBilledIndependently() {
+        UsageVMSnapshotVO snap1 = row(1L, 1000L, new Date(startDate.getTime() + 2 * HOUR), null);
+        UsageVMSnapshotVO snap2 = row(2L, 2000L, new Date(startDate.getTime() + 5 * HOUR), null);
+
+        Map<Long, UsageVO> usages = parseAndCapture(Arrays.asList(snap1, snap2), 2);
+
+        UsageVO u1 = usages.get(1L);
+        Assert.assertEquals(1000L, u1.getSize().longValue());
+        Assert.assertEquals(snap1.getCreated(), u1.getStartDate());
+        Assert.assertEquals(endDate, u1.getEndDate());
+        Assert.assertEquals(22.0, hours(u1), 0.001);
+
+        UsageVO u2 = usages.get(2L);
+        Assert.assertEquals(2000L, u2.getSize().longValue());
+        Assert.assertEquals(snap2.getCreated(), u2.getStartDate());
+        Assert.assertEquals(19.0, hours(u2), 0.001);
+    }
+
+    @Test
+    public void olderSnapshotKeepsAccruingAfterNewerOneIsCreatedInLaterPeriod() {
+        UsageVMSnapshotVO snap1 = row(1L, 1000L, new Date(startDate.getTime() - 30 * HOUR), null);
+        UsageVMSnapshotVO snap2 = row(2L, 2000L, new Date(startDate.getTime() + 6 * HOUR), null);
+
+        Map<Long, UsageVO> usages = parseAndCapture(Arrays.asList(snap1, snap2), 2);
+
+        Assert.assertEquals(startDate, usages.get(1L).getStartDate());
+        Assert.assertEquals(24.0, hours(usages.get(1L)), 0.001);
+        Assert.assertEquals(18.0, hours(usages.get(2L)), 0.001);
+    }
+
+    @Test
+    public void deletedSnapshotIsBilledUntilDeletion() {
+        Date deleted = new Date(startDate.getTime() + 4 * HOUR);
+        UsageVMSnapshotVO snap1 = row(1L, 1000L, new Date(startDate.getTime() - 10 * HOUR), deleted);
+        UsageVMSnapshotVO snap2 = row(2L, 2000L, new Date(startDate.getTime() - 5 * HOUR), null);
+
+        Map<Long, UsageVO> usages = parseAndCapture(Arrays.asList(snap1, snap2), 2);
+
+        Assert.assertEquals(deleted, usages.get(1L).getEndDate());
+        Assert.assertEquals(4.0, hours(usages.get(1L)), 0.001);
+        Assert.assertEquals(24.0, hours(usages.get(2L)), 0.001);
+    }
+
+    @Test
+    public void snapshotCreatedAndDeletedInSamePeriod() {
+        Date created = new Date(startDate.getTime() + 3 * HOUR);
+        Date deleted = new Date(startDate.getTime() + 8 * HOUR);
+
+        Map<Long, UsageVO> usages = parseAndCapture(Collections.singletonList(row(1L, 1000L, created, deleted)), 1);
+
+        Assert.assertEquals(5.0, hours(usages.get(1L)), 0.001);
+    }
+
+    @Test
+    public void snapshotDeletedBeforePeriodIsNotBilled() {
+        UsageVMSnapshotVO snap = row(1L, 1000L, new Date(startDate.getTime() - 10 * HOUR), new Date(startDate.getTime() - 2 * HOUR));
+
+        parseAndCapture(Collections.singletonList(snap), 0);
+    }
+
+    @Test
+    public void noRecords() {
+        Mockito.when(usageVMSnapshotDao.getUsageRecords(ACCOUNT_ID, DOMAIN_ID, startDate, endDate)).thenReturn(Collections.emptyList());
+
+        Assert.assertTrue(parser.parse(account, startDate, endDate));
+
+        Mockito.verify(usageDao, Mockito.never()).persist(Mockito.any(UsageVO.class));
+    }
+}