vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-10 00:14:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Fix PBS backup discovery regression from bounded polling

Browse source 
The RC3 memory bound for PBS backup polling summarized any group with
more than 8 snapshots into a single synthesized entry built from group
metadata. A synthesized entry has no verification, size, file, or
per-snapshot time data, so most real deployments saw every backup as
Unverified with no size, PBS files not listed, and a backup timeline
collapsed onto the latest backup day.

Keep the issue #1524 memory bounds but derive them from real data:
always fetch snapshots for stale groups, retain the newest bounded set
per group (limit raised from 8 to 100 to cover real keep policies), and
keep the newest-first global live-state cap. Remove the synthesized
group placeholder path entirely and update the monitoring subsystem
contract and tests to pin real-snapshot bounding.

Fixes #1541
Refs #1524
This commit is contained in:
rcourtman 2026-07-07 22:48:34 +01:00
parent 0ad22fe2d5
commit 292baf308b
3 changed files with 123 additions and 69 deletions
Download patch file Download diff file Expand all files Collapse all files

10
docs/release-control/v6/internal/subsystems/monitoring.md
View file

@ -37,10 +37,12 @@ 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. 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
bounded: groups are processed newest-first, per-group snapshots are capped to
the newest bounded set of real fetched snapshots, and the per-instance PBS
backup list must not grow without an explicit monitoring-owned limit. Bounding
must never synthesize placeholder backup entries from group metadata: a
placeholder drops verification, size, file, and per-snapshot time data, which
users read as broken discovery and failed verification. 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

60
internal/monitoring/monitor_backups.go
View file

