Backup usage: bill each size for the time it applied, and publish the metric only on change

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

@@ -24,7 +24,8 @@
 import com.cloud.utils.db.GenericDao;
 
 public interface UsageBackupDao extends GenericDao<UsageBackupVO, Long> {
-    void updateMetrics(Long vmId, Long backupOfferingId, Long size, Long virtualSize);
+    List<UsageBackupVO> listActiveUsage(Long vmId, Long backupOfferingId);
+    void updateMetrics(Long vmId, Long backupOfferingId, Long size, Long virtualSize, Date eventDate);
     void removeUsage(Long accountId, Long vmId, Long backupOfferingId, Date eventDate);
     List<UsageBackupVO> getUsageRecords(Long accountId, Date startDate, Date endDate);
 }

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

@@ -30,6 +30,7 @@
 import com.cloud.exception.CloudException;
 import com.cloud.usage.UsageBackupVO;
 import com.cloud.utils.DateUtil;
+import com.cloud.utils.db.Filter;
 import com.cloud.utils.db.GenericDaoBase;
 import com.cloud.utils.db.SearchCriteria;
 import com.cloud.utils.db.TransactionLegacy;
@@ -42,19 +43,50 @@ public class UsageBackupDaoImpl extends GenericDaoBase<UsageBackupVO, Long> impl
             " OR ((created <= ?) AND (removed >= ?)))";
 
     @Override
