Backup: publish the backup usage metric only when it changes

The backup sync published a BACKUP.USAGE.METRIC event for every VM on every
run (every 5 minutes by default) even when nothing changed, which made up
most of the usage_event rows. Remember the last value published per VM and
offering and skip unchanged ones. The usage server ignores unchanged sizes,
so republishing after a restart or on another management server is harmless.

Author: abh1sar

server/src/main/java/org/apache/cloudstack/backup/BackupManagerImpl.java

@@ -32,6 +32,7 @@
 import java.util.TimeZone;
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -249,6 +250,12 @@ public class BackupManagerImpl extends ManagerBase implements BackupManager {
     private static Map<String, BackupProvider> backupProvidersMap = new HashMap<>();
     private List<BackupProvider> backupProviders;
 
+    // Per zone, the backup usage metric last published for each (VM, backup offering) pair, as (size,
+    // protected size). The backup sync runs every few minutes but the usage server only needs the
+    // metric when it changes. After a restart or on another management server each value is
+    // published once more, which the usage server ignores as the size is unchanged.
+    private final Map<Long, Map<Pair<Long, Long>, Pair<Long, Long>>> lastPublishedBackupUsageMetrics = new ConcurrentHashMap<>();
+
     public AsyncJobDispatcher getAsyncJobDispatcher() {
         return asyncJobDispatcher;
     }
@@ -2014,7 +2021,7 @@ private void syncOutOfBandBackups(final BackupProvider backupProvider, DataCente
             }
         }
 
