package monitoring import ( "context" "encoding/json" "net/http" "net/http/httptest" "os" "strings" "sync" "testing" "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" ) func TestPollPBSInstanceDoesNotQueryExcludedDatastoreDetails(t *testing.T) { var mu sync.Mutex var requestedPaths []string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { mu.Lock() requestedPaths = append(requestedPaths, r.URL.Path) mu.Unlock() switch r.URL.Path { case "/api2/json/version": _ = json.NewEncoder(w).Encode(map[string]any{ "data": map[string]any{"version": "3.4"}, }) case "/api2/json/nodes/localhost/status": _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{}}) case "/api2/json/admin/datastore": _ = json.NewEncoder(w).Encode(map[string]any{ "data": []map[string]string{ {"store": "internal"}, {"store": "exthdd1500gb"}, }, }) case "/api2/json/admin/datastore/internal/rrd": _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{}}) case "/api2/json/admin/datastore/internal/status": _ = json.NewEncoder(w).Encode(map[string]any{ "data": map[string]any{ "total": 100.0, "used": 25.0, "avail": 75.0, }, }) case "/api2/json/admin/datastore/internal/gc": _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{}}) case "/api2/json/admin/datastore/internal/namespace": _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{}}) default: http.Error(w, "unexpected request", http.StatusInternalServerError) } })) defer server.Close() client, err := pbs.NewClient(pbs.ClientConfig{ Host: server.URL, TokenName: "root@pam!pulse-token", TokenValue: "secret", }) if err != nil { t.Fatalf("NewClient failed: %v", err) } monitor := &Monitor{ config: &config.Config{ PBSInstances: []config.PBSInstance{{ Name: "pbs-excludes", Host: server.URL, MonitorDatastores: true, ExcludeDatastores: []string{"ext*"}, }}, }, state: models.NewState(), authFailures: make(map[string]int), lastAuthAttempt: make(map[string]time.Time), pollStatusMap: make(map[string]*pollStatus), circuitBreakers: make(map[string]*circuitBreaker), } monitor.pollPBSInstance(context.Background(), "pbs-excludes", client) snapshot := monitor.state.GetSnapshot() if len(snapshot.PBSInstances) != 1 { t.Fatalf("PBS instances = %+v, want one", snapshot.PBSInstances) } if snapshot.PBSInstances[0].ID != PBSMonitorResourceID("pbs-excludes") { t.Fatalf("PBS runtime identity = %q, want canonical monitor identity", snapshot.PBSInstances[0].ID) } datastores := snapshot.PBSInstances[0].Datastores if len(datastores) != 1 || datastores[0].Name != "internal" { t.Fatalf("datastores = %+v, want only internal", datastores) } mu.Lock() defer mu.Unlock() for _, path := range requestedPaths { if path == "/api2/json/nodes" { t.Fatalf("API-token poll queried the superuser-only node-name endpoint: %s", path) } if strings.Contains(path, "exthdd1500gb") { t.Fatalf("excluded datastore received a detail request: %s", path) } } } // The hostname a PBS node reports about itself (GET /nodes) is // machine-identity evidence for connected-system grouping: it merges the // host agent running on the PBS machine into the PBS connection even when // the connection was configured by IP or DNS alias. The fetch is optional // collection — capture it when available, never fail the poll on it. func TestPollPBSInstanceCapturesReportedNodeName(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/api2/json/version": _ = json.NewEncoder(w).Encode(map[string]any{ "data": map[string]any{"version": "3.4"}, }) case "/api2/json/nodes": _ = json.NewEncoder(w).Encode(map[string]any{ "data": []map[string]any{{"node": "pbs01"}}, }) case "/api2/json/nodes/localhost/status": _ = json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{}}) default: http.Error(w, "unexpected request", http.StatusInternalServerError) } })) defer server.Close() client, err := pbs.NewClient(pbs.ClientConfig{Host: server.URL, User: "root@pam"}) if err != nil { t.Fatalf("NewClient failed: %v", err) } monitor := &Monitor{ config: &config.Config{ PBSInstances: []config.PBSInstance{{ Name: "pbs-nodename", Host: server.URL, }}, }, state: models.NewState(), authFailures: make(map[string]int), lastAuthAttempt: make(map[string]time.Time), pollStatusMap: make(map[string]*pollStatus), circuitBreakers: make(map[string]*circuitBreaker), } monitor.pollPBSInstance(context.Background(), "pbs-nodename", client) snapshot := monitor.state.GetSnapshot() if len(snapshot.PBSInstances) != 1 { t.Fatalf("PBS instances = %+v, want one", snapshot.PBSInstances) } instance := snapshot.PBSInstances[0] if instance.NodeName != "pbs01" { t.Fatalf("NodeName = %q, want %q", instance.NodeName, "pbs01") } if instance.Status != "online" || instance.ConnectionHealth != "healthy" { t.Fatalf("status = %q health = %q, want online/healthy", instance.Status, instance.ConnectionHealth) } } func TestMonitor_PollPBSInstance_AuthFailure(t *testing.T) { // Setup mock server that returns 401 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusUnauthorized) })) defer server.Close() // Setup client client, err := pbs.NewClient(pbs.ClientConfig{ Host: server.URL, TokenName: "root@pam!token", TokenValue: "secret", }) if err != nil { t.Fatal(err) } // Setup monitor m := &Monitor{ config: &config.Config{ PBSInstances: []config.PBSInstance{ {Name: "pbs-auth-fail", Host: server.URL, MonitorDatastores: true}, }, }, state: models.NewState(), authFailures: make(map[string]int), lastAuthAttempt: make(map[string]time.Time), pollStatusMap: make(map[string]*pollStatus), circuitBreakers: make(map[string]*circuitBreaker), // We need connectionHealth map initialized if SetConnectionHealth uses it? // models.NewState() handles it. } // Execute ctx := context.Background() m.pollPBSInstance(ctx, "pbs-auth-fail", client) // Verify // status should be offline // recordAuthFailure should have been called? // Monitor stores auth failures in memory map `authFailures`. // We can check `m.state.ConnectionHealth` for "pbs-pbs-auth-fail". // Verify manually using snapshot snapshot := m.state.GetSnapshot() if snapshot.ConnectionHealth["pbs-pbs-auth-fail"] { t.Error("Expected connection health to be false") } if len(snapshot.PBSInstances) != 1 { t.Fatalf("expected failed PBS poll to refresh one dashboard projection, got %+v", snapshot.PBSInstances) } if instance := snapshot.PBSInstances[0]; instance.Status != "offline" || instance.ConnectionHealth != "error" { t.Fatalf("failed PBS dashboard projection = %+v, want offline/error", instance) } // Regression: pollStatusMap must record the failure. A defer-arg bug // previously captured pollErr at register-time (always nil), so failed // polls were recorded as success and the connections aggregator reported // broken instances as healthy. if status := m.pollStatusMap["pbs::pbs-auth-fail"]; status == nil { t.Fatal("expected pollStatusMap entry for pbs::pbs-auth-fail, got nil") } else { if !status.LastSuccess.IsZero() { t.Errorf("expected LastSuccess to remain zero on auth failure, got %v", status.LastSuccess) } if status.ConsecutiveFailures == 0 { t.Error("expected ConsecutiveFailures > 0 after auth failure, got 0") } } // We can't easily check authFailures map as it is private and no getter (except checking if it backs off?) } func TestMonitor_PollPBSInstance_DatastoreDetails(t *testing.T) { // Setup mock server server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if strings.Contains(r.URL.Path, "/version") { json.NewEncoder(w).Encode(map[string]interface{}{ "data": map[string]interface{}{"version": "2.0"}, }) return } if strings.Contains(r.URL.Path, "/nodes/localhost/status") { // Fail node status w.WriteHeader(http.StatusInternalServerError) return } if strings.Contains(r.URL.Path, "/admin/datastore") && strings.HasSuffix(r.URL.Path, "/admin/datastore") { // GetDatastores list json.NewEncoder(w).Encode(map[string]interface{}{ "data": []map[string]interface{}{ {"store": "ds1", "comment": "comment1"}, // GetDatastores list returns small subset of fields {"store": "ds2", "comment": "comment2"}, {"store": "ds-error", "comment": "status failure"}, }, }) return } if strings.Contains(r.URL.Path, "/status") { // Datastore Status var data map[string]interface{} if strings.Contains(r.URL.Path, "ds1") { data = map[string]interface{}{"total": 100.0, "used": 50.0, "avail": 50.0} } else if strings.Contains(r.URL.Path, "ds2") { data = map[string]interface{}{"total-space": 200.0, "used-space": 100.0, "avail-space": 100.0, "deduplication-factor": 1.5, "status": "read_only"} } else if strings.Contains(r.URL.Path, "ds-error") { http.Error(w, "datastore unavailable", http.StatusInternalServerError) return } json.NewEncoder(w).Encode(map[string]interface{}{"data": data}) return } if strings.Contains(r.URL.Path, "/rrd") { // RRD json.NewEncoder(w).Encode(map[string]interface{}{"data": []interface{}{}}) return } if strings.Contains(r.URL.Path, "/namespace") { // ListNamespaces if strings.Contains(r.URL.Path, "ds1") { // DS 1: Fail namespaces w.WriteHeader(http.StatusInternalServerError) return } if strings.Contains(r.URL.Path, "ds2") { // DS 2: Varied namespaces json.NewEncoder(w).Encode(map[string]interface{}{ "data": []map[string]interface{}{ {"ns": "ns1"}, {"path": "ns2"}, // alternate field {"name": "ns3"}, // alternate field }, }) return } } // Catch-all success for rrd/status calls from client.GetDatastores (it calls internal methods) // Wait, client.GetDatastores calls /api2/json/admin/datastore // client.ListNamespaces calls /api2/json/admin/datastore/{store}/namespace? // No, client.ListNamespaces: req to /admin/datastore/%s/namespace w.WriteHeader(http.StatusOK) json.NewEncoder(w).Encode(map[string]interface{}{"data": []interface{}{}}) })) defer server.Close() client, err := pbs.NewClient(pbs.ClientConfig{Host: server.URL, TokenName: "root@pam!token", TokenValue: "val"}) if err != nil { t.Fatalf("Failed to create client: %v", err) } m := &Monitor{ config: &config.Config{ PBSInstances: []config.PBSInstance{ {Name: "pbs-details", Host: server.URL, MonitorDatastores: true}, }, }, state: models.NewState(), authFailures: make(map[string]int), lastAuthAttempt: make(map[string]time.Time), pollStatusMap: make(map[string]*pollStatus), circuitBreakers: make(map[string]*circuitBreaker), } m.pollPBSInstance(context.Background(), "pbs-details", client) // Verify State snapshot := m.state.GetSnapshot() var inst *models.PBSInstance for _, i := range snapshot.PBSInstances { if i.Name == "pbs-details" { copy := i inst = © break } } if inst == nil { t.Fatal("Instance not found") } if len(inst.Datastores) != 3 { t.Errorf("Expected 3 datastores, got %d", len(inst.Datastores)) } // Check DS2 size calculation and status propagation. var ds2 *models.PBSDatastore var dsError *models.PBSDatastore for _, ds := range inst.Datastores { if ds.Name == "ds2" { copy := ds ds2 = © } if ds.Name == "ds-error" { copy := ds dsError = © } } if ds2 != nil { if ds2.Total != 200 { t.Errorf("Expected DS2 total 200, got %d", ds2.Total) } if ds2.Status != "read_only" { t.Errorf("Expected DS2 status read_only, got %q", ds2.Status) } if len(ds2.Namespaces) != 4 { t.Errorf("Expected 4 namespaces for DS2, got %d", len(ds2.Namespaces)) } } else { t.Error("DS2 not found") } if dsError == nil { t.Error("ds-error not found") } else { if dsError.Status != "unavailable" { t.Errorf("Expected ds-error status unavailable, got %q", dsError.Status) } if !strings.Contains(dsError.Error, "Failed to get status") { t.Errorf("Expected ds-error to preserve status error, got %q", dsError.Error) } } } func TestPBSJobHealthEvidenceKeepsFreshnessSeparateFromConfidence(t *testing.T) { observedAt := time.Unix(1700007200, 0).UTC() facts := []pbs.JobHealthEvidence{ { ID: "sync-remote-a", Family: "sync", Store: "fast", LastRunState: "OK", LastRunUPID: "UPID:sync:1", LastRunEndtime: 1700000000, NextRun: 1700003600, UPID: "UPID:sync:1", TaskStatus: "OK", Confidence: "direct-task-match", EvidenceSource: pbs.JobEvidenceSourcePBSJobConfig, EvidenceScope: pbs.JobEvidenceScopeConfiguredJob, }, } evidence := pbsJobHealthEvidenceFromFacts(facts, observedAt) if len(evidence) != 1 { t.Fatalf("expected 1 evidence item, got %d", len(evidence)) } got := evidence[0] if got.Confidence != "direct-task-match" { t.Fatalf("confidence = %q, want direct-task-match", got.Confidence) } if got.EvidenceSource != pbs.JobEvidenceSourcePBSJobConfig || got.EvidenceScope != pbs.JobEvidenceScopeConfiguredJob { t.Fatalf("evidence source/scope = %q/%q, want configured PBS job", got.EvidenceSource, got.EvidenceScope) } if got.Freshness.State != "overdue" { t.Fatalf("freshness state = %q, want overdue", got.Freshness.State) } if got.Posture != "warning" || got.PostureReason != "job-overdue" { t.Fatalf("posture = %q/%q, want warning/job-overdue", got.Posture, got.PostureReason) } if got.LastRunState != "OK" || got.LastRunUPID != "UPID:sync:1" || got.LastRunEndtime != 1700000000 || got.NextRun != 1700003600 { t.Fatalf("raw PBS last-run fields were not preserved: %+v", got) } } func TestPBSJobHealthEvidenceLabelsBackupTasksAsObservedOnly(t *testing.T) { observedAt := time.Unix(1700007200, 0).UTC() facts := []pbs.JobHealthEvidence{ { ID: "backup:vm/100", Family: "backup", Store: "fast", UPID: "UPID:backup:1", WorkerType: "backup", WorkerID: "vm/100", TaskStatus: "OK", TaskStartTime: 1700000000, TaskEndTime: 1700000060, Confidence: pbs.JobEvidenceConfidenceObservedBackupTask, EvidenceSource: pbs.JobEvidenceSourcePBSTaskHistory, EvidenceScope: pbs.JobEvidenceScopeObservedTask, }, } evidence := pbsJobHealthEvidenceFromFacts(facts, observedAt) if len(evidence) != 1 { t.Fatalf("expected 1 evidence item, got %d", len(evidence)) } got := evidence[0] if got.Family != "backup" || got.Confidence != pbs.JobEvidenceConfidenceObservedBackupTask { t.Fatalf("expected observed backup task evidence, got %+v", got) } if got.EvidenceSource != pbs.JobEvidenceSourcePBSTaskHistory || got.EvidenceScope != pbs.JobEvidenceScopeObservedTask { t.Fatalf("evidence source/scope = %q/%q, want PBS task-history observed-task", got.EvidenceSource, got.EvidenceScope) } if got.Schedule != "" || got.NextRun != 0 || !got.Freshness.NextRun.IsZero() || got.Freshness.State == "scheduled" { t.Fatalf("backup task evidence must not become scheduled backup compliance: %+v", got) } } // TestPBSAndPMGPollSkipDisabledInstances asserts that the PBS and PMG poll // entry points short-circuit when their resolved instance config carries // `Disabled: true`. This is a source-level guardrail for the discovery // provider surface: the unified connections ledger surfaces `Disabled` as // `paused`, and the PBS/PMG pollers must not drive live API calls or // surface ingest while that flag is set, across restarts or reloads. func TestPBSAndPMGPollSkipDisabledInstances(t *testing.T) { data, err := os.ReadFile("monitor_pbs_pmg.go") if err != nil { t.Fatalf("failed to read monitor_pbs_pmg.go: %v", err) } source := string(data) // Both PBS and PMG poll flows must explicitly guard on Disabled. if count := strings.Count(source, "if instanceCfg.Disabled {"); count < 2 { t.Fatalf("monitor_pbs_pmg.go must contain the Disabled-skip guard in both PBS and PMG poll entry points; found %d", count) } // The guards must short-circuit the poll with an early return so no // downstream API client is constructed for a paused instance. for _, snippet := range []string{ "Skipping PBS poll: instance is paused", "Skipping PMG poll: instance is paused", } { if !strings.Contains(source, snippet) { t.Fatalf("monitor_pbs_pmg.go must emit debug-log %q when skipping a disabled instance so operators can correlate paused ledger rows with runtime behavior", snippet) } } }