@ -21,9 +21,17 @@ import (
)
const (
pbsBackupSnapshotFetchWorkers = 5
pbsBackupSnapshotsPerGroupLimit = 8
pbsBackupLiveStateLimit = 5000
pbsBackupSnapshotFetchWorkers = 5
// pbsBackupSnapshotsPerGroupLimit bounds how many real snapshots are
// retained per backup group (newest first). It must comfortably exceed
// common PBS keep policies: groups are always represented by real fetched
// snapshots, never synthesized group metadata, because a synthesized entry
// has no verification, size, file, or per-snapshot time data (issue #1541).
pbsBackupSnapshotsPerGroupLimit = 100
// pbsBackupLiveStateLimit bounds the per-instance PBS backup list as a
// whole. Groups are processed newest-first so the newest restore points
// win when the limit is hit (issue #1524).
pbsBackupLiveStateLimit = 5000
)
func pveBackupTemplateSubjectKey(instance, guestType, node string, vmid int) string {
@ -1447,7 +1455,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)
cacheCountMatches := len(cached.snapshots) == retainedPBSSnapshotCount(group.BackupCount)
// Check if the cached data is still within its TTL.
cacheAge := time.Since(m.pbsBackupCacheTimeFor(instanceName, key))
@ -1466,22 +1474,13 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien
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
}
requests = append(requests, pbsBackupFetchRequest{
datastore: ds.Name,
namespace: namespace,
group: group,
cached: cached,
})
projectedBackups += group.BackupCount
projectedBackups += retainedPBSSnapshotCount(group.BackupCount)
}
if len(requests) == 0 {
@ -1863,41 +1862,20 @@ func sortPBSBackupsByLatest(backups []models.PBSBackup) {
})
}
func expectedPBSBackupCacheSize(backupCount int) int {
// retainedPBSSnapshotCount is how many snapshots poll retention keeps for a
// group with the given backup count: the full group, capped at the per-group
// limit. It doubles as the expected cache size when deciding whether cached
// snapshots for a group are still complete.
func retainedPBSSnapshotCount(backupCount int) int {
if backupCount <= 0 {
return 0
}
if backupCount > pbsBackupSnapshotsPerGroupLimit {
return 1
return pbsBackupSnapshotsPerGroupLimit
}
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

122
internal/monitoring/monitor_backups_readstate_test.go
View file

@ -230,20 +230,44 @@ func TestFetchPBSBackupSnapshotsUsesBoundedWorkerPool(t *testing.T) {
}
}
func TestPollPBSBackupsSummarizesLargeGroupsWithoutFetchingSnapshots(t *testing.T) {
// Regression test for issue #1541: groups larger than the per-group limit
// must still be fetched for real, keeping verification, size, file, and
// per-snapshot time data for the newest bounded set. RC3 summarized such
// groups into a single synthesized entry, which surfaced as "Unverified",
// "No size", "PBS files not listed", and a collapsed backup timeline.
func TestPollPBSBackupsFetchesRealSnapshotsForLargeGroups(t *testing.T) {
t.Parallel()
const firstBackupTime = int64(1700000000)
snapshotCount := pbsBackupSnapshotsPerGroupLimit + 3
var snapshots strings.Builder
snapshots.WriteString(`{"data":[`)
for i := 0; i < snapshotCount; i++ {
if i > 0 {
snapshots.WriteByte(',')
}
_, _ = fmt.Fprintf(
&snapshots,
`{"backup-type":"vm","backup-id":"100","backup-time":%d,"size":2048,"owner":"root@pam","files":[{"filename":"drive-scsi0.img.fidx"}],"verification":{"state":"ok","upid":"UPID:pbs1"}}`,
firstBackupTime+int64(i),
)
}
snapshots.WriteString(`]}`)
snapshotsJSON := snapshots.String()
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,
`{"data":[{"backup-type":"vm","backup-id":"100","last-backup":%d,"backup-count":%d}]}`,
firstBackupTime+int64(snapshotCount-1),
snapshotCount,
)))
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}]}`))
_, _ = w.Write([]byte(snapshotsJSON))
default:
http.NotFound(w, r)
}
@ -262,20 +286,45 @@ func TestPollPBSBackupsSummarizesLargeGroupsWithoutFetchingSnapshots(t *testing.
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)
if got := atomic.LoadInt64(&snapshotCalls); got != 1 {
t.Fatalf("large PBS backup group fetched snapshots %d times, want 1", 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)
if len(snapshot.PBSBackups) != pbsBackupSnapshotsPerGroupLimit {
t.Fatalf("retained backups = %d, want newest %d real snapshots", len(snapshot.PBSBackups), pbsBackupSnapshotsPerGroupLimit)
}
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)
seenTimes := make(map[int64]struct{}, len(snapshot.PBSBackups))
for i, backup := range snapshot.PBSBackups {
if !backup.Verified {
t.Fatalf("backup %d lost verification state: %+v", i, backup)
}
if backup.Size != 2048 {
t.Fatalf("backup %d lost size: %+v", i, backup)
}
if len(backup.Files) != 1 || backup.Files[0] != "drive-scsi0.img.fidx" {
t.Fatalf("backup %d lost file list: %+v", i, backup)
}
seenTimes[backup.BackupTime.Unix()] = struct{}{}
}
if !got.BackupTime.Equal(time.Unix(1700000000, 0)) {
t.Fatalf("summary backup time = %s, want %s", got.BackupTime, time.Unix(1700000000, 0))
if len(seenTimes) != pbsBackupSnapshotsPerGroupLimit {
t.Fatalf("backup times collapsed: %d distinct times, want %d", len(seenTimes), pbsBackupSnapshotsPerGroupLimit)
}
newestTime := firstBackupTime + int64(snapshotCount-1)
oldestRetainedTime := newestTime - int64(pbsBackupSnapshotsPerGroupLimit-1)
for want := oldestRetainedTime; want <= newestTime; want++ {
if _, ok := seenTimes[want]; !ok {
t.Fatalf("expected newest snapshots retained, missing backup time %d", want)
}
}
// A second poll must reuse the bounded cache instead of refetching.
m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}})
if got := atomic.LoadInt64(&snapshotCalls); got != 1 {
t.Fatalf("second poll refetched snapshots (%d calls), want cached reuse", got)
}
if got := len(m.state.GetSnapshot().PBSBackups); got != pbsBackupSnapshotsPerGroupLimit {
t.Fatalf("retained backups after cached poll = %d, want %d", got, pbsBackupSnapshotsPerGroupLimit)
}
}
@ -292,10 +341,9 @@ func TestPollPBSBackupsBoundsLargeTopologyAcrossPolls(t *testing.T) {
}
_, _ = fmt.Fprintf(
&groups,
`{"backup-type":"vm","backup-id":"%d","last-backup":%d,"backup-count":%d}`,
`{"backup-type":"vm","backup-id":"%d","last-backup":%d,"backup-count":1}`,
i,
firstBackupTime+int64(i),
pbsBackupSnapshotsPerGroupLimit+20,
)
}
groups.WriteString(`]}`)
@ -308,7 +356,17 @@ func TestPollPBSBackupsBoundsLargeTopologyAcrossPolls(t *testing.T) {
_, _ = w.Write([]byte(groupsJSON))
case strings.Contains(r.URL.Path, "/admin/datastore/archive/snapshots"):
atomic.AddInt64(&snapshotCalls, 1)
_, _ = w.Write([]byte(`{"data":[]}`))
backupID := r.URL.Query().Get("backup-id")
id, err := strconv.Atoi(backupID)
if err != nil {
http.Error(w, "bad backup-id", http.StatusBadRequest)
return
}
_, _ = w.Write([]byte(fmt.Sprintf(
`{"data":[{"backup-type":"vm","backup-id":%q,"backup-time":%d,"size":1024,"verification":{"state":"ok"}}]}`,
backupID,
firstBackupTime+int64(id),
)))
default:
http.NotFound(w, r)
}
@ -332,16 +390,32 @@ func TestPollPBSBackupsBoundsLargeTopologyAcrossPolls(t *testing.T) {
if got := len(snapshot.PBSBackups); got != pbsBackupLiveStateLimit {
t.Fatalf("cycle %d PBS backup state size = %d, want %d", cycle, got, pbsBackupLiveStateLimit)
}
if got := atomic.LoadInt64(&snapshotCalls); got != 0 {
t.Fatalf("cycle %d large topology fetched snapshots %d times, want 0", cycle, got)
// Snapshot fetches must stay bounded by the live-state limit: newest
// groups are fetched once, groups beyond the limit are never fetched,
// and later cycles reuse the bounded cache without refetching.
if got := atomic.LoadInt64(&snapshotCalls); got != int64(pbsBackupLiveStateLimit) {
t.Fatalf("cycle %d cumulative snapshot fetches = %d, want %d", cycle, got, pbsBackupLiveStateLimit)
}
newest := snapshot.PBSBackups[0]
if newest.VMID != strconv.Itoa(groupCount-1) || newest.BackupTime.Unix() != firstBackupTime+int64(groupCount-1) {
t.Fatalf("cycle %d newest backup = %+v, want vmid %d at %d", cycle, newest, groupCount-1, firstBackupTime+int64(groupCount-1))
times := make(map[int64]struct{}, len(snapshot.PBSBackups))
for _, backup := range snapshot.PBSBackups {
if !backup.Verified {
t.Fatalf("cycle %d backup lost verification state: %+v", cycle, backup)
}
times[backup.BackupTime.Unix()] = struct{}{}
}
oldestRetained := snapshot.PBSBackups[len(snapshot.PBSBackups)-1]
if oldestRetained.VMID != strconv.Itoa(groupCount-pbsBackupLiveStateLimit) {
t.Fatalf("cycle %d oldest retained vmid = %s, want %d", cycle, oldestRetained.VMID, groupCount-pbsBackupLiveStateLimit)
if len(times) != pbsBackupLiveStateLimit {
t.Fatalf("cycle %d distinct backup times = %d, want %d", cycle, len(times), pbsBackupLiveStateLimit)
}
newestTime := firstBackupTime + int64(groupCount-1)
oldestRetainedTime := firstBackupTime + int64(groupCount-pbsBackupLiveStateLimit)
if _, ok := times[newestTime]; !ok {
t.Fatalf("cycle %d missing newest backup time %d", cycle, newestTime)
}
if _, ok := times[oldestRetainedTime]; !ok {
t.Fatalf("cycle %d missing oldest retained backup time %d", cycle, oldestRetainedTime)
}
if _, ok := times[oldestRetainedTime-1]; ok {
t.Fatalf("cycle %d retained backup older than the live-state window: %d", cycle, oldestRetainedTime-1)
}
}
}