diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 7c25ec01a6..c4bd25b847 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -6932,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_pbs_pmg.go b/internal/monitoring/monitor_pbs_pmg.go index a72f8c6bef..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 } } @@ -361,10 +367,12 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie // 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 { @@ -524,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, discoveryErrors map[string]error, 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, discoveryErrors) - - log.Info(). - Str("instance", inst). - Dur("duration", time.Since(start)). - Msg("Completed background PBS backup polling") - }(datastoreSnapshot, namespaceErrors, 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..588bc47943 --- /dev/null +++ b/internal/monitoring/pbs_datastore_discovery_test.go @@ -0,0 +1,197 @@ +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": + f.m.SetBackupPollingInterval(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_protection_observation_test.go b/internal/monitoring/pbs_protection_observation_test.go index 78f7a85c39..3688689701 100644 --- a/internal/monitoring/pbs_protection_observation_test.go +++ b/internal/monitoring/pbs_protection_observation_test.go @@ -296,6 +296,10 @@ 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 @@ -328,8 +332,20 @@ func newPBSNamespacePollFixture(t *testing.T) *pbsNamespacePollFixture { 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} @@ -347,6 +363,7 @@ func newPBSNamespacePollFixture(t *testing.T) *pbsNamespacePollFixture { 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" @@ -356,6 +373,7 @@ func newPBSNamespacePollFixture(t *testing.T) *pbsNamespacePollFixture { 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) 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) + } +}