-    public void updateMetrics(final Long vmId, Long backupOfferingId, final Long size, final Long virtualSize) {
-        try (TransactionLegacy txn = TransactionLegacy.open(TransactionLegacy.USAGE_DB)) {
-            SearchCriteria<UsageBackupVO> sc = this.createSearchCriteria();
-            sc.addAnd("vmId", SearchCriteria.Op.EQ, vmId);
-            sc.addAnd("backupOfferingId", SearchCriteria.Op.EQ, backupOfferingId);
-            UsageBackupVO vo = findOneBy(sc);
-            if (vo != null) {
-                vo.setSize(size);
-                vo.setProtectedSize(virtualSize);
-                update(vo.getId(), vo);
+    public List<UsageBackupVO> listActiveUsage(Long vmId, Long backupOfferingId) {
+        SearchCriteria<UsageBackupVO> sc = this.createSearchCriteria();
+        sc.addAnd("vmId", SearchCriteria.Op.EQ, vmId);
+        sc.addAnd("backupOfferingId", SearchCriteria.Op.EQ, backupOfferingId);
+        sc.addAnd("removed", SearchCriteria.Op.NULL);
+        return listBy(sc, new Filter(UsageBackupVO.class, "created", false));
+    }
+
+    @Override
+    public void updateMetrics(final Long vmId, final Long backupOfferingId, final Long size, final Long virtualSize, final Date eventDate) {
+        final long newSize = size != null ? size : 0L;
+        final long newProtectedSize = virtualSize != null ? virtualSize : 0L;
+        TransactionLegacy txn = TransactionLegacy.open(TransactionLegacy.USAGE_DB);
+        try {
+            txn.start();
+            List<UsageBackupVO> activeUsage = listActiveUsage(vmId, backupOfferingId);
+            if (activeUsage.isEmpty()) {
+                logger.warn("No active backup usage for VM [{}] and backup offering [{}], ignoring backup metrics of size [{}] and protected size [{}].",
+                        vmId, backupOfferingId, newSize, newProtectedSize);
+                txn.commit();
+                return;
+            }
+
+            UsageBackupVO latest = activeUsage.get(0);
+            if (activeUsage.size() == 1 && latest.getSize() == newSize && latest.getProtectedSize() == newProtectedSize) {
+                txn.commit();
+                return;
+            }
+
+            // Close the active rows and open one with the new size; this also merges duplicates.
+            for (UsageBackupVO usage : activeUsage) {
+                usage.setRemoved(eventDate);
+                update(usage.getId(), usage);
             }
+            UsageBackupVO newUsage = new UsageBackupVO(latest.getZoneId(), latest.getAccountId(), latest.getDomainId(), vmId, backupOfferingId, eventDate);
+            newUsage.setSize(newSize);
+            newUsage.setProtectedSize(newProtectedSize);
+            persist(newUsage);
+            txn.commit();
         } catch (final Exception e) {
+            txn.rollback();
             logger.error("Error updating backup metrics: " + e.getMessage(), e);
+        } finally {
+            txn.close();
         }
     }
 

engine/schema/src/main/java/org/apache/cloudstack/backup/BackupUsageMetricVO.java

@@ -0,0 +1,107 @@
+// 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 org.apache.cloudstack.backup;
+
+import java.util.Date;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import javax.persistence.Temporal;
+import javax.persistence.TemporalType;
+
+import org.apache.cloudstack.api.InternalIdentity;
+
+/**
+ * The backup usage metric last published for a VM and backup offering.
+ */
+@Entity
+@Table(name = "backup_usage_metric")
+public class BackupUsageMetricVO implements InternalIdentity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    @Column(name = "id")
+    private long id;
+
+    @Column(name = "vm_id")
+    private long vmId;
+
+    @Column(name = "backup_offering_id")
+    private long backupOfferingId;
+
+    @Column(name = "size")
+    private long size;
+
+    @Column(name = "protected_size")
+    private long protectedSize;
+
+    @Column(name = "updated")
+    @Temporal(value = TemporalType.TIMESTAMP)
+    private Date updated;
+
+    protected BackupUsageMetricVO() {
+    }
+
+    public BackupUsageMetricVO(long vmId, long backupOfferingId, long size, long protectedSize, Date updated) {
+        this.vmId = vmId;
+        this.backupOfferingId = backupOfferingId;
+        this.size = size;
+        this.protectedSize = protectedSize;
+        this.updated = updated;
+    }
+
+    @Override
+    public long getId() {
+        return id;
+    }
+
+    public long getVmId() {
+        return vmId;
+    }
+
+    public long getBackupOfferingId() {
+        return backupOfferingId;
+    }
+
+    public long getSize() {
+        return size;
+    }
+
+    public void setSize(long size) {
+        this.size = size;
+    }
+
+    public long getProtectedSize() {
+        return protectedSize;
+    }
+
+    public void setProtectedSize(long protectedSize) {
+        this.protectedSize = protectedSize;
+    }
+
+    public Date getUpdated() {
+        return updated;
+    }
+
+    public void setUpdated(Date updated) {
+        this.updated = updated;
+    }
+}

engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupUsageMetricDao.java

@@ -0,0 +1,27 @@
+// 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 org.apache.cloudstack.backup.dao;
+
+import org.apache.cloudstack.backup.BackupUsageMetricVO;
+
+import com.cloud.utils.db.GenericDao;
+
+public interface BackupUsageMetricDao extends GenericDao<BackupUsageMetricVO, Long> {
+    BackupUsageMetricVO findByVmAndOffering(long vmId, long backupOfferingId);
+
+    int removeByVmAndOffering(long vmId, long backupOfferingId);
+}

engine/schema/src/main/java/org/apache/cloudstack/backup/dao/BackupUsageMetricDaoImpl.java

@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.cloudstack.backup.dao;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.cloudstack.backup.BackupUsageMetricVO;
+
+import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+
+public class BackupUsageMetricDaoImpl extends GenericDaoBase<BackupUsageMetricVO, Long> implements BackupUsageMetricDao {
+    private SearchBuilder<BackupUsageMetricVO> vmAndOfferingSearch;
+
+    @PostConstruct
+    protected void init() {
+        vmAndOfferingSearch = createSearchBuilder();
+        vmAndOfferingSearch.and("vmId", vmAndOfferingSearch.entity().getVmId(), SearchCriteria.Op.EQ);
+        vmAndOfferingSearch.and("backupOfferingId", vmAndOfferingSearch.entity().getBackupOfferingId(), SearchCriteria.Op.EQ);
+        vmAndOfferingSearch.done();
+    }
+
+    private SearchCriteria<BackupUsageMetricVO> createVmAndOfferingCriteria(long vmId, long backupOfferingId) {
+        SearchCriteria<BackupUsageMetricVO> sc = vmAndOfferingSearch.create();
+        sc.setParameters("vmId", vmId);
+        sc.setParameters("backupOfferingId", backupOfferingId);
+        return sc;
+    }
+
+    @Override
+    public BackupUsageMetricVO findByVmAndOffering(long vmId, long backupOfferingId) {
+        return findOneBy(createVmAndOfferingCriteria(vmId, backupOfferingId));
+    }
+
+    @Override
+    public int removeByVmAndOffering(long vmId, long backupOfferingId) {
+        return remove(createVmAndOfferingCriteria(vmId, backupOfferingId));
+    }
+}

engine/schema/src/main/resources/META-INF/cloudstack/core/spring-engine-schema-core-daos-context.xml

@@ -273,6 +273,7 @@
   <bean id="backupDaoImpl" class="org.apache.cloudstack.backup.dao.BackupDaoImpl" />
   <bean id="backupDetailsDaoImpl" class="org.apache.cloudstack.backup.dao.BackupDetailsDaoImpl" />
   <bean id="backupRepositoryDaoImpl" class="org.apache.cloudstack.backup.dao.BackupRepositoryDaoImpl" />
+  <bean id="backupUsageMetricDaoImpl" class="org.apache.cloudstack.backup.dao.BackupUsageMetricDaoImpl" />
   <bean id="directDownloadCertificateDaoImpl" class="org.apache.cloudstack.direct.download.DirectDownloadCertificateDaoImpl" />
   <bean id="directDownloadCertificateHostMapDaoImpl" class="org.apache.cloudstack.direct.download.DirectDownloadCertificateHostMapDaoImpl" />
   <bean id="routerHealthCheckResultsDaoImpl" class="com.cloud.network.dao.RouterHealthCheckResultDaoImpl" />

engine/schema/src/main/resources/META-INF/db/schema-42210to42220.sql

@@ -18,3 +18,15 @@
 --;
 -- Schema upgrade from 4.22.1.0 to 4.22.2.0
 --;
+
+-- Last backup usage metric published per VM and backup offering
+CREATE TABLE IF NOT EXISTS `cloud`.`backup_usage_metric` (
+    `id` bigint unsigned NOT NULL auto_increment COMMENT 'id',
+    `vm_id` bigint unsigned NOT NULL COMMENT 'VM ID',
+    `backup_offering_id` bigint unsigned NOT NULL COMMENT 'Backup offering ID',
+    `size` bigint unsigned NOT NULL COMMENT 'Backup size last published',
+    `protected_size` bigint unsigned NOT NULL COMMENT 'Protected size last published',
+    `updated` datetime NOT NULL COMMENT 'Date the metric was last published',
+    PRIMARY KEY (`id`),
+    UNIQUE KEY `uk_backup_usage_metric__vm_id__backup_offering_id` (`vm_id`, `backup_offering_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8;