-        private void updateBackupUsageRecords(final BackupProvider backupProvider, DataCenter dataCenter) {
+        protected void updateBackupUsageRecords(final BackupProvider backupProvider, DataCenter dataCenter) {
             List<Long> vmIdsWithBackups = backupDao.listVmIdsWithBackupsInZone(dataCenter.getId());
             List<VMInstanceVO> vmsWithBackups;
             if (vmIdsWithBackups.size() == 0) {
@@ -2025,6 +2032,8 @@ private void updateBackupUsageRecords(final BackupProvider backupProvider, DataC
             List<VMInstanceVO> vmsWithBackupOffering = vmInstanceDao.listByZoneAndBackupOffering(dataCenter.getId(), null); //should return including removed
             Set<VMInstanceVO> vms =  Stream.concat(vmsWithBackups.stream(), vmsWithBackupOffering.stream()) .collect(Collectors.toSet());
 
+            Map<Pair<Long, Long>, Pair<Long, Long>> lastPublished = lastPublishedBackupUsageMetrics.getOrDefault(dataCenter.getId(), Collections.emptyMap());
+            Map<Pair<Long, Long>, Pair<Long, Long>> published = new HashMap<>();
             for (final VirtualMachine vm : vms) {
 
                 Map<Long, Pair<Long, Long>> backupOfferingToSizeMap = new HashMap<>();
@@ -2055,12 +2064,20 @@ private void updateBackupUsageRecords(final BackupProvider backupProvider, DataC
                     Pair<Long, Long> sizes = entry.getValue();
                     Long backupSize = sizes.first();
                     Long protectedSize = sizes.second();
+                    Pair<Long, Long> key = new Pair<>(vm.getId(), offeringId);
+                    Pair<Long, Long> metric = new Pair<>(backupSize, protectedSize);
+                    published.put(key, metric);
+                    if (metric.equals(lastPublished.get(key))) {
+                        continue;
+                    }
                     UsageEventUtils.publishUsageEvent(EventTypes.EVENT_VM_BACKUP_USAGE_METRIC, vm.getAccountId(),
                             vm.getDataCenterId(), vm.getId(), "Backup-" + vm.getHostName() + "-" + vm.getUuid(),
                             offeringId, null, backupSize, protectedSize,
                             Backup.class.getSimpleName(), vm.getUuid());
                 }
             }
+            // Replacing the map also drops pairs that are no longer reported, e.g. expunged VMs.
+            lastPublishedBackupUsageMetrics.put(dataCenter.getId(), published);
         }
 
         private Backup checkAndUpdateIfBackupEntryExistsForRestorePoint(Backup.RestorePoint restorePoint, List<Backup> backupsInDb, VirtualMachine vm) {

server/src/test/java/org/apache/cloudstack/backup/BackupManagerTest.java

@@ -898,6 +898,55 @@ public void testBackupSyncTask() {
         }
     }
 
+    private void verifyBackupUsageMetricPublished(MockedStatic<UsageEventUtils> usageEventUtilsMocked, int times, Long size, Long protectedSize) {
+        usageEventUtilsMocked.verify(() -> UsageEventUtils.publishUsageEvent(Mockito.eq(EventTypes.EVENT_VM_BACKUP_USAGE_METRIC), Mockito.anyLong(),
+                Mockito.anyLong(), Mockito.anyLong(), Mockito.any(), Mockito.anyLong(), Mockito.any(), Mockito.eq(size), Mockito.eq(protectedSize),
+                Mockito.any(), Mockito.any()), times(times));
+    }
+
+    @Test
+    public void updateBackupUsageRecordsPublishesMetricOnlyWhenItChanges() {
+        Long dataCenterId = 1L;
+        Long vmId = 2L;
+        Long offeringId = 3L;
+
+        DataCenterVO dataCenter = mock(DataCenterVO.class);
+        when(dataCenter.getId()).thenReturn(dataCenterId);
+        VMInstanceVO vm = mock(VMInstanceVO.class);
+        when(vm.getId()).thenReturn(vmId);
+        when(backupDao.listVmIdsWithBackupsInZone(dataCenterId)).thenReturn(List.of(vmId));
+        when(vmInstanceDao.listByIdsIncludingRemoved(List.of(vmId))).thenReturn(List.of(vm));
+        when(vmInstanceDao.listByZoneAndBackupOffering(dataCenterId, null)).thenReturn(List.of(vm));
+
+        BackupVO backup = new BackupVO();
+        backup.setBackupOfferingId(offeringId);
+        backup.setSize(100L);
+        backup.setProtectedSize(1000L);
+        when(backupDao.listByVmId(null, vmId)).thenReturn(List.of(backup));
+
+        BackupManagerImpl.BackupSyncTask backupSyncTask = backupManager.new BackupSyncTask(backupManager);
+        BackupProvider backupProvider = mock(BackupProvider.class);
+        try (MockedStatic<UsageEventUtils> usageEventUtilsMocked = Mockito.mockStatic(UsageEventUtils.class)) {
+            backupSyncTask.updateBackupUsageRecords(backupProvider, dataCenter);
+            backupSyncTask.updateBackupUsageRecords(backupProvider, dataCenter);
+            verifyBackupUsageMetricPublished(usageEventUtilsMocked, 1, 100L, 1000L);
+
+            backup.setSize(200L);
+            backupSyncTask.updateBackupUsageRecords(backupProvider, dataCenter);
+            backupSyncTask.updateBackupUsageRecords(backupProvider, dataCenter);
+            verifyBackupUsageMetricPublished(usageEventUtilsMocked, 1, 200L, 1000L);
+
+            // A VM that is no longer reported is forgotten, so it is published again if it comes back.
+            when(backupDao.listVmIdsWithBackupsInZone(dataCenterId)).thenReturn(List.of());
+            when(vmInstanceDao.listByZoneAndBackupOffering(dataCenterId, null)).thenReturn(List.of());
+            backupSyncTask.updateBackupUsageRecords(backupProvider, dataCenter);
+            when(backupDao.listVmIdsWithBackupsInZone(dataCenterId)).thenReturn(List.of(vmId));
+            when(vmInstanceDao.listByZoneAndBackupOffering(dataCenterId, null)).thenReturn(List.of(vm));
+            backupSyncTask.updateBackupUsageRecords(backupProvider, dataCenter);
+            verifyBackupUsageMetricPublished(usageEventUtilsMocked, 2, 200L, 1000L);
+        }
+    }
+
     @Test
     public void checkCallerAccessToBackupScheduleVmTestExecuteAccessCheckMethods() {
         long vmId = 1L;