diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 84655d858..2b21f4c57 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/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. diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index f432673c6..38a2dd80c 100644 --- a/internal/monitoring/monitor_backups.go +++ b/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") diff --git a/internal/monitoring/monitor_backups_readstate_test.go b/internal/monitoring/monitor_backups_readstate_test.go index 8ddf6ef4c..b1d0be043 100644 --- a/internal/monitoring/monitor_backups_readstate_test.go +++ b/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()