Bound PBS backup polling memory

Refs #1524

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -36,12 +36,16 @@ PBS backup snapshot refresh is a bounded monitoring hot path: group-level
 snapshot fetches must run through the fixed worker pool in
 `internal/monitoring/monitor_backups.go`, reuse cached snapshots on per-group
 fetch failures, and must not allocate one goroutine or buffered result slot per
-backup group in large PBS datastores. PBS backup group cache metadata must be
-pruned to the retained group set after a completed poll, while preserving cache
-metadata only for groups still observed or intentionally reused after transient
-datastore failures. Recovery-point ingestion started by backup polling must be
-serialized and coalesced so slow store writes cannot retain one full backup
-point batch per poll cycle.
+backup group in large PBS datastores. Live PBS backup state is intentionally
+bounded: groups are processed newest-first, large groups are represented by
+their newest group metadata instead of every snapshot, per-group snapshots are
+capped to the newest bounded set, and the per-instance PBS backup list must not
+grow without an explicit monitoring-owned limit. PBS backup group cache metadata
+must be pruned to the retained group set after a completed poll, while
+preserving cache metadata only for groups still observed or intentionally reused
+after transient datastore failures. Recovery-point ingestion started by backup
+polling must be serialized and coalesced so slow store writes cannot retain one
+full backup point batch per poll cycle.
 Removed host-agent reconnect blocks are identity-scoped: matching may use the
 canonical host ID or token-qualified machine/hostname continuity, but must never
 block a distinct live host by hostname alone.

internal/monitoring/monitor_backups.go

@@ -20,7 +20,11 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-const pbsBackupSnapshotFetchWorkers = 5
+const (
+	pbsBackupSnapshotFetchWorkers   = 5
+	pbsBackupSnapshotsPerGroupLimit = 8
+	pbsBackupLiveStateLimit         = 5000
+)
 
 func pveBackupTemplateSubjectKey(instance, guestType, node string, vmid int) string {
 	return alerts.BuildBackupPVETemplateSubjectKey(instance, guestType, node, vmid)
@@ -1412,9 +1416,21 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
 			}
 
 			datastoreHadSuccess = true
+			sortPBSBackupGroupsByLatest(groups)
 			requests := make([]pbsBackupFetchRequest, 0, len(groups))
+			projectedBackups := len(allBackups)
 
 			for _, group := range groups {
+				if projectedBackups >= pbsBackupLiveStateLimit {
+					log.Warn().
+						Str("instance", instanceName).
+						Str("datastore", ds.Name).
+						Str("namespace", namespace).
+						Int("limit", pbsBackupLiveStateLimit).
+						Msg("PBS backup live-state limit reached; skipping remaining groups")
+					break
+				}
+
 				key := pbsBackupGroupKey{
 					datastore:  ds.Name,
 					namespace:  namespace,
@@ -1431,6 +1447,7 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
 
 				lastBackupTime := time.Unix(group.LastBackup, 0)
 				hasCachedData := len(cached.snapshots) > 0
+				cacheCountMatches := len(cached.snapshots) == expectedPBSBackupCacheSize(group.BackupCount)
 
 				// Check if the cached data is still within its TTL.
 				cacheAge := time.Since(m.pbsBackupCacheTimeFor(instanceName, key))
@@ -1440,11 +1457,21 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
 				// is newer, or the cache TTL has expired (to pick up verification changes).
 				if hasCachedData &&
 					cacheStillFresh &&
-					len(cached.snapshots) == group.BackupCount &&
+					cacheCountMatches &&
 					!lastBackupTime.After(cached.latest) {
 
-					allBackups = append(allBackups, cached.snapshots...)
+					allBackups = appendPBSBackupsWithinLimit(allBackups, cached.snapshots)
 					groupsReused++
+					projectedBackups = len(allBackups)
+					continue
+				}
+
+				if group.BackupCount > pbsBackupSnapshotsPerGroupLimit ||
+					projectedBackups+group.BackupCount > pbsBackupLiveStateLimit {
+					allBackups = appendPBSBackupWithinLimit(allBackups, pbsBackupFromGroup(instanceName, ds.Name, namespace, group))
+					m.setPBSBackupCacheTime(instanceName, key, time.Now())
+					groupsReused++
+					projectedBackups = len(allBackups)
 					continue
 				}
 
@@ -1454,6 +1481,7 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
 					group:     group,
 					cached:    cached,
 				})
