From 19d8b4f81a81f71dd137b90b5b12de15e50fd65b Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Thu, 8 Oct 2026 06:27:50 +0100 Subject: [PATCH] Resolve saved hosts an operator link joins to the link's primary After a restart, an agent an operator linked into a live resource, such as an agent inside a vSphere VM, exists only through saved-host continuity until it reports again. Three consumers then disagreed about what a reference to it meant. The read state kept the saved agent as its own offline row and resolved references to it there. The resources API, which seeds a registry from that listing and re-applies the links, folded it into the VM and merged its stale telemetry. Alert intent resolved through the published registry, where the agent is absent, so its agent: alerts kept their literal reference. Maintenance or NeverAutoRemediate set on the VM reached neither the agent's alerts nor its Patrol findings in that window, while the API said it applied. An operator link names one identity; telemetry separation is the only reason the saved row stays apart. Link application now holds a saved row beside the link's primary instead of merging it, in every registry. A reference that resolves to the saved row (its canonical ID, agent:, its machine ID or an alias it shares with the primary) resolves to the primary, as a folded row's references do, while each row keeps its own telemetry. An in-memory marker on rows continuity introduced rides clones, so the resources API's registry holds them the same way. A saved primary leaves a live member its own identity. Alert intent now resolves references and ancestors through the published registry with saved hosts overlaid, the read state Patrol resolves through. That also lets intent set on an unlinked saved host reach its offline alert. The resolver runs under the alert manager's lock once per metric check, so the overlay is kept per published registry generation, publication builds it before alerts are evaluated, and a build that overlaps a mock-mode switch is discarded. --- .../v6/internal/subsystems/agent-lifecycle.md | 7 +- .../v6/internal/subsystems/api-contracts.md | 26 +- .../v6/internal/subsystems/monitoring.md | 36 +++ .../internal/subsystems/storage-recovery.md | 13 + .../internal/subsystems/unified-resources.md | 66 ++++- .../api/resources_link_continuity_test.go | 124 ++++++++ .../issue1913_host_continuity_test.go | 46 ++- internal/monitoring/monitor.go | 1 + internal/monitoring/monitor_alert_intent.go | 130 ++++++++- .../monitoring/monitor_alert_intent_test.go | 149 ++++++++++ internal/monitoring/monitor_alert_sync.go | 5 + ...nitor_host_agent_removal_lifecycle_test.go | 29 ++ internal/unifiedresources/clone_test.go | 15 + internal/unifiedresources/monitor_adapter.go | 24 ++ internal/unifiedresources/registry.go | 115 +++++++- internal/unifiedresources/resolve_test.go | 267 ++++++++++++++++++ internal/unifiedresources/types.go | 7 + 17 files changed, 1025 insertions(+), 35 deletions(-) create mode 100644 internal/api/resources_link_continuity_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 4fb1902f6f..68915c88a5 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -4356,9 +4356,10 @@ row's state while the read state folds the agent, as the resources API does. A finding naming a saved host that has not reported since a restart by its agent reference reads the row stored under that host's canonical ID, because the read state overlays host continuity as the -resources API listing does; a saved host joined to its guest by a link -stays its own row there, so its findings read the agent's own state -until it reports again. The +resources API listing does. A saved host joined to its guest by a link +stays its own row there, without lending the guest its saved payload, +but the registry holds it under the guest, so its findings read the +guest's state, as they did before the restart. The resolution only reads identity. It does not treat the folded agent as removed, offline or unenrolled, does not turn a continuity-only host into a live sighting, and grants no heartbeat, command, install or diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 7af172ed10..a722d20815 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -5583,17 +5583,21 @@ reference it cannot place is read as given. Resolving first is what keeps Patrol on the row the resources API writes for the same reference: an exact read first would let a row left under a folded ID, even a default-valued one, hide the maintenance window the -operator then set on the merged resource. Three cases still resolve +operator then set on the merged resource. A saved host that has not +reported since a restart but is linked into a live resource resolves +alike everywhere: the read state keeps it as its own offline row, so its +saved telemetry never reaches the resource, but holds it under the +link's primary, and the resources API's registry, seeded from that +listing, holds it the same way instead of folding it. A finding, an +alert and an operator-state write naming the saved agent by its +canonical ID or `agent:` therefore all reach the resource the +agent was folded into before the restart. Only when the saved agent is +itself the link's primary and the resource is not its guest does each +keep its own identity until the agent reports. Two cases still resolve differently in the resources API, whose registry re-ingests that listing: references to records it replays from supplemental providers -on top of the listing; manual links in mock mode, whose view applies -none; and a saved host that has not reported since a restart but is -linked into a live resource. The read state keeps that host as its own -offline row, because continuity records never join links, while the -resources API's ingest re-applies the link and folds it, so Patrol -reads the saved agent's own row there, as alert intent does for that -agent's canonical ID, and state set only on the link primary does not -reach the agent's findings until it reports again. The read +on top of the listing, and manual links in mock mode, whose view +applies none. The read state answers from a registry it keeps (a continuity overlay is shared and reused for up to two seconds while the published registry and the saved hosts are unchanged), so findings do not each clone the unified @@ -5602,7 +5606,9 @@ left without a resource store, which the router never does, builds a snapshot registry per call. Pinned by `TestPatrolFindingOperatorStateFollowsLinkFoldedReference` and `TestPatrolFindingOperatorStateResolvesContinuityBackedHost` in -`internal/api/ai_handlers_more_test.go` and +`internal/api/ai_handlers_more_test.go`, +`TestResourcesAPIAndPatrolAgreeOnContinuityAgentLinkedIntoGuest` in +`internal/api/resources_link_continuity_test.go` and `TestContract_FindingsResourceOperatorStateProviderIsWired` in `internal/api/contract_test.go`. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index abcb21555c..d5d6043dc6 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -5386,6 +5386,42 @@ canonical aliases retain their precedence and ambiguous aliases remain unresolved. This avoids cloning a full resource for every policy lookup while keeping alert intent tied to the same identity rules as resource reads. +Alert intent resolves references and ancestors through the read state Patrol +resolves through and the resources API seeds from: the published registry with +saved hosts that have not reported since a restart overlaid +(`operatorIntentIdentity`). A saved host exists only there, so against the +published registry alone its alerts (`agent:`) read their literal +reference after a restart and missed intent set on the host, and, when an +operator link joins it to a live resource, intent set on the link primary that +the registry now resolves the saved host to. Alerts resolve under the alert +manager's lock, once per metric check, so the overlay is kept for the published +registry generation it was built from. Publication builds it before alert +evaluation, outside that lock; a lookup that races ahead of publication, or the +first one after leaving mock mode, builds it under the lock. A generation is +the registry the adapter publishes and when it last changed +(`MonitorAdapter.Generation`), since two publications can share a +`LastRebuiltAt`. A finished build is stored only if that generation is still +the published one and the mock-mode epoch it started in is current, checked and +stored inside the mock-mode fence, so a build that overlapped a mode switch is +discarded. A lookup made while mock mode is on drops the stored overlay; a +switch to mock and back with no lookup in between keeps it, since it still +describes the published generation. A saved-host change that arrives without a +new generation shows on the next one. Mock mode carries no saved hosts: lookups +then resolve against the published registry as before. +`monitor_alert_intent_test.go` +(`TestOperatorIntentIdentityKeepsOneOverlayPerGeneration`, +`TestOperatorIntentReachesSavedHostAlertsAfterRestart`) pins the reuse, a +generation sharing or preceding the last one's timestamp, the mock switch, a +rebuild made from inside an admitted fence call, and an unlinked saved host's +intent reaching its offline alert. +`internal/monitoring/issue1913_host_continuity_test.go` +(`TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity`) pins a saved +agent's offline alert reading maintenance set on the guest it is linked into, +and `monitor_host_agent_removal_lifecycle_test.go` +(`TestHostAgentRemovalLifecycleHonorsSavedHostIntentAfterRestart`) pins the +restarted monitor's own host-offline check raising no alert for a saved host +the operator marked intentionally offline. + Alert restore precedes resource-store attachment during startup. Once the adapter attaches the persisted operator-policy resolver, monitoring asks Alerts to reconcile the restored set and refreshes shared alert state if it changed. diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 4306020392..011f1af2a7 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -6133,6 +6133,19 @@ no backup, snapshot, restore, retention, cleanup, or verification authority. The storage/recovery resource query and its admission, freshness, and persistence contracts are unchanged. +### Saved-host link holds open no recovery path + +`internal/unifiedresources/types.go` gains an unexported +`Resource.continuityOnly`: it marks a row saved-host continuity introduced (a +machine that has not reported since a restart), so an operator link holds the +row beside its primary instead of merging it, and, unless the saved row is +itself the link's effective primary (an agent linked into its guest makes the +guest primary), reference reads by its IDs resolve to the primary. It is +in-memory only, never serialized, persisted or exposed on the API, and storage +and recovery code neither reads it nor resolves references through the +registry, so recovery ownership, restore targeting and protection grouping are +unchanged. + ### Token creation and deletion preserve restart-time persistence truth The shared `internal/api` token creation path treats its persisted inventory as diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 02826378b2..3e5ec20f32 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -6617,10 +6617,46 @@ separate. supplemental refresh, overlay, node with auto-linked agent), `TestSourceMergesJudgeFreshnessByTheIngestStaleThresholds` and `TestMetricMergeFreshnessMatchesTheStalePass`. -- Continuity records are the one record ingest that never joins links. A - saved enrollment only fills an absent machine, so it stays its own row - instead of lending its linked guest an offline verdict and an old agent - payload. +- Continuity records never merge into a link. A saved enrollment only fills + an absent machine, so it stays its own row instead of lending its linked + guest an offline verdict and an old agent payload. The link still names one + identity, so the pass holds the saved row beside the primary + (`holdLinkedResourceLocked`): every reference that resolves to the saved + row (its canonical ID, `agent:`, its machine ID or an alias it + shares with the primary) resolves to the primary instead, through + `ResolveReferenceID`, `GetByReference` and the alias index, as a + reference to a folded resource does. An availability check resolved after + the hold, as the resources API replays checks onto its seed, attaches to + the primary too; the monitor's read state does not re-resolve checks after + continuity, so a check naming the saved agent stays as the published + registry left it. Reference lookups by the saved agent's ID (a resource + detail read, the agent resource context), operator-state reads and writes + through the resources API, Patrol's finding projection and alert intent + therefore all reach the guest the agent was folded into before the + restart, while each row keeps its own telemetry. Saved rows carry an + in-memory marker (`Resource.continuityOnly`) that rides clones, so the + resources API, which + seeds a registry from the monitor's read state and re-applies the links, + holds them too instead of merging the saved agent into the guest; an + observation that later merges into the row clears it. A live member never + resolves to a saved one: when the saved row is the link's primary (a link + made from an agent's page to a resource that is not its guest), both rows + answer for themselves until the agent reports and the link folds them, so + intent set on the pair under the agent's ID does not reach the live + member's references in that window. Holds are recomputed on every link + pass; a hold whose primary a later link folds follows the fold, a saved + row held under a saved primary follows that primary's hold, and a cycle of + saved rows resolves each to itself. Exact reads are untouched: `Get` still + returns the saved row, so the saved agent's own drawer reads its own facet + bundle and history, while its operator-state section reads and writes the + guest's state. +- Holds cover a link whose two members are both listed. A saved member + whose partner an earlier link already folded into a third resource, such + as an agent linked into a VM that is itself linked into another resource, + is not held: the monitor's published registry folds the VM before + continuity adds the agent, so the agent answers for itself until it + reports. Live chains already depend on the order their links are stored + in, which this does not change. - A link merge keeps the TrueNAS and vSphere payloads the primary lacks, so an agent chosen as primary over a TrueNAS system still carries the system's facet, and the system's pools re-parent to the merged row. @@ -6693,9 +6729,8 @@ separate. state before it reads, so a finding raised before the link reads the merged row's maintenance and monitoring mode, not an operator-state row left under the folded ID, whenever that read state folds the - agent. A saved agent that has not reported since a restart is not - folded there (continuity records never join links), so its findings - read its own row until it reports again. History does + agent, or holds the saved agent beside the guest after a restart + (above). History does not follow the fold: a history binding persists, and `expandHistoryAliases` walks bindings in both directions, so binding a folded ID would join the two journals for good. Folded IDs are never @@ -6750,7 +6785,22 @@ thresholds on the rebuild, supplemental and overlay paths, and `internal/monitoring/issue1913_host_continuity_test.go` (`TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity`) pins the published inventory with no continuity, an unrelated absent machine and the -linked agent's own saved enrollment. +linked agent's own saved enrollment, and in each case that the read state, +alert intent and a registry seeded from the read state resolve the agent to +the guest without merging the saved agent into it. +`resolve_test.go` pins the hold in the read state and the +resources-API registry, the fold once the agent reports +(`TestContinuityAgentLinkedIntoLiveGuestAnswersToTheGuest`), a saved primary +that leaves the live member its own identity +(`TestContinuityPrimaryLeavesLiveLinkMemberItsOwnIdentity`), an observation +merging into a saved row (`TestObservationMergedIntoASavedRowLetsItsLinkFold`) +and holds through later folds, saved primaries and cycles +(`TestLinkHoldsFollowTheirPrimaryThroughThePass`); `clone_test.go` +(`TestCloneResourceKeepsContinuityMarker`) pins the marker riding clones and +resource seeding. +`internal/api/resources_link_continuity_test.go` +(`TestResourcesAPIAndPatrolAgreeOnContinuityAgentLinkedIntoGuest`) pins the +resources API's write and Patrol's read on the same store. ### Provider link network corroboration diff --git a/internal/api/resources_link_continuity_test.go b/internal/api/resources_link_continuity_test.go new file mode 100644 index 0000000000..031102797c --- /dev/null +++ b/internal/api/resources_link_continuity_test.go @@ -0,0 +1,124 @@ +package api + +import ( + "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/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" +) + +// After a restart, an agent an operator linked into a live vSphere VM exists +// only through saved-host continuity until it reports again. The link names +// one identity, so the resources API stores operator state written through the +// agent's references on the VM and Patrol reads it there, on the same store. +// Neither the monitor's read state nor the registry the resources API seeds +// from it merges the saved agent into the VM: the agent keeps its own offline +// row and the VM keeps what vSphere reports. +func TestResourcesAPIAndPatrolAgreeOnContinuityAgentLinkedIntoGuest(t *testing.T) { + now := time.Now().UTC() + monitorCfg := &config.Config{DataPath: t.TempDir()} + seedMonitor, err := monitoring.New(monitorCfg) + if err != nil { + t.Fatalf("monitoring.New seed monitor: %v", err) + } + t.Cleanup(seedMonitor.Stop) + if _, err := seedMonitor.ApplyHostReport(agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "agent-app-guest", Version: "6.0.0-rc.1", IntervalSeconds: 30}, + Host: agentshost.HostInfo{ID: "host-app-guest", MachineID: "machine-app-guest", Hostname: "app-guest", Platform: "linux"}, + Timestamp: now, + }, &config.APITokenRecord{ID: "token-1", Name: "Token One"}); err != nil { + t.Fatalf("ApplyHostReport seed continuity: %v", err) + } + monitor, err := monitoring.New(monitorCfg) + if err != nil { + t.Fatalf("monitoring.New restarted monitor: %v", err) + } + t.Cleanup(monitor.Stop) + + vmRecords := []unifiedresources.IngestRecord{{ + SourceID: "vc-1:vm:vm-42", + Resource: unifiedresources.Resource{ + Type: unifiedresources.ResourceTypeVM, + Technology: "vmware", + Name: "app-guest", + Status: unifiedresources.StatusOnline, + LastSeen: now, + VMware: &unifiedresources.VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"}, + }, + Identity: unifiedresources.ResourceIdentity{Hostnames: []string{"app-guest"}}, + }} + ids := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + ids.PopulateSnapshotAndSupplemental(models.StateSnapshot{LastUpdate: now}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceVMware: vmRecords, + }) + if len(ids.VMs()) != 1 { + t.Fatalf("vSphere estate has %d VMs, want 1", len(ids.VMs())) + } + vmID := ids.VMs()[0].ID() + agentID := unifiedresources.MachineIdentityCanonicalID(unifiedresources.ResourceTypeAgent, "machine-app-guest") + + handlers := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) + t.Cleanup(func() { _ = handlers.CloseStores() }) + store, err := handlers.getStore("default") + if err != nil { + t.Fatalf("resource store: %v", err) + } + // Linked from the agent's page, so the agent is the stored primary; an + // agent inside a guest still folds into the guest. + if err := store.AddLink(unifiedresources.ResourceLink{ResourceA: agentID, ResourceB: vmID, PrimaryID: agentID}); err != nil { + t.Fatalf("add link: %v", err) + } + monitor.SetResourceStore(unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store))) + vsphere := newTestSupplementalUsageProvider(unifiedresources.SourceVMware) + vsphere.settleAtWithRecords(now, vmRecords) + monitor.SetSupplementalRecordsProvider(unifiedresources.SourceVMware, vsphere) + seed := &countingUnifiedSeedProvider{listing: monitor.UnifiedResourceSnapshot} + handlers.SetStateProvider(seed) + router := &Router{monitor: monitor, resourceHandlers: handlers} + + listing, _ := monitor.UnifiedResourceSnapshot() + readStateRows := map[string]unifiedresources.Resource{} + for _, resource := range listing { + readStateRows[resource.ID] = resource + } + if agent, ok := readStateRows[agentID]; !ok || agent.Status != unifiedresources.StatusOffline { + t.Fatalf("read state agent row = %+v (listed=%v), want the saved agent's own offline row", agent, ok) + } + if guest, ok := readStateRows[vmID]; !ok || guest.Agent != nil { + t.Fatalf("read state guest = %+v (listed=%v), want vSphere's VM without the saved agent", guest, ok) + } + + registry, err := handlers.buildRegistry("default") + if err != nil { + t.Fatalf("resources API registry: %v", err) + } + agent, listed := registry.Get(agentID) + if !listed || agent.Status != unifiedresources.StatusOffline { + t.Fatalf("resources API agent row = %+v (listed=%v), want the saved agent's own offline row", agent, listed) + } + if guest, ok := registry.Get(vmID); !ok || guest.Agent != nil || guest.Status != unifiedresources.StatusOnline { + t.Fatalf("resources API guest = %+v (listed=%v), want vSphere's online VM without the saved agent", guest, ok) + } + + refs := []string{agentID, "agent:host-app-guest"} + for _, ref := range refs { + if got := putOperatorStateThroughAPI(t, handlers, ref, activeMaintenanceBody(now, "guest migration")); got != vmID { + t.Fatalf("maintenance written through %s landed on %q, want the linked VM %s", ref, got, vmID) + } + } + seed.clones = 0 + provider := router.patrolResourceOperatorStateProvider("default") + for _, ref := range append(refs, vmID) { + projection, ok := provider.OperatorStateProjection(ref, now) + if !ok || projection.MaintenanceWindow == nil || projection.MaintenanceWindow.Reason != "guest migration" { + t.Fatalf("finding on %s projected %+v (ok=%v), want the linked VM %s's maintenance window", ref, projection, ok, vmID) + } + } + if seed.clones != 0 { + t.Fatalf("finding projection cloned the unified listing %d times; it must resolve through the monitor's read state", seed.clones) + } +} diff --git a/internal/monitoring/issue1913_host_continuity_test.go b/internal/monitoring/issue1913_host_continuity_test.go index 192540bc5c..b5b93a56c5 100644 --- a/internal/monitoring/issue1913_host_continuity_test.go +++ b/internal/monitoring/issue1913_host_continuity_test.go @@ -4,6 +4,7 @@ import ( "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" @@ -97,7 +98,11 @@ func TestHostContinuityPreservesLiveAgentAfterReenrollment(t *testing.T) { // broadcasts and the resources API seeds from flips with an unrelated absent // machine. A saved enrollment only fills an absent machine: when the linked // agent itself is gone, its continuity row stays separate and the guest keeps -// what vSphere reports. +// what vSphere reports. The link still names one identity, so references to +// the agent reach the guest in every case: alert intent reads maintenance set +// on the guest for the agent's alerts, and the resources API, which seeds a +// registry from the read state and re-applies the links, agrees without +// merging the saved agent into the guest. func TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity(t *testing.T) { now := time.Now().UTC() guestAgent := models.Host{ @@ -153,11 +158,19 @@ func TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity(t *testing.T if err := links.AddLink(unifiedresources.ResourceLink{ResourceA: vmID, ResourceB: agentID, PrimaryID: vmID}); err != nil { t.Fatalf("add link: %v", err) } + start, end := now.Add(-time.Hour), now.Add(time.Hour) + if err := links.SetResourceOperatorState(unifiedresources.ResourceOperatorState{ + CanonicalID: vmID, MaintenanceStartAt: &start, MaintenanceEndAt: &end, MaintenanceReason: "guest migration", + }); err != nil { + t.Fatalf("set guest maintenance: %v", err) + } state := models.NewState() if tc.reporting { state.Hosts = []models.Host{guestAgent} } - m := &Monitor{state: state} + manager := alerts.NewManagerWithDataDir(t.TempDir()) + t.Cleanup(manager.Stop) + m := &Monitor{state: state, alertManager: manager} if len(tc.continuity) > 0 { continuity := config.NewHostContinuityStore(t.TempDir(), nil) for _, entry := range tc.continuity { @@ -188,6 +201,35 @@ func TestManualLinkToSupplementalGuestHoldsWithAndWithoutContinuity(t *testing.T if guest == nil { t.Fatalf("linked VM %s missing from published inventory", vmID) } + preview, err := manager.PreviewIntentPolicy(alerts.AlertIntentPolicyPreviewRequest{ + ResourceID: "agent:" + guestAgent.ID, ResourceType: "agent", + Signal: string(alerts.AlertIntentSignalOffline), ConditionActive: true, + }) + if err != nil { + t.Fatal(err) + } + if preview.Reason != "operator_maintenance" { + t.Fatalf("agent offline alert intent = %+v, want the guest's maintenance window", preview) + } + readState, ok := m.GetUnifiedReadStateOrSnapshot().(interface { + ResolveCanonicalResourceID(string) (string, bool) + }) + if !ok { + t.Fatal("monitor read state cannot resolve references") + } + if resolved, ok := readState.ResolveCanonicalResourceID("agent:" + guestAgent.ID); !ok || resolved != vmID { + t.Fatalf("read state resolved the agent to %q (%v), want the guest %s", resolved, ok, vmID) + } + resourcesAPI := unifiedresources.NewRegistry(links) + resourcesAPI.IngestResources(resources) + for _, ref := range []string{agentID, "agent:" + guestAgent.ID} { + if resolved, ok := resourcesAPI.ResolveReferenceID(ref); !ok || resolved != vmID { + t.Fatalf("resources API resolved %q to %q (%v), want the guest %s", ref, resolved, ok, vmID) + } + } + if apiGuest, ok := resourcesAPI.Get(vmID); !ok || (!tc.reporting && apiGuest.Agent != nil) { + t.Fatalf("resources API guest = %+v (listed=%v), want it free of the saved enrollment", apiGuest, ok) + } if tc.reporting { if agent != nil { t.Fatalf("linked agent %s still published standalone", agentID) diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index fca79736c0..550f922c64 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1249,6 +1249,7 @@ type Monitor struct { metricsHistory *MetricsHistory metricsStore *metrics.Store // Persistent SQLite metrics storage alertManager *alerts.Manager + operatorIntentIdentity atomic.Pointer[operatorIntentIdentity] // read state alert intent resolves through (monitor_alert_intent.go) alertResolvedAICallback func(*alerts.Alert) alertTriggeredAICallback func(*alerts.Alert) alertPushCallback func(*alerts.Alert) diff --git a/internal/monitoring/monitor_alert_intent.go b/internal/monitoring/monitor_alert_intent.go index e00ff0cada..1b5648b4ad 100644 --- a/internal/monitoring/monitor_alert_intent.go +++ b/internal/monitoring/monitor_alert_intent.go @@ -2,9 +2,11 @@ package monitoring import ( "strings" + "sync" "time" "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rs/zerolog/log" ) @@ -27,9 +29,20 @@ func (m *Monitor) installOperatorIntentResolver(store ResourceStoreInterface) { if m == nil || m.alertManager == nil { return } + identity := m.newOperatorIntentIdentity(store) + m.operatorIntentIdentity.Store(identity) + identity.refresh() + identityStore := func() any { + if identity == nil { + return store + } + return identity.current() + } identityReader, hasIdentityReader := store.(resourceIntentIdentityReader) if hasIdentityReader && identityReader != nil { - m.alertManager.SetResourceIntentIdentityResolver(identityReader.ResolveCanonicalResourceID) + m.alertManager.SetResourceIntentIdentityResolver(func(ref string) (string, bool) { + return identityStore().(resourceIntentIdentityReader).ResolveCanonicalResourceID(ref) + }) } else { m.alertManager.SetResourceIntentIdentityResolver(nil) } @@ -38,10 +51,11 @@ func (m *Monitor) installOperatorIntentResolver(store ResourceStoreInterface) { m.alertManager.SetOperatorIntentContextResolver(nil) return } - ancestorReader, hasAncestorReader := store.(resourceIntentAncestorReader) + _, hasAncestorReader := store.(resourceIntentAncestorReader) m.alertManager.SetOperatorIntentContextResolver(func(resourceID string, now time.Time) (alerts.OperatorIntentContext, bool) { + identity := identityStore() if hasIdentityReader { - if canonicalID, found := identityReader.ResolveCanonicalResourceID(resourceID); found { + if canonicalID, found := identity.(resourceIntentIdentityReader).ResolveCanonicalResourceID(resourceID); found { resourceID = canonicalID } } @@ -81,7 +95,7 @@ func (m *Monitor) installOperatorIntentResolver(store ResourceStoreInterface) { applyActiveMaintenance(state, resourceID, false) } if hasAncestorReader { - for _, ancestorID := range ancestorReader.ResolveCanonicalResourceAncestors(resourceID) { + for _, ancestorID := range identity.(resourceIntentAncestorReader).ResolveCanonicalResourceAncestors(resourceID) { ancestorState, ancestorFound, ancestorErr := reader.GetResourceOperatorState(ancestorID) if ancestorErr != nil { log.Warn().Err(ancestorErr).Str("resourceID", resourceID).Str("ancestorID", ancestorID).Msg("Failed to read inherited operator maintenance state") @@ -102,6 +116,114 @@ func (m *Monitor) installOperatorIntentResolver(store ResourceStoreInterface) { } } +// operatorIntentIdentity is the store alert intent resolves resource +// references and ancestors through: the published registry with saved hosts +// that have not reported since a restart overlaid, the read state Patrol +// resolves through and the resources API seeds from. A saved host exists only +// there, and an operator link holds it under the link's primary, so the +// published registry alone left the saved host's alerts reading their literal +// reference and missed intent set on that host or on the primary. +// +// The alert manager resolves under its own lock, once per metric check, so +// the overlay is kept for the published generation it was built from. +// Publication builds it before evaluating alerts (refresh, outside that +// lock); a lookup that races ahead of publication builds it itself. A +// saved-host change between generations shows on the next one. Mock mode +// carries no saved hosts: the published registry answers and any real-mode +// overlay is dropped. +type operatorIntentIdentity struct { + monitor *Monitor + store ResourceStoreInterface + published unifiedresources.ReadState + generations interface { + Generation() unifiedresources.RegistryGeneration + } + + mu sync.Mutex + built bool + builtFor unifiedresources.RegistryGeneration + readState any +} + +// newOperatorIntentIdentity returns nil for a store without generations or +// the reference and ancestor readers, which alert intent then uses as is. +func (m *Monitor) newOperatorIntentIdentity(store ResourceStoreInterface) *operatorIntentIdentity { + published, isReadState := store.(unifiedresources.ReadState) + generations, versioned := store.(interface { + Generation() unifiedresources.RegistryGeneration + }) + _, resolves := store.(resourceIntentIdentityReader) + _, walksAncestors := store.(resourceIntentAncestorReader) + if !isReadState || !versioned || !resolves || !walksAncestors { + return nil + } + return &operatorIntentIdentity{monitor: m, store: store, published: published, generations: generations} +} + +// current returns the read state for the current published generation. +func (i *operatorIntentIdentity) current() any { + if mock.IsMockEnabled() { + i.mu.Lock() + i.built, i.readState = false, nil + i.mu.Unlock() + return i.store + } + generation := i.generations.Generation() + i.mu.Lock() + if i.built && i.builtFor == generation { + readState := i.readState + i.mu.Unlock() + return readState + } + i.mu.Unlock() + return i.build(generation) +} + +// refresh builds the read state for the current published generation unless +// it is already built. +func (i *operatorIntentIdentity) refresh() { + if i == nil || mock.IsMockEnabled() { + return + } + generation := i.generations.Generation() + i.mu.Lock() + fresh := i.built && i.builtFor == generation + i.mu.Unlock() + if !fresh { + i.build(generation) + } +} + +// build overlays saved hosts on the published registry outside i.mu, so a +// concurrent lookup waits on neither the overlay nor the store reads it makes. +// It keeps the result only while the mock-mode epoch it started in and the +// generation it read are still current. The check and the store run inside +// the fence, so a mode switch, which advances the fence after flipping the +// mode, either waits for them or makes them refuse: a build that read no +// saved hosts because mock mode was on never stands in for real mode. +func (i *operatorIntentIdentity) build(generation unifiedresources.RegistryGeneration) any { + scope := i.monitor.mockModeFence.begin() + if mock.IsMockEnabled() { + return i.store + } + readState := any(i.store) + overlay := i.monitor.readStateWithStandaloneHostContinuity(i.published) + if _, ok := overlay.(resourceIntentIdentityReader); ok { + if _, ok := overlay.(resourceIntentAncestorReader); ok { + readState = overlay + } + } + scope.run(func() { + if mock.IsMockEnabled() || i.generations.Generation() != generation { + return + } + i.mu.Lock() + i.built, i.builtFor, i.readState = true, generation, readState + i.mu.Unlock() + }) + return readState +} + func (m *Monitor) resolveBackupIntentContext(_ string, instance, node string, vmid int, now time.Time) (alerts.BackupIntentContext, bool) { if m == nil || m.state == nil || vmid <= 0 { return alerts.BackupIntentContext{}, false diff --git a/internal/monitoring/monitor_alert_intent_test.go b/internal/monitoring/monitor_alert_intent_test.go index 1b3c0485e2..2ec3c86355 100644 --- a/internal/monitoring/monitor_alert_intent_test.go +++ b/internal/monitoring/monitor_alert_intent_test.go @@ -5,6 +5,8 @@ import ( "time" "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" @@ -302,3 +304,150 @@ func TestSyncAlertsToStateCarriesLiveMetricStatusOfHeldAlert(t *testing.T) { t.Fatalf("live status = %+v, want latched at 78 clearing at 75", status) } } + +// After a restart, a host that has not reported again exists only through +// saved-host continuity, and the host-offline alert continuity raises names it +// as agent:. Intent the operator set on the saved host's row must +// reach that alert, as it does for Patrol findings and resources API reads, +// although the published registry alone does not know the host. +func TestOperatorIntentReachesSavedHostAlertsAfterRestart(t *testing.T) { + now := time.Now().UTC() + saved := config.HostContinuityEntry{ + HostID: "host-retired", MachineID: "machine-retired", Hostname: "retired", + Platform: "linux", LastSeen: now.Add(-10 * time.Minute), + } + continuity := config.NewHostContinuityStore(t.TempDir(), nil) + if err := continuity.Upsert(saved); err != nil { + t.Fatal(err) + } + store := unifiedresources.NewMemoryStore() + savedID := unifiedresources.MachineIdentityCanonicalID(unifiedresources.ResourceTypeAgent, saved.MachineID) + if err := store.SetResourceOperatorState(unifiedresources.ResourceOperatorState{ + CanonicalID: savedID, IntentionallyOffline: true, + }); err != nil { + t.Fatal(err) + } + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store)) + adapter.PopulateFromSnapshot(models.StateSnapshot{LastUpdate: now}) + if _, ok := adapter.ResolveCanonicalResourceID("agent:" + saved.HostID); ok { + t.Fatal("published registry already knows the saved host; the case needs it absent") + } + + manager := alerts.NewManagerWithDataDir(t.TempDir()) + t.Cleanup(manager.Stop) + m := &Monitor{state: models.NewState(), alertManager: manager, hostContinuityStore: continuity} + m.SetResourceStore(adapter) + + preview, err := manager.PreviewIntentPolicy(alerts.AlertIntentPolicyPreviewRequest{ + ResourceID: "agent:" + saved.HostID, ResourceType: "agent", + Signal: string(alerts.AlertIntentSignalOffline), ConditionActive: true, + }) + if err != nil { + t.Fatal(err) + } + if preview.Reason != "operator_expected_offline" { + t.Fatalf("saved host offline alert intent = %+v, want the intentionally-offline state set on %s", preview, savedID) + } + + // The saved host reports again: the published registry knows it, and its + // alerts keep reading the same row. + adapter.PopulateFromSnapshot(models.StateSnapshot{ + Hosts: []models.Host{{ + ID: saved.HostID, MachineID: saved.MachineID, Hostname: saved.Hostname, + Platform: "linux", Status: "online", LastSeen: now, IntervalSeconds: 30, + }}, + LastUpdate: now.Add(time.Second), + }) + m.state.Hosts = []models.Host{{ID: saved.HostID, MachineID: saved.MachineID, Hostname: saved.Hostname, Status: "online", LastSeen: now}} + preview, err = manager.PreviewIntentPolicy(alerts.AlertIntentPolicyPreviewRequest{ + ResourceID: "agent:" + saved.HostID, ResourceType: "agent", + Signal: string(alerts.AlertIntentSignalOffline), ConditionActive: true, + }) + if err != nil { + t.Fatal(err) + } + if preview.Reason != "operator_expected_offline" { + t.Fatalf("reporting host offline alert intent = %+v, want the same intentionally-offline state", preview) + } +} + +// Alert intent keeps one saved-host overlay per published generation, so +// lookups made under the alert manager's lock reuse what publication built. +// A new publication replaces it even when it shares the last one's timestamp +// or is stamped earlier, and mock mode answers from the published registry +// and drops the real-mode overlay. +func TestOperatorIntentIdentityKeepsOneOverlayPerGeneration(t *testing.T) { + now := time.Now().UTC() + saved := config.HostContinuityEntry{ + HostID: "host-retired", MachineID: "machine-retired", Hostname: "retired", + Platform: "linux", LastSeen: now.Add(-10 * time.Minute), + } + continuity := config.NewHostContinuityStore(t.TempDir(), nil) + if err := continuity.Upsert(saved); err != nil { + t.Fatal(err) + } + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(unifiedresources.NewMemoryStore())) + adapter.PopulateFromSnapshot(models.StateSnapshot{LastUpdate: now.Add(time.Hour)}) + m := &Monitor{state: models.NewState(), hostContinuityStore: continuity} + identity := m.newOperatorIntentIdentity(adapter) + if identity == nil { + t.Fatal("monitor adapter offers no operator intent identity") + } + resolvesSavedHost := func(readState any) bool { + _, ok := readState.(resourceIntentIdentityReader).ResolveCanonicalResourceID("agent:" + saved.HostID) + return ok + } + + identity.refresh() + first := identity.current() + if first == any(adapter) || !resolvesSavedHost(first) { + t.Fatal("refresh did not build the saved-host overlay") + } + if again := identity.current(); again != first { + t.Fatal("lookup rebuilt the overlay within one generation") + } + + // A rebuild of a snapshot stamped in the future keeps its time, so two + // publications can share a timestamp. + adapter.PopulateFromSnapshot(models.StateSnapshot{LastUpdate: now.Add(time.Hour)}) + identity.refresh() + sameStamp := identity.current() + if sameStamp == first || !resolvesSavedHost(sameStamp) { + t.Fatal("a publication sharing the last one's timestamp kept its overlay") + } + + adapter.PopulateFromSnapshot(models.StateSnapshot{LastUpdate: now}) + identity.refresh() + second := identity.current() + if second == sameStamp || !resolvesSavedHost(second) { + t.Fatal("an earlier-stamped generation kept the previous overlay") + } + if again := identity.current(); again != second { + t.Fatal("lookups rebuild the overlay after an earlier-stamped generation") + } + + previous := mock.IsMockEnabled() + if err := mock.SetEnabled(true); err != nil { + t.Fatalf("enable mock mode: %v", err) + } + t.Cleanup(func() { _ = mock.SetEnabled(previous) }) + if got := identity.current(); got != any(adapter) { + t.Fatal("mock mode resolved through the real-mode saved-host overlay") + } + identity.refresh() + if err := mock.SetEnabled(false); err != nil { + t.Fatalf("disable mock mode: %v", err) + } + // Alert evaluation resolves from inside an admitted mock-mode fence call, + // so the rebuild's own fenced store nests in it. + var third any + if !m.mockModeFence.begin().run(func() { third = identity.current() }) { + t.Fatal("fence refused the lookup") + } + if third == any(adapter) || !resolvesSavedHost(third) { + t.Fatal("leaving mock mode kept answering from the published registry alone") + } + if again := identity.current(); again != third { + t.Fatal("the rebuild made inside a fenced call was not kept") + } +} diff --git a/internal/monitoring/monitor_alert_sync.go b/internal/monitoring/monitor_alert_sync.go index b85b085e15..a53f30a6e6 100644 --- a/internal/monitoring/monitor_alert_sync.go +++ b/internal/monitoring/monitor_alert_sync.go @@ -103,6 +103,11 @@ func (m *Monitor) syncUnifiedResourceAlertsToState(resources []unifiedresources. return } + // Alert intent resolves through the published registry with saved hosts + // overlaid, under the alert manager's lock. Build that overlay for the new + // generation here, outside the lock, so its lookups need not. + m.operatorIntentIdentity.Load().refresh() + // Resource publication can make persisted canonical policy resolvable for // restored native-ID alerts that preceded the first observation at startup. scope.run(func() { m.alertManager.ReconcileOperatorIntentState() }) diff --git a/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go b/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go index e9d81cf562..98837683c6 100644 --- a/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go +++ b/internal/monitoring/monitor_host_agent_removal_lifecycle_test.go @@ -619,6 +619,35 @@ func TestHostAgentRemovalLifecycleKeepsOfflineRowRemovableAfterRestart(t *testin } } +// After a restart a host that has not reported again exists only as its saved +// continuity row, and its offline alert names it agent:. The intent +// the operator set on that row must still hold the alert back: alert intent +// resolves through the read state that overlays saved hosts, not through the +// published registry alone, which no longer lists the host. +func TestHostAgentRemovalLifecycleHonorsSavedHostIntentAfterRestart(t *testing.T) { + dataPath := t.TempDir() + now := time.Now().UTC() + token := &config.APITokenRecord{ID: "intent-restart-token", CreatedAt: now.Add(-time.Hour)} + report := hostRemovalLifecycleReport("intent-restart-host", "intent-restart-machine", "intent-restart-agent", "intent-restart.local", "linux", now) + if _, err := newHostRemovalLifecycleMonitor(t, dataPath).ApplyHostReport(report, token); err != nil { + t.Fatalf("initial ApplyHostReport: %v", err) + } + + restarted := newHostRemovalLifecycleMonitor(t, dataPath) + store := unifiedresources.NewMemoryStore() + savedID := unifiedresources.MachineIdentityCanonicalID(unifiedresources.ResourceTypeAgent, "intent-restart-machine") + if err := store.SetResourceOperatorState(unifiedresources.ResourceOperatorState{CanonicalID: savedID, IntentionallyOffline: true}); err != nil { + t.Fatalf("set saved host intent: %v", err) + } + restarted.SetResourceStore(unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store))) + for i := 0; i < 3; i++ { + restarted.evaluateHostAgents(now.Add(10*time.Minute + time.Duration(i)*time.Second)) + } + if alerts := restarted.alertManager.GetActiveAlerts(); len(alerts) != 0 { + t.Fatalf("intentionally offline saved host raised %+v after restart, want no alert", alerts) + } +} + func TestHostAgentRemovalLifecycleDoesNotPoisonDuplicateActiveIdentity(t *testing.T) { monitor := newHostRemovalLifecycleMonitor(t, t.TempDir()) now := time.Now().UTC() diff --git a/internal/unifiedresources/clone_test.go b/internal/unifiedresources/clone_test.go index e52c6ff3e6..5e3a88e1b2 100644 --- a/internal/unifiedresources/clone_test.go +++ b/internal/unifiedresources/clone_test.go @@ -1050,3 +1050,18 @@ func TestCloneMaterializedResourceDetachesCanonicalPolicy(t *testing.T) { t.Fatal("nil materialized clone changed absence") } } + +// The resources API seeds its registry from the monitor's read state, so the +// marker on a row saved-host continuity introduced must ride the clone, or the +// API's link pass would fold the saved agent into its linked guest. +func TestCloneResourceKeepsContinuityMarker(t *testing.T) { + saved := Resource{ID: "agent-saved", Type: ResourceTypeAgent, continuityOnly: true} + if clone := cloneResource(&saved); !clone.continuityOnly { + t.Fatal("clone dropped the continuity marker") + } + rr := NewRegistry(nil) + rr.IngestResources([]Resource{saved}) + if seeded, ok := rr.Get(saved.ID); !ok || !seeded.continuityOnly { + t.Fatalf("seeded row = %+v (listed=%v), want the continuity marker kept", seeded, ok) + } +} diff --git a/internal/unifiedresources/monitor_adapter.go b/internal/unifiedresources/monitor_adapter.go index 910a61f5a3..cced6ca6e2 100644 --- a/internal/unifiedresources/monitor_adapter.go +++ b/internal/unifiedresources/monitor_adapter.go @@ -372,6 +372,30 @@ func (a *MonitorAdapter) LastRebuiltAt() time.Time { return a.lastRebuiltAt } +// RegistryGeneration identifies one generation a MonitorAdapter published: +// the registry it serves and when that registry last changed. LastRebuiltAt +// alone does not, since two rebuilds of a snapshot stamped in the future +// share its time; a rebuild swaps the registry and an incremental update +// restamps it. +type RegistryGeneration struct { + registry *ResourceRegistry + rebuiltAt int64 +} + +// Generation returns the generation the adapter currently publishes. +func (a *MonitorAdapter) Generation() RegistryGeneration { + if a == nil { + return RegistryGeneration{} + } + a.mu.RLock() + defer a.mu.RUnlock() + generation := RegistryGeneration{registry: a.registry} + if !a.lastRebuiltAt.IsZero() { + generation.rebuiltAt = a.lastRebuiltAt.UnixNano() + } + return generation +} + // TryReplaceRegistryForRead rebuilds the registry on behalf of a read path // (/api/state, websocket hydrate). Unlike the ingest-driven populate methods // it never queues behind an in-flight rebuild and skips entirely while the diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index ca34db477c..02928ca3a0 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -111,6 +111,12 @@ type ResourceRegistry struct { // writes these instead of the merged projection, so a link never lends // one machine's identity keys to another resource's durable pin. linkOwnPins map[string]*ResourceIdentityPin + // linkHolds maps a resource that exists only through saved-host + // continuity, and that an operator link would otherwise fold away, to + // the link's primary (holdLinkedResourceLocked). Both rows stay listed + // with their own telemetry; references to the held row resolve to the + // primary. Every link pass recomputes it. + linkHolds map[string]string // supersededIndex maps record-declared retired canonical IDs to the live // resource that superseded them, so references persisted under a retired // ID (availability links, API reads) keep resolving. An empty value marks @@ -695,11 +701,12 @@ func (rr *ResourceRegistry) ingestRecords(source DataSource, records []IngestRec // Snapshot and resource ingest apply the links, so record ingest must too, // or the monitor's rebuilt registry keeps both rows while REST, which // seeds from that registry, shows them merged. Continuity records only - // fill absent machines: a saved enrollment joined to its linked guest - // would lend the guest its offline verdict and old agent payload. - if !onlyMissing { - rr.applyManualLinks(thresholds) - } + // fill absent machines and never merge into a link (a saved enrollment + // joined to its linked guest would lend the guest its offline verdict + // and old agent payload), but the link still names their identity: the + // pass holds them beside the primary and resolves references to them + // there. + rr.applyManualLinks(thresholds) rr.refreshStorageConsumersLocked() rr.refreshPBSRollupsLocked() rr.refreshStoragePostureLocked() @@ -1584,8 +1591,18 @@ func (rr *ResourceRegistry) ResolveReferenceID(ref string) (string, bool) { } // The second result requests a one-time alias-index build after the read lock -// is released. Exact and source references never pay that cost. +// is released. Exact and source references never pay that cost. A reference +// to a saved link member held beside its primary resolves to the primary, +// as one to a folded member does: the operator's link names one identity. func (rr *ResourceRegistry) resolveReferenceIDLocked(ref string) (string, bool) { + resolvedID, needsAliasIndex := rr.resolveReferenceRowIDLocked(ref) + if resolvedID == "" { + return "", needsAliasIndex + } + return rr.linkHoldTargetLocked(resolvedID), false +} + +func (rr *ResourceRegistry) resolveReferenceRowIDLocked(ref string) (string, bool) { if ref == "" { return "", false } @@ -1645,7 +1662,7 @@ func (rr *ResourceRegistry) uniqueCanonicalIdentityResourceIDLocked(ref string) matches := map[string]struct{}{} for resourceID, resource := range rr.resources { if resourceMatchesCanonicalIdentityReference(resource, ref) { - matches[resourceID] = struct{}{} + matches[rr.linkHoldTargetLocked(resourceID)] = struct{}{} } } return uniqueResourceIDMatch(matches) @@ -3069,6 +3086,7 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res if onlyMissing && rr.resources[resource.ID] != nil { return "" } + resource.continuityOnly = onlyMissing stampPhysicalDiskTemperatureReading(&resource, source, sourceID) rr.resources[resource.ID] = &resource rr.indexPhysicalDiskHardwareLocked(&resource) @@ -3095,6 +3113,7 @@ func (rr *ResourceRegistry) ingestRecord(source DataSource, sourceID string, res rr.matcher.Add(existing.ID, existing.Identity) return existing.ID } + resource.continuityOnly = onlyMissing stampPhysicalDiskTemperatureReading(&resource, source, sourceID) rr.resources[resource.ID] = &resource rr.indexPhysicalDiskHardwareLocked(&resource) @@ -3671,7 +3690,8 @@ func (rr *ResourceRegistry) resolveAvailabilityLinkedResource(ref string, incomi exactID := CanonicalResourceID(ref) if existing := rr.resources[exactID]; existing != nil { if !isAvailabilityOwnedResource(*existing) { - return exactID + // A saved link member held beside its primary answers there. + return rr.linkHoldTargetLocked(exactID) } return "" } @@ -3684,6 +3704,9 @@ func (rr *ResourceRegistry) resolveAvailabilityLinkedResource(ref string, incomi // fail-closed. eligible := func(candidateID string) string { candidateID = CanonicalResourceID(candidateID) + if candidateID != "" { + candidateID = rr.linkHoldTargetLocked(candidateID) + } existing := rr.resources[candidateID] if existing != nil && !isAvailabilityOwnedResource(*existing) { return candidateID @@ -3909,6 +3932,8 @@ func (rr *ResourceRegistry) mergeInto(existing *Resource, incoming Resource, sou if existing == nil { return } + // A source reported this row, so it is an observation again. + existing.continuityOnly = false rr.setSourceParent(existing, source, incoming.ParentID) @@ -4864,6 +4889,10 @@ func mergeVMwareData(existing *VMwareData, incoming *VMwareData) *VMwareData { } func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Duration) { + if rr.linkHolds != nil { + rr.linkHolds = nil + rr.canonicalIdentityIndex = nil + } if len(rr.links) == 0 { return } @@ -4913,6 +4942,10 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura primary, other = other, primary primaryID, otherID = otherID, primaryID } + if primary.continuityOnly || other.continuityOnly { + rr.holdLinkedResourceLocked(primaryID, otherID, other) + continue + } rr.recordLinkOwnPin(primaryID, primary) rr.recordLinkOwnPin(otherID, other) @@ -4945,6 +4978,69 @@ func (rr *ResourceRegistry) recordLinkOwnPin(id string, resource *Resource) { rr.linkOwnPins[id] = nil } +// holdLinkedResourceLocked keeps a link whose member exists only through +// saved-host continuity out of the fold. A saved enrollment is not an +// observation: folded into a live resource it would lend that resource its +// offline verdict and old payload, and a live resource folded into it would +// hide live telemetry behind it. The link still names one identity, so a +// saved member the fold would have taken in answers to the primary, +// as a folded one does. A live member keeps its own references even when the +// saved member is the primary: a live row never resolves to a saved one. The +// first link to claim a member holds it, as the first fold takes it in. +func (rr *ResourceRegistry) holdLinkedResourceLocked(primaryID, otherID string, other *Resource) { + if !other.continuityOnly { + return + } + if _, held := rr.linkHolds[otherID]; held { + return + } + if rr.linkHolds == nil { + rr.linkHolds = make(map[string]string) + } + rr.linkHolds[otherID] = primaryID + rr.canonicalIdentityIndex = nil +} + +// linkHoldPrimaryLocked returns the row a held link member's references +// resolve to: its primary, followed through a later fold of the primary into +// another resource and through further holds (a saved primary held under +// another link). Empty when the row is not held or the chain does not end on +// a listed row. +func (rr *ResourceRegistry) linkHoldPrimaryLocked(resourceID string) string { + primaryID, held := rr.linkHolds[resourceID] + if !held { + return "" + } + seen := map[string]struct{}{resourceID: {}} + for { + if rr.resources[primaryID] == nil { + holderID, _ := rr.linkFoldHolderLocked(primaryID) + if holderID == "" { + return "" + } + primaryID = holderID + } + next, held := rr.linkHolds[primaryID] + if !held { + return primaryID + } + if _, cycle := seen[primaryID]; cycle { + return "" + } + seen[primaryID] = struct{}{} + primaryID = next + } +} + +// linkHoldTargetLocked maps a resolved row to the row its references answer +// to: the hold primary for a held link member, the row itself otherwise. +func (rr *ResourceRegistry) linkHoldTargetLocked(resourceID string) string { + if primaryID := rr.linkHoldPrimaryLocked(resourceID); primaryID != "" { + return primaryID + } + return resourceID +} + func (rr *ResourceRegistry) mergeResourceData(primary *Resource, other *Resource, thresholds map[DataSource]time.Duration) { if other == nil || primary == nil { return @@ -5448,6 +5544,9 @@ func (rr *ResourceRegistry) buildCanonicalIdentityIndexLocked() { if resource.Canonical == nil { continue } + // A saved link member held beside its primary shares the primary's + // identity, so an alias both carry (a hostname) stays unambiguous. + resourceID = rr.linkHoldTargetLocked(resourceID) indexCandidate(resource.Canonical.PrimaryID, resourceID) indexCandidate(resource.Canonical.PlatformID, resourceID) for _, alias := range resource.Canonical.Aliases { diff --git a/internal/unifiedresources/resolve_test.go b/internal/unifiedresources/resolve_test.go index d45bbc7e3a..ee236ddcc2 100644 --- a/internal/unifiedresources/resolve_test.go +++ b/internal/unifiedresources/resolve_test.go @@ -455,3 +455,270 @@ func TestResolveResourceContext_NotFound(t *testing.T) { t.Fatalf("expected no resolved resource for missing lookup, got %#v", resolved.Resource) } } + +// linkedGuestEstate is an agent inside a vSphere VM, joined by an operator +// link, as Pulse sees it after a restart before the agent reports again: the +// published registry has only the VM, and saved-host continuity fills in the +// agent. +type linkedGuestEstate struct { + saved models.Host + vmRecords []IngestRecord + agentID string + vmID string +} + +func newLinkedGuestEstate(t *testing.T) linkedGuestEstate { + t.Helper() + now := time.Now().UTC() + estate := linkedGuestEstate{ + saved: models.Host{ + ID: "host-app-guest", MachineID: "machine-app-guest", Hostname: "app-guest", + Platform: "linux", Status: "offline", LastSeen: now.Add(-10 * time.Minute), + Memory: models.Memory{Total: 8 << 30, Used: 7 << 30, Free: 1 << 30, Usage: 88}, + }, + vmRecords: []IngestRecord{{ + SourceID: "vc-1:vm:vm-42", + Resource: Resource{ + Type: ResourceTypeVM, + Technology: "vmware", + Name: "app-guest", + Status: StatusOnline, + LastSeen: now, + Metrics: &ResourceMetrics{Memory: &MetricValue{Percent: 40, Source: SourceVMware}}, + VMware: &VMwareData{ConnectionID: "vc-1", ManagedObjectID: "vm-42", EntityType: "vm"}, + }, + Identity: ResourceIdentity{Hostnames: []string{"app-guest"}}, + }}, + } + unlinked := NewMonitorAdapter(NewRegistry(nil)) + unlinked.PopulateSnapshotAndSupplemental( + models.StateSnapshot{Hosts: []models.Host{estate.saved}, LastUpdate: now}, + map[DataSource][]IngestRecord{SourceVMware: estate.vmRecords}, + ) + if len(unlinked.VMs()) != 1 || len(unlinked.Hosts()) != 1 { + t.Fatalf("unlinked estate = %d VMs, %d agents, want one of each", len(unlinked.VMs()), len(unlinked.Hosts())) + } + estate.vmID, estate.agentID = unlinked.VMs()[0].ID(), unlinked.Hosts()[0].ID() + return estate +} + +// readStateAfterRestart publishes the VM alone and overlays the saved agent, +// as Monitor.GetUnifiedReadStateOrSnapshot does. +func (e linkedGuestEstate) readStateAfterRestart(store ResourceStore) *MonitorAdapter { + published := NewMonitorAdapter(NewRegistry(store)) + published.PopulateSnapshotAndSupplemental( + models.StateSnapshot{LastUpdate: time.Now().UTC()}, + map[DataSource][]IngestRecord{SourceVMware: e.vmRecords}, + ) + return ReadStateWithHostContinuity(published, []IngestRecord{HostIngestRecord(e.saved)}).(*MonitorAdapter) +} + +// An operator link names one identity, so every reference to the saved +// agent resolves to the VM, as it does once the agent reports and folds in. +// The agent's saved telemetry stays on its own row in the read state and in +// the resources API, which seeds a registry from that listing and re-applies +// the links. +func TestContinuityAgentLinkedIntoLiveGuestAnswersToTheGuest(t *testing.T) { + estate := newLinkedGuestEstate(t) + store := NewMemoryStore() + if err := store.AddLink(ResourceLink{ResourceA: estate.vmID, ResourceB: estate.agentID, PrimaryID: estate.agentID}); err != nil { + t.Fatalf("add link: %v", err) + } + readState := estate.readStateAfterRestart(store) + resources := NewRegistry(store) + resources.IngestResources(readState.GetAll()) + + for _, registry := range []struct { + name string + registry *ResourceRegistry + }{ + {"read state", readState.registry}, + {"resources API", resources}, + } { + t.Run(registry.name, func(t *testing.T) { + rr := registry.registry + agent, ok := rr.Get(estate.agentID) + if !ok || agent.Status != StatusOffline { + t.Fatalf("saved agent row = %+v (listed=%v), want its own offline row", agent, ok) + } + vm, ok := rr.Get(estate.vmID) + if !ok { + t.Fatalf("linked VM %s missing", estate.vmID) + } + if vm.Agent != nil || vm.Status != StatusOnline || vm.Metrics == nil || vm.Metrics.Memory == nil || vm.Metrics.Memory.Source != SourceVMware { + t.Fatalf("VM agent=%v status=%s metrics=%+v, want vSphere's online VM without the saved agent's payload", vm.Agent != nil, vm.Status, vm.Metrics) + } + // The hostname both rows carry stays unambiguous: the saved row shares + // the VM's identity. + for _, ref := range []string{estate.agentID, "agent:" + estate.saved.ID, estate.saved.MachineID, "app-guest"} { + if resolved, ok := rr.ResolveReferenceID(ref); !ok || resolved != estate.vmID { + t.Fatalf("ResolveReferenceID(%q) = %q (%v), want the link primary %s", ref, resolved, ok, estate.vmID) + } + if _, resolved, ok := rr.GetByReference(ref); !ok || resolved != estate.vmID { + t.Fatalf("GetByReference(%q) = %q (%v), want the link primary %s", ref, resolved, ok, estate.vmID) + } + } + if resolved, ok := rr.ResolveReferenceID(estate.vmID); !ok || resolved != estate.vmID { + t.Fatalf("VM reference resolved to %q (%v)", resolved, ok) + } + }) + } + if resolved, ok := readState.ResolveCanonicalResourceID("agent:" + estate.saved.ID); !ok || resolved != estate.vmID { + t.Fatalf("read-state canonical resolution = %q (%v), want %s", resolved, ok, estate.vmID) + } + + // Once the agent reports, the link folds it in as before. + live := estate.saved + live.Status, live.LastSeen = "online", time.Now().UTC() + published := NewMonitorAdapter(NewRegistry(store)) + published.PopulateSnapshotAndSupplemental( + models.StateSnapshot{Hosts: []models.Host{live}, LastUpdate: time.Now().UTC()}, + map[DataSource][]IngestRecord{SourceVMware: estate.vmRecords}, + ) + if len(published.Hosts()) != 0 { + t.Fatal("reporting agent still listed beside its linked VM") + } + if resolved, ok := published.ResolveCanonicalResourceID(estate.agentID); !ok || resolved != estate.vmID { + t.Fatalf("folded agent resolved to %q (%v), want %s", resolved, ok, estate.vmID) + } +} + +// A saved primary never captures a live member: the live row keeps its own +// references and telemetry, and the saved row its own, until the primary +// reports and the link folds them. +func TestContinuityPrimaryLeavesLiveLinkMemberItsOwnIdentity(t *testing.T) { + now := time.Now().UTC() + live := models.Host{ID: "host-live", MachineID: "machine-live", Hostname: "live", Platform: "linux", Status: "online", LastSeen: now, CPUUsage: 23} + saved := models.Host{ID: "host-saved", MachineID: "machine-saved", Hostname: "saved", Platform: "linux", Status: "offline", LastSeen: now.Add(-time.Hour)} + + ids := NewMonitorAdapter(NewRegistry(nil)) + ids.PopulateFromSnapshot(models.StateSnapshot{Hosts: []models.Host{live, saved}, LastUpdate: now}) + if len(ids.Hosts()) != 2 { + t.Fatalf("unlinked estate has %d agents, want 2", len(ids.Hosts())) + } + agentIDs := map[string]string{} + for _, host := range ids.Hosts() { + agentIDs[host.AgentID()] = host.ID() + } + liveID, savedID := agentIDs[live.ID], agentIDs[saved.ID] + + store := NewMemoryStore() + if err := store.AddLink(ResourceLink{ResourceA: savedID, ResourceB: liveID, PrimaryID: savedID}); err != nil { + t.Fatalf("add link: %v", err) + } + published := NewMonitorAdapter(NewRegistry(store)) + published.PopulateFromSnapshot(models.StateSnapshot{Hosts: []models.Host{live}, LastUpdate: now}) + readState := ReadStateWithHostContinuity(published, []IngestRecord{HostIngestRecord(saved)}).(*MonitorAdapter) + resources := NewRegistry(store) + resources.IngestResources(readState.GetAll()) + + for name, rr := range map[string]*ResourceRegistry{"read state": readState.registry, "resources API": resources} { + liveRow, ok := rr.Get(liveID) + if !ok || liveRow.Status != StatusOnline || len(liveRow.Sources) != 1 { + t.Fatalf("%s: live member = %+v (listed=%v), want its own online row", name, liveRow, ok) + } + if _, ok := rr.Get(savedID); !ok { + t.Fatalf("%s: saved primary row missing", name) + } + for ref, want := range map[string]string{liveID: liveID, "agent:" + live.ID: liveID, savedID: savedID, "agent:" + saved.ID: savedID} { + if resolved, ok := rr.ResolveReferenceID(ref); !ok || resolved != want { + t.Fatalf("%s: %q resolved to %q (%v), want %s", name, ref, resolved, ok, want) + } + } + } +} + +// An observation that merges into a saved row makes it a live row again, so +// the link folds it: the resources API's record replay is one such path. +func TestObservationMergedIntoASavedRowLetsItsLinkFold(t *testing.T) { + estate := newLinkedGuestEstate(t) + store := NewMemoryStore() + if err := store.AddLink(ResourceLink{ResourceA: estate.vmID, ResourceB: estate.agentID, PrimaryID: estate.vmID}); err != nil { + t.Fatalf("add link: %v", err) + } + rr := estate.readStateAfterRestart(store).registry + if _, held := rr.Get(estate.agentID); !held { + t.Fatal("saved agent not listed before it reports") + } + live := estate.saved + live.Status, live.LastSeen = "online", time.Now().UTC() + rr.IngestRecords(SourceAgent, []IngestRecord{HostIngestRecord(live)}) + if _, listed := rr.Get(estate.agentID); listed { + t.Fatal("reported agent still listed beside its linked VM") + } + if vm, ok := rr.Get(estate.vmID); !ok || vm.Agent == nil { + t.Fatalf("linked VM = %+v (listed=%v), want it carrying the reported agent", vm, ok) + } +} + +// Holds follow the links one pass applies, whatever their order: a primary a +// later link folds away hands its held members to the row holding it, a saved +// member held under a saved primary answers to that primary's own target, and +// a cycle of saved members resolves each to itself. +func TestLinkHoldsFollowTheirPrimaryThroughThePass(t *testing.T) { + saved := func(id string) Resource { + return Resource{ID: id, Type: ResourceTypeAgent, Name: id, Status: StatusOffline, Sources: []DataSource{SourceAgent}, continuityOnly: true} + } + vm := func(id string) Resource { + return Resource{ID: id, Type: ResourceTypeVM, Name: id, Status: StatusOnline, Sources: []DataSource{SourceVMware}} + } + for _, tc := range []struct { + name string + resources []Resource + links []ResourceLink + want map[string]string + }{ + { + name: "primary folded later", + resources: []Resource{saved("agent-a"), vm("vm-v"), vm("vm-w")}, + links: []ResourceLink{ + {ResourceA: "agent-a", ResourceB: "vm-v", PrimaryID: "vm-v"}, + {ResourceA: "vm-v", ResourceB: "vm-w", PrimaryID: "vm-w"}, + }, + want: map[string]string{"agent-a": "vm-w", "vm-v": "vm-w", "vm-w": "vm-w"}, + }, + { + name: "saved primary held in turn", + resources: []Resource{saved("agent-x"), saved("agent-a"), vm("vm-v")}, + links: []ResourceLink{ + {ResourceA: "agent-x", ResourceB: "agent-a", PrimaryID: "agent-a"}, + {ResourceA: "agent-a", ResourceB: "vm-v", PrimaryID: "vm-v"}, + }, + want: map[string]string{"agent-x": "vm-v", "agent-a": "vm-v", "vm-v": "vm-v"}, + }, + { + name: "cycle", + resources: []Resource{saved("agent-a"), saved("agent-b"), saved("agent-c")}, + links: []ResourceLink{ + {ResourceA: "agent-a", ResourceB: "agent-b", PrimaryID: "agent-a"}, + {ResourceA: "agent-b", ResourceB: "agent-c", PrimaryID: "agent-b"}, + {ResourceA: "agent-c", ResourceB: "agent-a", PrimaryID: "agent-c"}, + }, + want: map[string]string{"agent-a": "agent-a", "agent-b": "agent-b", "agent-c": "agent-c"}, + }, + } { + t.Run(tc.name, func(t *testing.T) { + store := NewMemoryStore() + for _, link := range tc.links { + if err := store.AddLink(link); err != nil { + t.Fatalf("add link: %v", err) + } + } + rr := NewRegistry(store) + rr.IngestResources(tc.resources) + for ref, want := range tc.want { + if resolved, ok := rr.ResolveReferenceID(ref); !ok || resolved != want { + t.Fatalf("%q resolved to %q (%v), want %s", ref, resolved, ok, want) + } + } + for _, resource := range tc.resources { + if !resource.continuityOnly { + continue + } + if row, listed := rr.Get(resource.ID); !listed || row.Status != StatusOffline { + t.Fatalf("saved row %s = %+v (listed=%v), want it kept", resource.ID, row, listed) + } + } + }) + } +} diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index 19cd64861b..6a049d8374 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -61,6 +61,13 @@ type Resource struct { // (recordManualLinkFold); report-merge excludes their pairs. linkFolds []ManualLinkFold + // continuityOnly marks a row saved-host continuity introduced: a machine + // no source has reported since a restart. It rides in-memory clones, so + // a registry seeded from a continuity read state still holds the row out + // of operator-link folds (holdLinkedResourceLocked). An observation that + // later merges into the row clears it. + continuityOnly bool + Tags []string `json:"tags,omitempty"` CustomURL string `json:"customUrl,omitempty"` Capabilities []ResourceCapability `json:"capabilities,omitempty"`