From bf63515ff46d2f6ba1e6f53ae0479ac055dddce0 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Tue, 18 Nov 2025 23:10:43 +0000 Subject: [PATCH] feat: avoid redundant PBS snapshot polling (Related to #717) --- internal/monitoring/monitor.go | 372 ++++++++++++++++++++++++++------- 1 file changed, 296 insertions(+), 76 deletions(-) diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 807405a..8a2449e 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -9470,10 +9470,32 @@ func (m *Monitor) removeFailedPMGInstance(instanceName string) { m.state.SetConnectionHealth("pmg-"+instanceName, false) } +type pbsBackupGroupKey struct { + datastore string + namespace string + backupType string + backupID string +} + +type cachedPBSGroup struct { + snapshots []models.PBSBackup + latest time.Time +} + +type pbsBackupFetchRequest struct { + datastore string + namespace string + group pbs.BackupGroup + cached cachedPBSGroup +} + // pollPBSBackups fetches all backups from PBS datastores func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, client *pbs.Client, datastores []models.PBSDatastore) { log.Debug().Str("instance", instanceName).Msg("Polling PBS backups") + // Cache existing PBS backups so we can avoid redundant API calls when no changes occurred. + existingGroups := m.buildPBSBackupCache(instanceName) + var allBackups []models.PBSBackup datastoreCount := len(datastores) // Number of datastores to query datastoreFetches := 0 // Number of successful datastore fetches @@ -9481,12 +9503,15 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien // Process each datastore for _, ds := range datastores { - // Get namespace paths - namespacePaths := make([]string, 0, len(ds.Namespaces)) - for _, ns := range ds.Namespaces { - namespacePaths = append(namespacePaths, ns.Path) + if ctx.Err() != nil { + log.Warn(). + Str("instance", instanceName). + Msg("PBS backup polling cancelled before completion") + return } + namespacePaths := namespacePathsForDatastore(ds) + log.Info(). Str("instance", instanceName). Str("datastore", ds.Name). @@ -9494,82 +9519,99 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien Strs("namespace_paths", namespacePaths). Msg("Processing datastore namespaces") - // Fetch backups from all namespaces concurrently - backupsMap, err := client.ListAllBackups(ctx, ds.Name, namespacePaths) - if err != nil { - log.Error().Err(err). - Str("instance", instanceName). - Str("datastore", ds.Name). - Msg("Failed to fetch PBS backups") - datastoreErrors++ - continue + datastoreHadSuccess := false + groupsReused := 0 + groupsRequested := 0 + + for _, namespace := range namespacePaths { + if ctx.Err() != nil { + log.Warn(). + Str("instance", instanceName). + Msg("PBS backup polling cancelled mid-datastore") + return + } + + groups, err := client.ListBackupGroups(ctx, ds.Name, namespace) + if err != nil { + log.Error(). + Err(err). + Str("instance", instanceName). + Str("datastore", ds.Name). + Str("namespace", namespace). + Msg("Failed to list PBS backup groups") + continue + } + + datastoreHadSuccess = true + requests := make([]pbsBackupFetchRequest, 0, len(groups)) + + for _, group := range groups { + key := pbsBackupGroupKey{ + datastore: ds.Name, + namespace: namespace, + backupType: group.BackupType, + backupID: group.BackupID, + } + cached := existingGroups[key] + + // Group deleted (no backups left) - ensure cached data is dropped. + if group.BackupCount == 0 { + continue + } + + lastBackupTime := time.Unix(group.LastBackup, 0) + hasCachedData := len(cached.snapshots) > 0 + // Only re-fetch when the backup count changes or the most recent backup is newer. + if hasCachedData && + len(cached.snapshots) == group.BackupCount && + !lastBackupTime.After(cached.latest) { + + allBackups = append(allBackups, cached.snapshots...) + groupsReused++ + continue + } + + requests = append(requests, pbsBackupFetchRequest{ + datastore: ds.Name, + namespace: namespace, + group: group, + cached: cached, + }) + } + + if len(requests) == 0 { + continue + } + + groupsRequested += len(requests) + fetched := m.fetchPBSBackupSnapshots(ctx, client, instanceName, requests) + if len(fetched) > 0 { + allBackups = append(allBackups, fetched...) + } } - datastoreFetches++ - - // Convert PBS backups to model backups - for namespace, snapshots := range backupsMap { - for _, snapshot := range snapshots { - backupTime := time.Unix(snapshot.BackupTime, 0) - - // Generate unique ID - id := fmt.Sprintf("pbs-%s-%s-%s-%s-%s-%d", - instanceName, ds.Name, namespace, - snapshot.BackupType, snapshot.BackupID, - snapshot.BackupTime) - - // Extract file names from files (which can be strings or objects) - var fileNames []string - for _, file := range snapshot.Files { - switch f := file.(type) { - case string: - fileNames = append(fileNames, f) - case map[string]interface{}: - if filename, ok := f["filename"].(string); ok { - fileNames = append(fileNames, filename) - } - } + if datastoreHadSuccess { + datastoreFetches++ + log.Info(). + Str("instance", instanceName). + Str("datastore", ds.Name). + Int("namespaces", len(namespacePaths)). + Int("groups_reused", groupsReused). + Int("groups_refreshed", groupsRequested). + Msg("PBS datastore processed") + } else { + // Preserve cached data for this datastore if we couldn't fetch anything new. + log.Warn(). + Str("instance", instanceName). + Str("datastore", ds.Name). + Msg("No namespaces succeeded for PBS datastore; using cached backups") + for key, entry := range existingGroups { + if key.datastore != ds.Name || len(entry.snapshots) == 0 { + continue } - - // Extract verification status - verified := false - if snapshot.Verification != nil { - switch v := snapshot.Verification.(type) { - case string: - verified = v == "ok" - case map[string]interface{}: - if state, ok := v["state"].(string); ok { - verified = state == "ok" - } - } - - // Debug log verification data - log.Debug(). - Str("vmid", snapshot.BackupID). - Int64("time", snapshot.BackupTime). - Interface("verification", snapshot.Verification). - Bool("verified", verified). - Msg("PBS backup verification status") - } - - backup := models.PBSBackup{ - ID: id, - Instance: instanceName, - Datastore: ds.Name, - Namespace: namespace, - BackupType: snapshot.BackupType, - VMID: snapshot.BackupID, - BackupTime: backupTime, - Size: snapshot.Size, - Protected: snapshot.Protected, - Verified: verified, - Comment: snapshot.Comment, - Files: fileNames, - Owner: snapshot.Owner, - } - - allBackups = append(allBackups, backup) + allBackups = append(allBackups, entry.snapshots...) } + datastoreErrors++ } } @@ -9610,6 +9652,184 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien } } +func (m *Monitor) buildPBSBackupCache(instanceName string) map[pbsBackupGroupKey]cachedPBSGroup { + snapshot := m.state.GetSnapshot() + cache := make(map[pbsBackupGroupKey]cachedPBSGroup) + for _, backup := range snapshot.PBSBackups { + if backup.Instance != instanceName { + continue + } + key := pbsBackupGroupKey{ + datastore: backup.Datastore, + namespace: normalizePBSNamespacePath(backup.Namespace), + backupType: backup.BackupType, + backupID: backup.VMID, + } + entry := cache[key] + entry.snapshots = append(entry.snapshots, backup) + if backup.BackupTime.After(entry.latest) { + entry.latest = backup.BackupTime + } + cache[key] = entry + } + return cache +} + +func normalizePBSNamespacePath(ns string) string { + if ns == "/" { + return "" + } + return ns +} + +func namespacePathsForDatastore(ds models.PBSDatastore) []string { + if len(ds.Namespaces) == 0 { + return []string{""} + } + + seen := make(map[string]struct{}, len(ds.Namespaces)) + var paths []string + for _, ns := range ds.Namespaces { + path := normalizePBSNamespacePath(ns.Path) + if _, ok := seen[path]; ok { + continue + } + seen[path] = struct{}{} + paths = append(paths, path) + } + if len(paths) == 0 { + paths = append(paths, "") + } + return paths +} + +func (m *Monitor) fetchPBSBackupSnapshots(ctx context.Context, client *pbs.Client, instanceName string, requests []pbsBackupFetchRequest) []models.PBSBackup { + if len(requests) == 0 { + return nil + } + + results := make(chan []models.PBSBackup, len(requests)) + var wg sync.WaitGroup + sem := make(chan struct{}, 5) + + for _, req := range requests { + req := req + wg.Add(1) + go func() { + defer wg.Done() + + select { + case sem <- struct{}{}: + case <-ctx.Done(): + return + } + defer func() { <-sem }() + + log.Debug(). + Str("instance", instanceName). + Str("datastore", req.datastore). + Str("namespace", req.namespace). + Str("type", req.group.BackupType). + Str("id", req.group.BackupID). + Msg("Refreshing PBS backup group") + + snapshots, err := client.ListBackupSnapshots(ctx, req.datastore, req.namespace, req.group.BackupType, req.group.BackupID) + if err != nil { + log.Error(). + Err(err). + Str("instance", instanceName). + Str("datastore", req.datastore). + Str("namespace", req.namespace). + Str("type", req.group.BackupType). + Str("id", req.group.BackupID). + Msg("Failed to list PBS backup snapshots") + + if len(req.cached.snapshots) > 0 { + results <- req.cached.snapshots + } + return + } + + results <- convertPBSSnapshots(instanceName, req.datastore, req.namespace, snapshots) + }() + } + + go func() { + wg.Wait() + close(results) + }() + + var combined []models.PBSBackup + for backups := range results { + if len(backups) == 0 { + continue + } + combined = append(combined, backups...) + } + + return combined +} + +func convertPBSSnapshots(instanceName, datastore, namespace string, snapshots []pbs.BackupSnapshot) []models.PBSBackup { + backups := make([]models.PBSBackup, 0, len(snapshots)) + for _, snapshot := range snapshots { + backupTime := time.Unix(snapshot.BackupTime, 0) + id := fmt.Sprintf("pbs-%s-%s-%s-%s-%s-%d", + instanceName, datastore, namespace, + snapshot.BackupType, snapshot.BackupID, + snapshot.BackupTime) + + var fileNames []string + for _, file := range snapshot.Files { + switch f := file.(type) { + case string: + fileNames = append(fileNames, f) + case map[string]interface{}: + if filename, ok := f["filename"].(string); ok { + fileNames = append(fileNames, filename) + } + } + } + + verified := false + if snapshot.Verification != nil { + switch v := snapshot.Verification.(type) { + case string: + verified = v == "ok" + case map[string]interface{}: + if state, ok := v["state"].(string); ok { + verified = state == "ok" + } + } + + log.Debug(). + Str("vmid", snapshot.BackupID). + Int64("time", snapshot.BackupTime). + Interface("verification", snapshot.Verification). + Bool("verified", verified). + Msg("PBS backup verification status") + } + + backups = append(backups, models.PBSBackup{ + ID: id, + Instance: instanceName, + Datastore: datastore, + Namespace: namespace, + BackupType: snapshot.BackupType, + VMID: snapshot.BackupID, + BackupTime: backupTime, + Size: snapshot.Size, + Protected: snapshot.Protected, + Verified: verified, + Comment: snapshot.Comment, + Files: fileNames, + Owner: snapshot.Owner, + }) + } + + return backups +} + // checkMockAlerts checks alerts for mock data func (m *Monitor) checkMockAlerts() { defer recoverFromPanic("checkMockAlerts")