+				projectedBackups += group.BackupCount
 			}
 
 			if len(requests) == 0 {
@@ -1463,7 +1491,7 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
 			groupsRequested += len(requests)
 			fetched := m.fetchPBSBackupSnapshots(ctx, client, instanceName, requests)
 			if len(fetched) > 0 {
-				allBackups = append(allBackups, fetched...)
+				allBackups = appendPBSBackupsWithinLimit(allBackups, fetched)
 			}
 
 			// Record fetch time for each requested group so the TTL tracks freshness.
@@ -1511,7 +1539,7 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
 					if key.datastore != ds.Name || len(entry.snapshots) == 0 {
 						continue
 					}
-					allBackups = append(allBackups, entry.snapshots...)
+					allBackups = appendPBSBackupsWithinLimit(allBackups, entry.snapshots)
 					retainedGroups[key] = struct{}{}
 				}
 			}
@@ -1581,6 +1609,13 @@ func (m *Monitor) buildPBSBackupCache(instanceName string) map[pbsBackupGroupKey
 		}
 		cache[key] = entry
 	}
+	for key, entry := range cache {
+		sortPBSBackupsByLatest(entry.snapshots)
+		if len(entry.snapshots) > pbsBackupSnapshotsPerGroupLimit {
+			entry.snapshots = entry.snapshots[:pbsBackupSnapshotsPerGroupLimit]
+		}
+		cache[key] = entry
+	}
 	return cache
 }
 
