diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 782ffeeb9d..c4bd25b847 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -6907,6 +6907,21 @@ without deleting retained artifacts. The observation is persisted before reconciliation, so cached backup files cannot continue presenting current protection after collection becomes incomplete or unauthorized. +PBS namespace discovery is part of that collection evidence. An unavailable, +malformed or denied namespace listing must travel with the matching datastore +snapshot into background backup polling. A readable root namespace may still +supply accepted artifacts, but cannot turn failed child-namespace discovery +into a complete root-only inventory. Such a poll records partial history with +unknown or partial access, preserves unseen durable history and never guesses +child namespace requests. Recovery and a later successful namespace removal +resume ordinary complete, connection-scoped reconciliation. The PBS client's +existing legacy root-only response for an absent namespace endpoint is unchanged. +`TestPBSNamespaceDiscoveryFailureRetainsRecoveryHistory` crosses real HTTP +namespace discovery, background polling, durable ingest and canonical posture +on warm and replacement inventories; recovery, scoped removal, first-read and +legacy controls cover both safe retention and authoritative absence. No public +model, endpoint, policy, guest operation or restorable-backup claim is added. + Direct PBS backup enumeration emits two separate storage/recovery inputs: subject-linked recovery points and one typed provider observation for the polled PBS instance. A complete poll records complete history with sufficient @@ -6917,6 +6932,29 @@ denied access. Retained backup points survive failed enumeration, but the new provider observation immediately prevents those cached points from being presented as current protection truth. +Datastore discovery itself is part of that backup-scope result, including +the connectivity fallback. An admitted failed or malformed listing records +unavailable history while retaining existing raw artifacts, durable points and +snapshot-cache ages; only HTTP 401/403 establishes denied access. An explicit +successful empty array (including a fully excluded scope) instead reconciles +the connection's backup points, live rows and cache timestamps. Missing/null +`data` and unnamed datastore entries are rejected before any detail request, +not interpreted as authoritative absence. Both outcomes use the existing +serialized background backup admission, cadence and runtime deadline. +Disabled, not-due, in-flight and cancelled work is not an accepted empty scope; +backup availability remains separate from node connectivity. Ordinary complete +recovery resumes collection and protection without credential or guest changes. +`TestGetDatastoresRequiresExplicitNamedScope` and +`TestGetDatastoresAcceptsExplicitEmptyScope` pin the client envelope boundary. +`TestPBSDatastoreDiscoveryFailureQualifiesRetainedHistory` crosses real HTTP, +ordinary background polling, warm/rebuilt live state, durable ingestion and +canonical posture for repeated 401/403/404/503 and malformed listings. +`TestPBSDatastoreDiscoveryEmptyReconcilesOnlyAcceptedScope`, +`TestPBSDatastoreDiscoverySkippedScopeIsNotEmpty` and +`TestPBSDatastoreDiscoveryFailedConnectivityFallback` cover scoped removal, +exclusions, reappearance, admission/cancellation and failed fallback. These are +synthetic source checks, not installed PBS or guest/restore acceptance. + PBS mapping attaches provider scope and a typed evidence envelope to every successfully enumerated recovery point. Identity correlation is confirmed only for direct canonical identity and inferred only for an auditable unique diff --git a/internal/monitoring/monitor_backups.go b/internal/monitoring/monitor_backups.go index b64eb24ef1..0f42ba3186 100644 --- a/internal/monitoring/monitor_backups.go +++ b/internal/monitoring/monitor_backups.go @@ -1579,7 +1579,7 @@ type pbsBackupFetchRequest struct { } // pollPBSBackups fetches all backups from PBS datastores -func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, client *pbs.Client, datastores []models.PBSDatastore) { +func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, client *pbs.Client, datastores []models.PBSDatastore, namespaceErrors map[string]error) { log.Debug().Str("instance", instanceName).Msg("polling PBS backups") // Cache existing PBS backups so we can avoid redundant API calls when no changes occurred. @@ -1615,6 +1615,15 @@ func (m *Monitor) pollPBSBackups(ctx context.Context, instanceName string, clien datastoreHadSuccess := false datastoreNamespaceErrors := 0 datastoreTerminalNamespaceErrors := 0 + // The root namespace remains readable when discovering child namespaces + // fails. Its accepted rows are useful, but they cannot establish that + // undiscovered namespaces are empty or renew complete-provider evidence. + if err := namespaceErrors[ds.Name]; err != nil { + datastoreNamespaceErrors++ + if !shouldReuseCachedPBSBackups(err) { + datastoreTerminalNamespaceErrors++ + } + } groupsReused := 0 groupsRequested := 0 diff --git a/internal/monitoring/monitor_backups_readstate_test.go b/internal/monitoring/monitor_backups_readstate_test.go index d910e1bb77..a64405b742 100644 --- a/internal/monitoring/monitor_backups_readstate_test.go +++ b/internal/monitoring/monitor_backups_readstate_test.go @@ -348,7 +348,7 @@ func TestPollPBSBackupsFetchesRealSnapshotsForLargeGroups(t *testing.T) { } m := &Monitor{state: models.NewState()} - m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}) + m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil) if got := atomic.LoadInt64(&snapshotCalls); got != 1 { t.Fatalf("large PBS backup group fetched snapshots %d times, want 1", got) @@ -383,7 +383,7 @@ func TestPollPBSBackupsFetchesRealSnapshotsForLargeGroups(t *testing.T) { } // A second poll must reuse the bounded cache instead of refetching. - m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}) + m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil) if got := atomic.LoadInt64(&snapshotCalls); got != 1 { t.Fatalf("second poll refetched snapshots (%d calls), want cached reuse", got) } @@ -448,7 +448,7 @@ func TestPollPBSBackupsBoundsLargeTopologyAcrossPolls(t *testing.T) { m := &Monitor{state: models.NewState()} for cycle := 0; cycle < 3; cycle++ { - m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}) + m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil) snapshot := m.state.GetSnapshot() if got := len(snapshot.PBSBackups); got != pbsBackupLiveStateLimit { @@ -615,7 +615,7 @@ func TestPollPBSBackupsPrunesCacheTimesForDeletedGroups(t *testing.T) { m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ {Name: "archive"}, - }) + }, nil) m.mu.RLock() perGroup := m.pbsBackupCacheTime["pbs1"] @@ -1039,7 +1039,7 @@ func TestPollPBSBackups_DropsStaleCacheOnTerminalDatastoreError(t *testing.T) { m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ {Name: "archive"}, - }) + }, nil) snapshot := m.state.GetSnapshot() for _, backup := range snapshot.PBSBackups { @@ -1097,7 +1097,7 @@ func TestPollPBSBackups_PreservesCacheOnTransientDatastoreError(t *testing.T) { m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ {Name: "archive"}, - }) + }, nil) snapshot := m.state.GetSnapshot() var found bool diff --git a/internal/monitoring/monitor_extra_coverage_test.go b/internal/monitoring/monitor_extra_coverage_test.go index d151305fc4..c5d6a0ec05 100644 --- a/internal/monitoring/monitor_extra_coverage_test.go +++ b/internal/monitoring/monitor_extra_coverage_test.go @@ -2394,7 +2394,7 @@ func TestMonitor_PollPBSBackups_Extra(t *testing.T) { t.Fatal(err) } ds := []models.PBSDatastore{{Name: "ds1"}} - m.pollPBSBackups(context.Background(), "pbs1", client, ds) + m.pollPBSBackups(context.Background(), "pbs1", client, ds, nil) } func TestMonitor_RetryPVEPortFallback_Detailed_Extra(t *testing.T) { diff --git a/internal/monitoring/monitor_full_coverage_test.go b/internal/monitoring/monitor_full_coverage_test.go index d91a46f41a..a7956a8e57 100644 --- a/internal/monitoring/monitor_full_coverage_test.go +++ b/internal/monitoring/monitor_full_coverage_test.go @@ -222,7 +222,7 @@ func TestPollPBSBackups(t *testing.T) { } // Execute - m.pollPBSBackups(context.Background(), "pbs1", client, datastores) + m.pollPBSBackups(context.Background(), "pbs1", client, datastores, nil) // Verify found := false @@ -1115,7 +1115,7 @@ func TestPollPBSBackupsCachedIncompleteSnapshotLosesStaleTaskObservation(t *test WriteActive: false, }}) - m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}) + m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{{Name: "archive"}}, nil) if snapshotRequests.Load() != 1 { t.Fatalf("snapshot requests = %d, want 1 cached fallback attempt", snapshotRequests.Load()) diff --git a/internal/monitoring/monitor_pbs_coverage_test.go b/internal/monitoring/monitor_pbs_coverage_test.go index 20f1674150..140dabdf89 100644 --- a/internal/monitoring/monitor_pbs_coverage_test.go +++ b/internal/monitoring/monitor_pbs_coverage_test.go @@ -3,6 +3,7 @@ package monitoring import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "os" @@ -15,6 +16,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/recovery" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" @@ -809,3 +811,38 @@ func TestPBSPolledCapacityRequiresObservedRecovery(t *testing.T) { t.Fatalf("raised policy did not clear high-usage alert: %+v", active) } } + +func TestPBSNamespaceDiscoveryFirstPollDoesNotInventScope(t *testing.T) { + for _, status := range []int{http.StatusServiceUnavailable, http.StatusForbidden} { + t.Run(fmt.Sprintf("status-%d", status), func(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.namespaceStatus.Store(int32(status)) + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 1 || f.childReads.Load() != 0 { + t.Fatal("failed first discovery invented a child namespace or lost readable root history") + } + for _, p := range f.postures(t) { + if p.State == recovery.ProtectionStateProtected { + t.Errorf("incomplete first discovery claimed current protection: %+v", p) + } + } + }) + } +} + +func TestPBSNamespaceDiscoveryLegacyRootOnlyRemainsComplete(t *testing.T) { + f := newPBSNamespacePollFixture(t) + // The existing PBS client treats an absent namespace endpoint on legacy + // versions as a successful root-only inventory. Preserve that support. + f.namespaceStatus.Store(http.StatusNotFound) + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 1 || f.childReads.Load() != 0 { + t.Fatal("legacy root-only inventory was not collected") + } + for _, p := range f.postures(t) { + if p.SubjectResourceID == f.ids[0] && (p.State != recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 || + p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryComplete) { + t.Errorf("legacy root-only support was marked incomplete: %+v", p) + } + } +} diff --git a/internal/monitoring/monitor_pbs_pmg.go b/internal/monitoring/monitor_pbs_pmg.go index 1508a9217e..f0f9448bd3 100644 --- a/internal/monitoring/monitor_pbs_pmg.go +++ b/internal/monitoring/monitor_pbs_pmg.go @@ -13,6 +13,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/logging" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors" + "github.com/rcourtman/pulse-go-rewrite/internal/recovery" "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" "github.com/rcourtman/pulse-go-rewrite/pkg/pmg" @@ -319,6 +320,11 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) { m.recordAuthFailure(instanceName, "pbs") } + // The connectivity fallback also failed datastore discovery. Qualify + // retained backup evidence through the ordinary serialized admission. + if !stderrors.Is(datastoreErr, context.Canceled) { + m.pollPBSBackupsAsync(instanceName, client, instanceCfg, nil, nil, datastoreErr) + } return } } @@ -358,10 +364,15 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie Msg("PBS node status retrieved") } + // Keep discovery quality with the immutable input to the background backup + // poll. Public datastore presentation is not an authoritative scope receipt. + namespaceErrors := make(map[string]error) + var datastoreDiscoveryErr error // Poll datastores if enabled if instanceCfg.MonitorDatastores { datastores, err := client.GetDatastoresFiltered(ctx, includeDatastore) if err != nil { + datastoreDiscoveryErr = err monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0) log.Error().Err(monErr).Str("instance", instanceName).Msg("failed to get datastores") } else { @@ -417,6 +428,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0) if err != nil { + namespaceErrors[ds.Store] = err log.Warn().Err(err). Str("instance", instanceName). Str("datastore", ds.Store). @@ -520,96 +532,90 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie Msg("Added PBS datastores to unified storage view") } - // Poll backups if enabled - if instanceCfg.MonitorBackups { - if len(pbsInst.Datastores) == 0 { - log.Debug(). - Str("instance", instanceName). - Msg("No PBS datastores available for backup polling") - } else if !m.backupPollingEnabledSetting() { - log.Debug(). - Str("instance", instanceName). - Msg("Skipping PBS backup polling - globally disabled") - } else { - now := time.Now() - - m.mu.Lock() - lastPoll := m.lastPBSBackupPoll[instanceName] - if m.pbsBackupPollers == nil { - m.pbsBackupPollers = make(map[string]bool) - } - inProgress := m.pbsBackupPollers[instanceName] - m.mu.Unlock() - - shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) - if !shouldPoll { - if reason != "" { - log.Debug(). - Str("instance", instanceName). - Str("reason", reason). - Msg("Skipping PBS backup polling this cycle") - } - } else if inProgress { - log.Debug(). - Str("instance", instanceName). - Msg("PBS backup polling already in progress") - } else { - datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores)) - copy(datastoreSnapshot, pbsInst.Datastores) - - // Atomically check and set poller flag - m.mu.Lock() - if m.pbsBackupPollers[instanceName] { - // Race: another goroutine started between our check and lock - m.mu.Unlock() - log.Debug(). - Str("instance", instanceName). - Msg("PBS backup polling started by another goroutine") - } else { - m.pbsBackupPollers[instanceName] = true - m.lastPBSBackupPoll[instanceName] = newLast - m.mu.Unlock() - - go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) { - defer recoverFromPanic(fmt.Sprintf("pollPBSBackups-%s", inst)) - defer func() { - m.mu.Lock() - delete(m.pbsBackupPollers, inst) - m.lastPBSBackupPoll[inst] = time.Now() - m.mu.Unlock() - }() - - log.Info(). - Str("instance", inst). - Int("datastores", len(ds)). - Msg("Starting background PBS backup polling") - - // The per-cycle ctx is canceled as soon as the main polling loop finishes, - // so derive the backup poll context from the long-lived runtime context instead. - parentCtx := m.getRuntimeContext() - if parentCtx == nil { - parentCtx = context.Background() - } - backupCtx, cancel := context.WithTimeout(parentCtx, 5*time.Minute) - defer cancel() - - m.pollPBSBackups(backupCtx, inst, pbsClient, ds) - - log.Info(). - Str("instance", inst). - Dur("duration", time.Since(start)). - Msg("Completed background PBS backup polling") - }(datastoreSnapshot, instanceName, now, client) - } - } - } - } else { - log.Debug(). - Str("instance", instanceName). - Msg("PBS backup monitoring disabled") + // A successful empty scope and a failed discovery both need an accepted + // backup-poll result; neither may silently keep the last good protection. + if !stderrors.Is(datastoreDiscoveryErr, context.Canceled) { + m.pollPBSBackupsAsync(instanceName, client, instanceCfg, pbsInst.Datastores, namespaceErrors, datastoreDiscoveryErr) } } +// pollPBSBackupsAsync admits all backup-scope outcomes through one worker. +// Failed discovery never becomes an empty enumeration, and a skipped/disabled +// discovery is not absence evidence. Keeping failure/empty publication inside +// this admission prevents it from racing an older in-flight backup worker. +func (m *Monitor) pollPBSBackupsAsync(instanceName string, client *pbs.Client, instanceCfg *config.PBSInstance, datastores []models.PBSDatastore, namespaceErrors map[string]error, discoveryErr error) { + if !instanceCfg.MonitorBackups || !instanceCfg.MonitorDatastores || !m.backupPollingEnabledSetting() { + return + } + now := time.Now() + m.mu.RLock() + lastPoll := m.lastPBSBackupPoll[instanceName] + m.mu.RUnlock() + shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) + if !shouldPoll { + log.Debug().Str("instance", instanceName).Str("reason", reason).Msg("Skipping PBS backup polling this cycle") + return + } + + datastoreSnapshot := append([]models.PBSDatastore(nil), datastores...) + m.mu.Lock() + if m.pbsBackupPollers[instanceName] { + m.mu.Unlock() + log.Debug().Str("instance", instanceName).Msg("PBS backup polling already in progress") + return + } + if m.pbsBackupPollers == nil { + m.pbsBackupPollers = make(map[string]bool) + } + if m.lastPBSBackupPoll == nil { + m.lastPBSBackupPoll = make(map[string]time.Time) + } + m.pbsBackupPollers[instanceName] = true + m.lastPBSBackupPoll[instanceName] = newLast + m.mu.Unlock() + + go func() { + defer recoverFromPanic(fmt.Sprintf("pollPBSBackups-%s", instanceName)) + defer func() { + m.mu.Lock() + delete(m.pbsBackupPollers, instanceName) + m.lastPBSBackupPoll[instanceName] = time.Now() + m.mu.Unlock() + }() + + // The main polling loop cancels its per-cycle context after returning. + // Backup work keeps the existing independent, runtime-scoped deadline. + parentCtx := m.getRuntimeContext() + if parentCtx == nil { + parentCtx = context.Background() + } + backupCtx, cancel := context.WithTimeout(parentCtx, 5*time.Minute) + defer cancel() + if backupCtx.Err() != nil { + return + } + if discoveryErr != nil { + // One unreadable discovery scope, not a guessed datastore count. + // Only actual authentication/authorization status establishes denial. + denied := 0 + if status, ok := pbs.HTTPStatus(discoveryErr); ok && (status == 401 || status == 403) { + denied = 1 + } + observation, err := buildPBSProtectionProviderObservation(instanceName, 1, 0, 1, denied, time.Now().UTC()) + if err != nil { + log.Warn().Err(err).Str("instance", instanceName).Msg("Failed to build PBS discovery failure observation") + return + } + m.ingestProtectionProviderObservationsAsync([]recovery.ProtectionProviderObservation{observation}) + } else { + // The explicit empty list is authoritative and reconciles only this + // connection; the ordinary poller also drops its live rows/cache. + m.pollPBSBackups(backupCtx, instanceName, client, datastoreSnapshot, namespaceErrors) + } + log.Info().Str("instance", instanceName).Dur("duration", time.Since(now)).Msg("Completed background PBS backup polling") + }() +} + // recordPBSNodeHistory writes the API-observed CPU and memory measurements to // the PBS service's existing metrics target (agent:). When no // host agent can be corroborated, the Backups drawer still has useful history diff --git a/internal/monitoring/pbs_datastore_discovery_test.go b/internal/monitoring/pbs_datastore_discovery_test.go new file mode 100644 index 0000000000..f9207ea64d --- /dev/null +++ b/internal/monitoring/pbs_datastore_discovery_test.go @@ -0,0 +1,199 @@ +package monitoring + +import ( + "context" + "fmt" + "net/http" + "reflect" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" + "github.com/rcourtman/pulse-go-rewrite/internal/recovery" +) + +func TestPBSDatastoreDiscoveryFailureQualifiesRetainedHistory(t *testing.T) { + for _, scenario := range []struct { + name string + status int + body string + denied bool + }{ + {name: "server failure", status: 503}, + {name: "unauthenticated", status: 401, denied: true}, + {name: "forbidden", status: 403, denied: true}, + {name: "missing endpoint is not permission evidence", status: 404}, + {name: "malformed envelope", body: `{"data":`}, + {name: "missing data", body: `{}`}, + {name: "null data", body: `{"data":null}`}, + {name: "nameless datastore", body: `{"data":[{}]}`}, + } { + for _, cold := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/cold-%t", scenario.name, cold), func(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.poll(t, "pbs-main") + before := f.points(t, "pbs-main") + if len(before) != 2 { + t.Fatal("healthy control did not ingest root and tenant") + } + if cold { + vms := f.m.state.GetSnapshot().VMs + f.m.state = models.NewState() + f.m.state.UpdateVMs(vms) + } + raw := f.m.state.GetSnapshot().PBSBackups + key := pbsBackupGroupKey{datastore: "archive", backupType: "vm", backupID: "100"} + cacheAt := f.m.pbsBackupCacheTimeFor("pbs-main", key) + reads := f.backupReads.Load() + f.datastoreStatus.Store(int32(scenario.status)) + if scenario.body != "" { + f.datastoreBody.Store(&scenario.body) + } + for range 2 { + f.poll(t, "pbs-main") + if !reflect.DeepEqual(before, f.points(t, "pbs-main")) || !reflect.DeepEqual(raw, f.m.state.GetSnapshot().PBSBackups) { + t.Error("failed discovery deleted or renewed retained artifacts") + } + if f.backupReads.Load() != reads || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(cacheAt) { + t.Error("failed discovery guessed a backup read or renewed cache") + } + permissions := operationaltrust.EvidencePermissionsUnknown + if scenario.denied { + permissions = operationaltrust.EvidencePermissionsDenied + } + for _, p := range f.postures(t) { + if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 { + t.Errorf("failed discovery left protection healthy: %+v", p) + continue + } + provider := p.ProviderStates[0] + if provider.HistoryCompleteness != recovery.ProtectionHistoryUnavailable || provider.Permissions != permissions || provider.JobState != recovery.OutcomeFailed { + t.Errorf("failure quality lost: %+v", provider) + } + } + // Backup history availability is not node connectivity. + inst := pbsInstanceByName(t, f.m.state.GetSnapshot(), "pbs-main") + if inst.Status != "online" || inst.ConnectionHealth != "healthy" { + t.Errorf("discovery failure rewrote connectivity: %+v", inst) + } + } + f.datastoreStatus.Store(0) + f.datastoreBody.Store(nil) + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 2 { + t.Error("ordinary recovery did not retain both archives") + } + for _, p := range f.postures(t) { + if p.State != recovery.ProtectionStateProtected { + t.Errorf("complete recovery did not restore protection: %+v", p) + } + } + }) + } + } +} + +func TestPBSDatastoreDiscoveryEmptyReconcilesOnlyAcceptedScope(t *testing.T) { + for _, excluded := range []bool{false, true} { + t.Run(fmt.Sprintf("excluded-%t", excluded), func(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.poll(t, "pbs-other") + other := f.points(t, "pbs-other") + f.poll(t, "pbs-main") + reads := f.backupReads.Load() + if excluded { + f.m.config.PBSInstances[0].ExcludeDatastores = []string{"archive"} + } else { + empty := `{"data":[]}` + f.datastoreBody.Store(&empty) + } + for range 2 { + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 0 { + t.Error("accepted empty scope retained nonexistent/removed archives") + } + for _, row := range f.m.state.GetSnapshot().PBSBackups { + if row.Instance == "pbs-main" { + t.Error("accepted empty scope retained a live archive") + } + } + if !reflect.DeepEqual(other, f.points(t, "pbs-other")) || f.backupReads.Load() != reads { + t.Error("empty reconciliation crossed scope or guessed a backup request") + } + f.m.mu.RLock() + cacheSize := len(f.m.pbsBackupCacheTime["pbs-main"]) + f.m.mu.RUnlock() + if cacheSize != 0 { + t.Error("confirmed empty scope retained snapshot cache timestamps") + } + } + f.datastoreBody.Store(nil) + f.m.config.PBSInstances[0].ExcludeDatastores = nil + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 2 || !reflect.DeepEqual(other, f.points(t, "pbs-other")) { + t.Error("scope restoration did not rediscover archives independently") + } + }) + } +} + +func TestPBSDatastoreDiscoverySkippedScopeIsNotEmpty(t *testing.T) { + for _, skip := range []string{"backups disabled", "datastores disabled", "global disabled", "not due", "in flight", "cancelled cycle", "cancelled runtime"} { + t.Run(skip, func(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.poll(t, "pbs-main") + before, raw := f.points(t, "pbs-main"), f.m.state.GetSnapshot().PBSBackups + empty := `{"data":[]}` + f.datastoreBody.Store(&empty) + ctx := context.Background() + switch skip { + case "backups disabled": + f.m.config.PBSInstances[0].MonitorBackups = false + case "datastores disabled": + f.m.config.PBSInstances[0].MonitorDatastores = false + case "global disabled": + f.m.SetBackupPollingEnabled(false) + case "not due": + // Model an already-configured cadence. Changing cadence via the + // live setter deliberately clears timestamps for a catch-up poll. + f.m.config.BackupPollingInterval = time.Hour + case "in flight": + f.m.pbsBackupPollers["pbs-main"] = true + case "cancelled cycle": + cancelled, cancel := context.WithCancel(ctx) + cancel() + ctx = cancelled + case "cancelled runtime": + cancelled, cancel := context.WithCancel(ctx) + cancel() + f.m.runtimeCtx = cancelled + } + f.m.pollPBSInstance(ctx, "pbs-main", f.client) + if skip == "in flight" { + delete(f.m.pbsBackupPollers, "pbs-main") + } + f.wait(t) + if !reflect.DeepEqual(before, f.points(t, "pbs-main")) || !reflect.DeepEqual(raw, f.m.state.GetSnapshot().PBSBackups) { + t.Error("skipped/cancelled discovery was treated as accepted empty scope") + } + }) + } +} + +func TestPBSDatastoreDiscoveryFailedConnectivityFallback(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.poll(t, "pbs-main") + before := f.points(t, "pbs-main") + f.versionStatus.Store(http.StatusServiceUnavailable) + f.datastoreStatus.Store(http.StatusServiceUnavailable) + f.poll(t, "pbs-main") + if !reflect.DeepEqual(before, f.points(t, "pbs-main")) { + t.Error("failed fallback destroyed history") + } + for _, p := range f.postures(t) { + if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 || p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryUnavailable { + t.Errorf("failed fallback left last good protection: %+v", p) + } + } +} diff --git a/internal/monitoring/pbs_poll_cancel_test.go b/internal/monitoring/pbs_poll_cancel_test.go index fbbc7d0cf5..148b2ef0a8 100644 --- a/internal/monitoring/pbs_poll_cancel_test.go +++ b/internal/monitoring/pbs_poll_cancel_test.go @@ -74,7 +74,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) { case "during-task-correlation": datastores = []models.PBSDatastore{{Name: "archive"}} } - f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores) + f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores, nil) f.wait(t) f.assertProtected(t, "pbs-main") key := pbsBackupGroupKey{datastore: "archive", backupType: "vm", backupID: "100"} @@ -86,7 +86,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) { done := make(chan struct{}) go func() { defer close(done) - f.m.pollPBSBackups(ctx, "pbs-main", f.client, datastores) + f.m.pollPBSBackups(ctx, "pbs-main", f.client, datastores, nil) }() select { case <-blocked: @@ -116,7 +116,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) { // Do not clear or expire cache again. The next ordinary poll must // independently fetch the verification change the aborted poll lost. phase.Store(2) - f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores) + f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores, nil) f.wait(t) raw = f.m.state.GetSnapshot().PBSBackups mapped, err := proxmoxmapper.FromPBSBackupsWithEvidence(raw, buildPBSGuestCandidates(f.m.GetUnifiedReadStateOrSnapshot()), time.Now()) @@ -137,7 +137,7 @@ func TestPBSOrdinaryPollCancellationDoesNotRenewUnpublishedCache(t *testing.T) { } // Once the matching failed row has been published, normal TTL reuse // must avoid an unnecessary fourth read without restoring verified. - f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores) + f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, datastores, nil) f.wait(t) if snapshotReads.Load() != 3 || !f.m.pbsBackupCacheTimeFor("pbs-main", key).Equal(publishedAt) || f.postures(t)[0].Verification != recovery.ProtectionVerificationUnverified { t.Error("cache-hit poll reread or renewed snapshots, or restored stale verification") diff --git a/internal/monitoring/pbs_poll_posture_test.go b/internal/monitoring/pbs_poll_posture_test.go index acfc4b7dcf..9cdcc553ed 100644 --- a/internal/monitoring/pbs_poll_posture_test.go +++ b/internal/monitoring/pbs_poll_posture_test.go @@ -131,7 +131,7 @@ func (f *pbsPosturePollFixture) poll(t *testing.T, instance string, mode int32, f.m.mu.Lock() clear(f.m.pbsBackupCacheTime) f.m.mu.Unlock() - f.m.pollPBSBackups(context.Background(), instance, f.client, datastores) + f.m.pollPBSBackups(context.Background(), instance, f.client, datastores, nil) f.wait(t) } @@ -289,7 +289,7 @@ func TestPBSOrdinaryPollSnapshotFailuresDoNotClaimCompleteHistory(t *testing.T) } // No TTL manipulation on the next poll: a failed refresh must not // become a healthy cached read while snapshot access is still lost. - f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds) + f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds, nil) f.wait(t) } f.poll(t, "pbs-main", 0, ds) diff --git a/internal/monitoring/pbs_protection_observation_test.go b/internal/monitoring/pbs_protection_observation_test.go index 4667265c3c..3688689701 100644 --- a/internal/monitoring/pbs_protection_observation_test.go +++ b/internal/monitoring/pbs_protection_observation_test.go @@ -8,14 +8,18 @@ import ( "net/http" "net/http/httptest" "os" + "reflect" "strings" "sync/atomic" "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" "github.com/rcourtman/pulse-go-rewrite/internal/recovery" + recoverystore "github.com/rcourtman/pulse-go-rewrite/internal/recovery/store" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" ) @@ -186,7 +190,7 @@ func TestPBSOrdinaryPollNewestRemovalInvalidatesCappedCache(t *testing.T) { phase.Store(mode) // No cache clearing, TTL expiry or monitor replacement: these // are consecutive ordinary polls of the same connection. - f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds) + f.m.pollPBSBackups(context.Background(), "pbs-main", f.client, ds, nil) f.wait(t) } frames := map[string]any{} @@ -284,3 +288,273 @@ func TestPBSOrdinaryPollNewestRemovalInvalidatesCappedCache(t *testing.T) { } } } + +// Namespace discovery is part of backup enumeration, not just optional drawer +// metadata. Drive the public poll entry point and its background backup worker +// so a missing namespace cannot masquerade as an authoritative root-only scan. +type pbsNamespacePollFixture struct { + m *Monitor + store *recoverystore.Store + client *pbs.Client + datastoreStatus atomic.Int32 + datastoreBody atomic.Pointer[string] + versionStatus atomic.Int32 + backupReads atomic.Int32 + namespaceStatus atomic.Int32 + namespaceRemoved atomic.Bool + rootEmpty atomic.Bool + backupAt atomic.Int64 + childReads atomic.Int32 + ids []string +} + +func newPBSNamespacePollFixture(t *testing.T) *pbsNamespacePollFixture { + t.Helper() + m, manager := recoveryIngestTestMonitor(t) + f := &pbsNamespacePollFixture{m: m} + f.backupAt.Store(time.Now().UTC().Add(-time.Hour).Truncate(time.Second).Unix()) + m.state = models.NewState() + m.state.UpdateVMs([]models.VM{ + {ID: makeGuestID("cluster", "node-a", 100), VMID: 100, Instance: "cluster", Node: "node-a", Name: "guest-100", Status: "running"}, + {ID: makeGuestID("cluster", "node-b", 101), VMID: 101, Instance: "cluster", Node: "node-b", Name: "guest-101", Status: "running"}, + }) + for _, id := range []int{100, 101} { + f.ids = append(f.ids, unifiedresources.ProxmoxGuestCanonicalID(unifiedresources.ResourceTypeVM, "cluster", id)) + } + var err error + f.store, err = manager.StoreForOrg("default") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { f.wait(t); _ = f.store.Close() }) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + var data any = []any{} + switch r.URL.Path { + case "/api2/json/version": + if status := f.versionStatus.Load(); status != 0 { + http.Error(w, "fixture version unavailable", int(status)) + return + } + data = map[string]any{"version": "3.4.2"} + case "/api2/json/admin/datastore": + if status := f.datastoreStatus.Load(); status != 0 { + http.Error(w, "fixture discovery unavailable", int(status)) + return + } + if body := f.datastoreBody.Load(); body != nil { + _, _ = w.Write([]byte(*body)) + return + } + data = []map[string]any{{"store": "archive"}} + case "/api2/json/admin/datastore/archive/status": + data = map[string]any{"total": 4096, "used": 1024, "avail": 3072} + case "/api2/json/admin/datastore/archive/namespace": + status := int(f.namespaceStatus.Load()) + if status == 200 { + _, _ = w.Write([]byte(`{"data":`)) // incomplete, not a complete empty list + return + } + if status != 0 { + http.Error(w, "fixture namespace discovery unavailable", status) + return + } + if !f.namespaceRemoved.Load() { + data = []map[string]any{{"ns": "tenant"}} + } + case "/api2/json/admin/datastore/archive/groups": + f.backupReads.Add(1) + id := "100" + if r.URL.Query().Get("ns") == "tenant" { + id = "101" + f.childReads.Add(1) + } + if id != "100" || !f.rootEmpty.Load() { + data = []map[string]any{{"backup-type": "vm", "backup-id": id, "last-backup": f.backupAt.Load(), "backup-count": 1}} + } + case "/api2/json/admin/datastore/archive/snapshots": + f.backupReads.Add(1) + id := r.URL.Query().Get("backup-id") + if r.URL.Query().Get("ns") == "tenant" { + f.childReads.Add(1) + } + data = []map[string]any{{"backup-type": "vm", "backup-id": id, "backup-time": f.backupAt.Load(), "size": 1024, + "files": []string{"index.json.blob"}, "verification": map[string]any{"state": "ok"}, "comment": "guest-" + id}} + } + _ = json.NewEncoder(w).Encode(map[string]any{"data": data}) + })) + t.Cleanup(server.Close) + f.client, err = pbs.NewClient(pbs.ClientConfig{Host: server.URL, TokenName: "fixture@pbs!reader", TokenValue: "fixture-only"}) + if err != nil { + t.Fatal(err) + } + m.config = &config.Config{EnableBackupPolling: true, BackupPollingCycles: 1, PBSInstances: []config.PBSInstance{ + {Name: "pbs-main", Host: server.URL, MonitorDatastores: true, MonitorBackups: true}, + {Name: "pbs-other", Host: server.URL, MonitorDatastores: true, MonitorBackups: true}, + }} + m.lastPBSBackupPoll = make(map[string]time.Time) + m.pollStatusMap = make(map[string]*pollStatus) + return f +} + +func (f *pbsNamespacePollFixture) wait(t *testing.T) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + f.m.mu.RLock() + polling := len(f.m.pbsBackupPollers) > 0 + f.m.mu.RUnlock() + f.m.recoveryIngestMu.Lock() + ingesting := f.m.recoveryIngestRunning + f.m.recoveryIngestMu.Unlock() + if !polling && !ingesting { + return + } + if time.Now().After(deadline) { + t.Fatal("PBS namespace poll/ingest did not finish") + } + time.Sleep(time.Millisecond) + } +} + +func (f *pbsNamespacePollFixture) poll(t *testing.T, instance string) { + t.Helper() + f.m.pollPBSInstance(context.Background(), instance, f.client) + f.wait(t) +} + +func (f *pbsNamespacePollFixture) points(t *testing.T, instance string) map[string]recovery.RecoveryPoint { + t.Helper() + points, _, err := f.store.ListPoints(context.Background(), recovery.ListPointsOptions{Limit: 1000}) + if err != nil { + t.Fatal(err) + } + out := make(map[string]recovery.RecoveryPoint) + for _, point := range points { + if point.RepositoryRef != nil && point.RepositoryRef.Namespace == instance { + out[point.ID] = point + } + } + return out +} + +func (f *pbsNamespacePollFixture) postures(t *testing.T) []recovery.ProtectionPosture { + t.Helper() + postures, total, err := f.store.ListProtectionPostures(context.Background(), recovery.ProtectionPostureQuery{SubjectResourceIDs: f.ids}) + if err != nil || total != len(f.ids) { + t.Fatalf("postures total=%d err=%v", total, err) + } + return postures +} + +func TestPBSNamespaceDiscoveryFailureRetainsRecoveryHistory(t *testing.T) { + for _, status := range []int{http.StatusServiceUnavailable, http.StatusForbidden, http.StatusOK} { + for _, cold := range []bool{false, true} { + t.Run(fmt.Sprintf("status-%d/cold-%t", status, cold), func(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.poll(t, "pbs-main") + before := f.points(t, "pbs-main") + if len(before) != 2 { + t.Fatalf("initial points=%d, want root and tenant", len(before)) + } + for _, p := range f.postures(t) { + if p.State != recovery.ProtectionStateProtected { + t.Fatalf("healthy control=%+v", p) + } + } + if cold { + // Monitor replacement removes live inventory, not durable history. + vms := f.m.state.GetSnapshot().VMs + f.m.state = models.NewState() + f.m.state.UpdateVMs(vms) + } + f.namespaceStatus.Store(int32(status)) + f.backupAt.Add(15 * 60) + reads := f.childReads.Load() + for range 2 { + f.poll(t, "pbs-main") + after := f.points(t, "pbs-main") + // A new root archive is accepted while the source-wide scan + // is partial. Retain both prior historical points, not only + // the unreadable child; reconcile old root history only when + // an ordinary complete enumeration is authoritative again. + if len(after) != len(before)+1 { + t.Errorf("partial discovery points=%d, want retained history plus one new root point (%d)", len(after), len(before)+1) + } + for id, point := range before { + if !reflect.DeepEqual(after[id], point) { + t.Error("historical point was deleted or renewed without an observation") + } + } + freshRootPoints := 0 + for id, point := range after { + if _, historical := before[id]; historical { + continue + } + if point.Details["namespace"] != "" || point.CompletedAt == nil || point.CompletedAt.Unix() != f.backupAt.Load() { + t.Error("new point did not come from the accepted fresh root read") + } + freshRootPoints++ + } + if freshRootPoints != 1 { + t.Errorf("accepted fresh root points=%d, want 1", freshRootPoints) + } + permissions := operationaltrust.EvidencePermissionsUnknown + if status == http.StatusForbidden { + permissions = operationaltrust.EvidencePermissionsPartial + } + for _, p := range f.postures(t) { + if p.State == recovery.ProtectionStateProtected || len(p.ProviderStates) != 1 || + p.ProviderStates[0].HistoryCompleteness != recovery.ProtectionHistoryPartial || + p.ProviderStates[0].Permissions != permissions || p.ProviderStates[0].JobState != recovery.OutcomeWarning { + t.Errorf("discovery failure presented as complete/healthy: %+v", p) + } + } + if f.childReads.Load() != reads { + t.Error("failed discovery guessed a hidden namespace read") + } + } + }) + } + } +} + +func TestPBSNamespaceDiscoveryRecoveryAndScopedRemoval(t *testing.T) { + f := newPBSNamespacePollFixture(t) + f.poll(t, "pbs-other") + other := f.points(t, "pbs-other") + f.poll(t, "pbs-main") + f.namespaceStatus.Store(http.StatusServiceUnavailable) + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 2 { + t.Error("transient discovery failure removed tenant history") + } + f.namespaceStatus.Store(0) + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 2 { + t.Fatal("recovery did not restore namespace scope") + } + for _, p := range f.postures(t) { + for _, provider := range p.ProviderStates { + if provider.HistoryCompleteness != recovery.ProtectionHistoryComplete { + t.Errorf("recovered discovery still incomplete: %+v", p) + } + } + } + f.namespaceRemoved.Store(true) + f.poll(t, "pbs-main") + if got := len(f.points(t, "pbs-main")); got != 1 { + t.Errorf("authoritative namespace removal retained %d points, want root only", got) + } + if !reflect.DeepEqual(other, f.points(t, "pbs-other")) { + t.Error("namespace removal crossed connection scope") + } + f.rootEmpty.Store(true) + f.poll(t, "pbs-main") + if len(f.points(t, "pbs-main")) != 0 { + t.Error("authoritative complete empty scan retained points") + } + if !reflect.DeepEqual(other, f.points(t, "pbs-other")) { + t.Error("empty scan crossed connection scope") + } +} diff --git a/pkg/pbs/client.go b/pkg/pbs/client.go index e0271ef9e4..e0330f58df 100644 --- a/pkg/pbs/client.go +++ b/pkg/pbs/client.go @@ -933,6 +933,17 @@ func (c *Client) GetDatastoresFiltered( Msg("Failed to parse PBS datastore list response") return nil, fmt.Errorf("failed to parse datastore list: %w", err) } + // Only an explicit array establishes the datastore scope. In particular, + // missing/null data and nameless rows are not an authoritative empty list: + // callers use a successful empty enumeration to reconcile backup history. + if datastoreList.Data == nil { + return nil, fmt.Errorf("datastore list response contains no data array") + } + for _, ds := range datastoreList.Data { + if strings.TrimSpace(ds.Store) == "" { + return nil, fmt.Errorf("datastore list response contains an unnamed datastore") + } + } // Now get status for each datastore var datastores []Datastore diff --git a/pkg/pbs/client_datastore_discovery_test.go b/pkg/pbs/client_datastore_discovery_test.go new file mode 100644 index 0000000000..dd8b653e67 --- /dev/null +++ b/pkg/pbs/client_datastore_discovery_test.go @@ -0,0 +1,47 @@ +package pbs + +import ( + "context" + "net/http" + "testing" +) + +func TestGetDatastoresRequiresExplicitNamedScope(t *testing.T) { + for _, body := range []string{ + `null`, `{}`, `{"data":null}`, `{"data":{}}`, `{"data":`, + `{"data":[null]}`, `{"data":[{}]}`, `{"data":[{"store":" "}]}`, + `{"data":[{"store":"valid"},{}]}`, + } { + t.Run(body, func(t *testing.T) { + detailReads := 0 + client, server := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api2/json/admin/datastore" { + detailReads++ + t.Error("malformed scope caused a datastore detail read") + } + _, _ = w.Write([]byte(body)) + }) + defer server.Close() + // Validate the whole scope before filtering or issuing detail reads; + // an exclusion must not turn malformed provider data into absence. + rows, err := client.GetDatastoresFiltered(context.Background(), func(string) bool { return false }) + if err == nil || len(rows) != 0 || detailReads != 0 { + t.Fatalf("invalid scope accepted: rows=%v err=%v detailReads=%d", rows, err, detailReads) + } + }) + } +} + +func TestGetDatastoresAcceptsExplicitEmptyScope(t *testing.T) { + client, server := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api2/json/admin/datastore" { + t.Errorf("empty scope caused a detail read: %s", r.URL.Path) + } + _, _ = w.Write([]byte(`{"data":[]}`)) + }) + defer server.Close() + rows, err := client.GetDatastores(context.Background()) + if err != nil || len(rows) != 0 { + t.Fatalf("explicit empty scope rejected: rows=%v err=%v", rows, err) + } +}