Address review feedback: use DeleteLabelValues, extract resync method

Replace blanket Reset() with targeted DeleteLabelValues to avoid
briefly wiping metrics for schedules that still exist. Track known
schedules in a set and only delete stale entries on each resync.

Extract the wait.Until closure into resyncBackupMetrics() so tests
can call it directly without goroutine timing. Replace hand-rolled
collectGaugeCount helper with testutil.CollectAndCount.

Signed-off-by: Shubham Pampattiwar <spampatt@redhat.com>

pkg/controller/backup_controller.go

@@ -107,10 +107,11 @@ type backupReconciler struct {
 	credentialFileStore           credentials.FileStore
 	maxConcurrentK8SConnections   int
 	defaultSnapshotMoveData       bool
-	globalCRClient                kbclient.Client
-	itemBlockWorkerCount          int
-	concurrentBackups             int
-	globalVolumePoliciesConfigMap string
+	globalCRClient                        kbclient.Client
+	itemBlockWorkerCount                  int
+	concurrentBackups                     int
+	globalVolumePoliciesConfigMap         string
+	knownSchedulesWithSuccessfulBackup    sets.Set[string]
 }
 
 func NewBackupReconciler(
@@ -204,30 +205,43 @@ func (b *backupReconciler) updateTotalBackupMetric() {
 		time.Sleep(5 * time.Second)
 
 		wait.Until(
-			func() {
-				// recompute backup_total metric
-				backups := &velerov1api.BackupList{}
-				err := b.kbClient.List(context.Background(), backups, &kbclient.ListOptions{LabelSelector: labels.Everything()})
-				if err != nil {
-					b.logger.Error(err, "Error computing backup_total metric")
-				} else {
-					b.metrics.SetBackupTotal(int64(len(backups.Items)))
-
-					// recompute backup_last_successful_timestamp metric for each
-					// schedule (including the empty schedule, i.e. ad-hoc backups).
-					// Reset first to prune stale entries for deleted schedules.
-					b.metrics.ResetBackupLastSuccessfulTimestamp()
-					for schedule, timestamp := range getLastSuccessBySchedule(backups.Items) {
-						b.metrics.SetBackupLastSuccessfulTimestamp(schedule, timestamp)
-					}
-				}
-			},
+			b.resyncBackupMetrics,
 			backupResyncPeriod,
 			b.ctx.Done(),
 		)
 	}()
 }
 
+func (b *backupReconciler) resyncBackupMetrics() {
+	backups := &velerov1api.BackupList{}
+	err := b.kbClient.List(context.Background(), backups, &kbclient.ListOptions{LabelSelector: labels.Everything()})
+	if err != nil {
+		b.logger.Error(err, "Error computing backup_total metric")
+		return
+	}
+
+	b.metrics.SetBackupTotal(int64(len(backups.Items)))
+
+	currentSchedules := getLastSuccessBySchedule(backups.Items)
+	for schedule, timestamp := range currentSchedules {
+		b.metrics.SetBackupLastSuccessfulTimestamp(schedule, timestamp)
+	}
+
+	// Remove metrics for schedules that no longer have successful backups
+	if b.knownSchedulesWithSuccessfulBackup != nil {
+		for schedule := range b.knownSchedulesWithSuccessfulBackup {
+			if _, exists := currentSchedules[schedule]; !exists {
+				b.metrics.DeleteBackupLastSuccessfulTimestamp(schedule)
+			}
+		}
+	}
+
+	b.knownSchedulesWithSuccessfulBackup = sets.New[string]()
+	for schedule := range currentSchedules {
+		b.knownSchedulesWithSuccessfulBackup.Insert(schedule)
+	}
+}
+
 // getLastSuccessBySchedule finds the most recent completed backup for each schedule
 // and returns a map of schedule name -> completion time of the most recent completed
 // backup. This map includes an entry for ad-hoc/non-scheduled backups, where the key