@@ -1733,13 +1768,20 @@ func (m *Monitor) fetchPBSBackupSnapshots(ctx context.Context, client *pbs.Clien
 		if len(backups) == 0 {
 			continue
 		}
-		combined = append(combined, backups...)
+		combined = appendPBSBackupsWithinLimit(combined, backups)
 	}
 
 	return combined
 }
 
 func convertPBSSnapshots(instanceName, datastore, namespace string, snapshots []pbs.BackupSnapshot) []models.PBSBackup {
+	sort.SliceStable(snapshots, func(i, j int) bool {
+		return snapshots[i].BackupTime > snapshots[j].BackupTime
+	})
+	if len(snapshots) > pbsBackupSnapshotsPerGroupLimit {
+		snapshots = snapshots[:pbsBackupSnapshotsPerGroupLimit]
+	}
+
 	backups := make([]models.PBSBackup, 0, len(snapshots))
 	for _, snapshot := range snapshots {
 		backupTime := time.Unix(snapshot.BackupTime, 0)
@@ -1800,6 +1842,73 @@ func convertPBSSnapshots(instanceName, datastore, namespace string, snapshots []
 	return backups
 }
 
+func sortPBSBackupGroupsByLatest(groups []pbs.BackupGroup) {
+	sort.SliceStable(groups, func(i, j int) bool {
+		if groups[i].LastBackup == groups[j].LastBackup {
+			if groups[i].BackupType == groups[j].BackupType {
+				return groups[i].BackupID < groups[j].BackupID
+			}
+			return groups[i].BackupType < groups[j].BackupType
+		}
+		return groups[i].LastBackup > groups[j].LastBackup
+	})
+}
+
+func sortPBSBackupsByLatest(backups []models.PBSBackup) {
+	sort.SliceStable(backups, func(i, j int) bool {
+		if backups[i].BackupTime.Equal(backups[j].BackupTime) {
+			return backups[i].ID < backups[j].ID
+		}
+		return backups[i].BackupTime.After(backups[j].BackupTime)
+	})
+}
+
+func expectedPBSBackupCacheSize(backupCount int) int {
+	if backupCount <= 0 {
+		return 0
+	}
+	if backupCount > pbsBackupSnapshotsPerGroupLimit {
+		return 1
+	}
+	return backupCount
+}
+
+func pbsBackupFromGroup(instanceName, datastore, namespace string, group pbs.BackupGroup) models.PBSBackup {
+	backupTime := time.Unix(group.LastBackup, 0)
+	backupID := fmt.Sprintf("pbs-%s-%s-%s-%s-%s-%d",
+		instanceName, datastore, namespace,
+		group.BackupType, group.BackupID,
+		group.LastBackup)
+
+	return models.PBSBackup{
+		ID:         backupID,
+		Instance:   instanceName,
+		Datastore:  datastore,
+		Namespace:  namespace,
+		BackupType: group.BackupType,
+		VMID:       group.BackupID,
+		BackupTime: backupTime,
+	}
+}
+
+func appendPBSBackupWithinLimit(backups []models.PBSBackup, backup models.PBSBackup) []models.PBSBackup {
+	if len(backups) >= pbsBackupLiveStateLimit {
+		return backups
+	}
+	return append(backups, backup)
+}
+
+func appendPBSBackupsWithinLimit(backups []models.PBSBackup, additions []models.PBSBackup) []models.PBSBackup {
+	if len(backups) >= pbsBackupLiveStateLimit || len(additions) == 0 {
+		return backups
+	}
+	remaining := pbsBackupLiveStateLimit - len(backups)
+	if len(additions) > remaining {
+		additions = additions[:remaining]
+	}
+	return append(backups, additions...)
+}
+
 // pollBackupTasks polls backup tasks from a PVE instance
 func (m *Monitor) pollBackupTasks(ctx context.Context, instanceName string, client PVEClientInterface) {
 	log.Debug().Str("instance", instanceName).Msg("polling backup tasks")

internal/monitoring/monitor_backups_readstate_test.go

@@ -230,6 +230,120 @@ func TestFetchPBSBackupSnapshotsUsesBoundedWorkerPool(t *testing.T) {
 	}
 }
 
+func TestPollPBSBackupsSummarizesLargeGroupsWithoutFetchingSnapshots(t *testing.T) {
+	t.Parallel()
+
+	var snapshotCalls int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.Contains(r.URL.Path, "/admin/datastore/archive/groups"):
+			_, _ = w.Write([]byte(fmt.Sprintf(
+				`{"data":[{"backup-type":"vm","backup-id":"100","last-backup":1700000000,"backup-count":%d}]}`,
+				pbsBackupSnapshotsPerGroupLimit+1,
+			)))
+		case strings.Contains(r.URL.Path, "/admin/datastore/archive/snapshots"):
+			atomic.AddInt64(&snapshotCalls, 1)
+			_, _ = w.Write([]byte(`{"data":[{"backup-type":"vm","backup-id":"100","backup-time":1700000000}]}`))
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer server.Close()
+
+	client, err := pbs.NewClient(pbs.ClientConfig{
+		Host:       server.URL,
+		TokenName:  "root@pam!token",
+		TokenValue: "secret",
+	})
+	if err != nil {
+		t.Fatalf("failed to create PBS client: %v", err)
+	}
+
+	m := &Monitor{state: models.NewState()}
+	m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}})
+
+	if got := atomic.LoadInt64(&snapshotCalls); got != 0 {
+		t.Fatalf("large PBS backup group fetched snapshots %d times, want 0", got)
+	}
+
+	snapshot := m.state.GetSnapshot()
+	if len(snapshot.PBSBackups) != 1 {
+		t.Fatalf("expected one summarized PBS backup artifact, got %d: %+v", len(snapshot.PBSBackups), snapshot.PBSBackups)
+	}
+	got := snapshot.PBSBackups[0]
+	if got.Instance != "pbs1" || got.Datastore != "archive" || got.BackupType != "vm" || got.VMID != "100" {
+		t.Fatalf("unexpected summarized PBS backup: %+v", got)
+	}
+	if !got.BackupTime.Equal(time.Unix(1700000000, 0)) {
+		t.Fatalf("summary backup time = %s, want %s", got.BackupTime, time.Unix(1700000000, 0))
+	}
+}
+
+func TestConvertPBSSnapshotsKeepsOnlyRecentBoundedSnapshots(t *testing.T) {
+	t.Parallel()
+
+	snapshots := make([]pbs.BackupSnapshot, 0, pbsBackupSnapshotsPerGroupLimit+3)
+	for i := 0; i < pbsBackupSnapshotsPerGroupLimit+3; i++ {
+		snapshots = append(snapshots, pbs.BackupSnapshot{
+			BackupType: "vm",
+			BackupID:   "100",
+			BackupTime: int64(1700000000 + i),
+		})
+	}
+
+	backups := convertPBSSnapshots("pbs1", "archive", "", snapshots)
+	if len(backups) != pbsBackupSnapshotsPerGroupLimit {
+		t.Fatalf("converted backups = %d, want %d", len(backups), pbsBackupSnapshotsPerGroupLimit)
+	}
+	if got, want := backups[0].BackupTime.Unix(), int64(1700000000+pbsBackupSnapshotsPerGroupLimit+2); got != want {
+		t.Fatalf("first backup time = %d, want newest %d", got, want)
+	}
+	for i := 1; i < len(backups); i++ {
+		if backups[i].BackupTime.After(backups[i-1].BackupTime) {
+			t.Fatalf("backups not sorted newest first at %d: %s after %s", i, backups[i].BackupTime, backups[i-1].BackupTime)
+		}
+	}
+}
+
+func TestBuildPBSBackupCacheKeepsNewestPerGroup(t *testing.T) {
+	t.Parallel()
+
+	backups := make([]models.PBSBackup, 0, pbsBackupSnapshotsPerGroupLimit+3)
+	for i := 0; i < pbsBackupSnapshotsPerGroupLimit+3; i++ {
+		backupTime := time.Unix(int64(1700000000+i), 0)
+		backups = append(backups, models.PBSBackup{
+			ID:         fmt.Sprintf("pbs-pbs1-archive-vm-100-%d", backupTime.Unix()),
+			Instance:   "pbs1",
+			Datastore:  "archive",
+			BackupType: "vm",
+			VMID:       "100",
+			BackupTime: backupTime,
+		})
+	}
+
+	state := models.NewState()
+	state.UpdatePBSBackups("pbs1", backups)
+	m := &Monitor{state: state}
+
+	entry := m.buildPBSBackupCache("pbs1")[pbsBackupGroupKey{
+		datastore:  "archive",
+		backupType: "vm",
+		backupID:   "100",
+	}]
+
+	if len(entry.snapshots) != pbsBackupSnapshotsPerGroupLimit {
+		t.Fatalf("cached snapshots = %d, want %d", len(entry.snapshots), pbsBackupSnapshotsPerGroupLimit)
+	}
+	if got, want := entry.snapshots[0].BackupTime.Unix(), int64(1700000000+pbsBackupSnapshotsPerGroupLimit+2); got != want {
+		t.Fatalf("first cached backup time = %d, want newest %d", got, want)
+	}
+	for i := 1; i < len(entry.snapshots); i++ {
+		if entry.snapshots[i].BackupTime.After(entry.snapshots[i-1].BackupTime) {
+			t.Fatalf("cache snapshots not sorted newest first at %d: %s after %s", i, entry.snapshots[i].BackupTime, entry.snapshots[i-1].BackupTime)
+		}
+	}
+}
+
 func TestPollPBSBackupsPrunesCacheTimesForDeletedGroups(t *testing.T) {
 	t.Parallel()