Merge core-runtime PBS datastore alert evaluation

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,28 @@
 
 ## Purpose
 
+### PBS datastore alert evaluation belongs to the live poll
+
+After publishing freshly polled PBS datastore storage rows, the poller invokes
+the existing storage alert evaluator with capacity-trend evidence. It must not
+depend on mock ticks or unified metric evaluation: that path does not evaluate
+PBS storage thresholds. Existing storage policy, aliases, suppression and
+connectivity semantics remain alerts-owned; this wiring adds no new policy.
+
+Absent or partial capacity counters must not resolve an existing usage
+incident. Explicit empty-store counters (positive total, zero used, free equal
+to total) may recover it. Repeated polls and subsequent unified alert sync must
+preserve the usage incident identity, including recurrence with alternate PBS
+counter names. Independent backup-posture incidents remain separate.
+
+Verification: `TestPBSPolledCapacityRequiresObservedRecovery` in
+`internal/monitoring/monitor_pbs_coverage_test.go` crosses synthetic HTTP
+decoding, live polling and unified sync. It checks an 80% threshold at 85%,
+missing counter retention, confirmed-empty recovery and 86% recurrence with an
+explicit one-point minimum delta. This is local capacity-path evidence, not
+installed recipient receipt, restart qualification or exhaustive connectivity
+validation.
+
 ### Host-local addresses are not PVE identity
 
 Automatic host/PVE network matching excludes non-global-unicast addresses,

internal/monitoring/monitor_pbs_coverage_test.go

@@ -8,12 +8,14 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"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/unifiedresources"
 	"github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
 )
 
@@ -568,3 +570,82 @@ func TestPBSMetricAvailabilityAlertLifecycle(t *testing.T) {
 		t.Fatal("missing resolved history")
 	}
 }
+
+// Exercise HTTP decoding, poll evaluation and subsequent unified alert sync together.
+// This is synthetic integration evidence, not installed notification receipt.
+func TestPBSPolledCapacityRequiresObservedRecovery(t *testing.T) {
+	var response atomic.Value
+	response.Store(`{"data":{"total":1000,"used":850,"avail":150}}`)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		switch r.URL.Path {
+		case "/api2/json/version":
+			_, _ = w.Write([]byte(`{"data":{"version":"3.4.2"}}`))
+		case "/api2/json/nodes/localhost/status":
+			_, _ = w.Write([]byte(`{"data":{"cpu":0.1,"memory":{"used":100,"total":1000}}}`))
+		case "/api2/json/admin/datastore":
+			_, _ = w.Write([]byte(`{"data":[{"store":"backups"}]}`))
+		case "/api2/json/admin/datastore/backups/status":
+			_, _ = w.Write([]byte(response.Load().(string)))
+		default:
+			_, _ = w.Write([]byte(`{"data":[]}`))
+		}
+	}))
+	defer server.Close()
+	manager := alerts.NewManagerWithDataDir(t.TempDir())
+	defer manager.Stop()
+	manager.UpdateConfig(alerts.AlertConfig{Enabled: true, ActivationState: alerts.ActivationActive, MinimumDelta: 1,
+		TimeThresholds: map[string]int{"storage": 0}, StorageDefault: alerts.HysteresisThreshold{Trigger: 80, Clear: 70}})
+	instance := config.PBSInstance{Name: "pbs-capacity", Host: server.URL, MonitorDatastores: true}
+	monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance})
+	monitor.alertManager = manager
+	client := newPBSHealthTestClient(t, server.URL)
+	adapter := unifiedresources.NewMonitorAdapter(nil)
+	poll := func() {
+		t.Helper()
+		for range 5 {
+			monitor.pollPBSInstance(context.Background(), instance.Name, client)
+			adapter.PopulateFromSnapshot(monitor.state.GetSnapshot())
+			monitor.syncUnifiedResourceAlertsToState(adapter.GetAll())
+		}
+	}
+	poll()
+	active := manager.GetActiveAlerts()
+	if len(active) != 1 || active[0].Type != "usage" || active[0].Value != 85 {
+		t.Fatalf("expected one polled capacity incident: %+v", active)
+	}
+	original := active[0]
+	if original.ResourceID != "pbs-pbs-capacity-backups" {
+		t.Fatalf("unexpected datastore identity: %q", original.ResourceID)
+	}
+	for _, missing := range []string{
+		`{"data":null}`,
+		`{"data":{}}`,
+		`{"data":{"total":1000}}`,
+		`{"data":{"total":1000,"used":0}}`,
+	} {
+		response.Store(missing)
+		poll()
+		active = manager.GetActiveAlerts()
+		if len(active) != 1 || active[0].ID != original.ID || active[0].Value != 85 || !active[0].StartTime.Equal(original.StartTime) || manager.GetResolvedAlert(original.ID) != nil {
+			t.Fatalf("missing capacity %s changed incident: %+v", missing, active)
+		}
+	}
+	response.Store(`{"data":{"total":1000,"used":0,"avail":1000}}`)
+	poll()
+	if len(manager.GetActiveAlerts()) != 0 {
+		t.Fatal("observed empty datastore did not recover")
+	}
+	if resolved := manager.GetResolvedAlert(original.ID); resolved == nil || resolved.Value != 0 || !resolved.StartTime.Equal(original.StartTime) {
+		t.Fatalf("incorrect recovery: %+v", resolved)
+	}
+	// Alternate PBS counter names must feed the same policy and identity.
+	// Stay below the separate 90% backup-posture incident threshold; the
+	// configured minimum delta of one permits this immediate recurrence.
+	response.Store(`{"data":{"total-space":1000,"used-space":860,"avail-space":140}}`)
+	poll()
+	active = manager.GetActiveAlerts()
+	if len(active) != 1 || active[0].ID != original.ID || active[0].Value != 86 || !active[0].StartTime.After(original.StartTime) {
+		t.Fatalf("incorrect recurrent incident: %+v", active)
+	}
+}

internal/monitoring/monitor_pbs_pmg.go

@@ -502,6 +502,14 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
 			pbsStorages = append(pbsStorages, pbsStorage)
 		}
 		m.state.UpdateStorageForInstance("pbs-"+instanceName, pbsStorages)
+		// PBS storage is not evaluated by the unified metric path (which
+		// handles TrueNAS and VMware storage). Evaluate fresh poll observations
+		// here, as the PVE storage poller does, rather than relying on mock ticks.
+		if m.alertManager != nil {
+			for _, storage := range pbsStorages {
+				m.alertManager.CheckStorageWithCapacityTrend(storage, m.storageCapacityTrend(storage, time.Now()))
+			}
+		}
 		log.Debug().
 			Str("instance", instanceName).
 			Int("storageEntries", len(pbsStorages)).