Distinguish failed PBS discovery from an empty backup scope

Carry failed and accepted-empty discovery through ordinary serialized backup polling. Preserve unreadable history, reconcile only confirmed absence, and reject malformed datastore envelopes before destructive reconciliation.

Contract-Neutral: Enforces existing PBS scope-evidence semantics; no API, model, policy or permission change.
Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-23 23:09:27 +01:00
parent a57ec58ef6
commit aaa6aaa807
6 changed files with 385 additions and 87 deletions
@@ -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
+89 -87
View File
@@ -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:<PBS source ID>). When no
// host agent can be corroborated, the Backups drawer still has useful history
@@ -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)
}
}
}
@@ -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)
+11
View File
@@ -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
@@ -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)
}
}