From 7f026a40ce80cbe5d4843633127790d2d3605d14 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:22:34 +0100 Subject: [PATCH 1/3] Reconcile operator links and splits with generation freshness Retain PR2732 cccb43ecdf792caa6036eb422aa8ae3da1fe00dc and PR2741 e2f2a6138adb465a9101f55b460659df709bb38f behaviours against current monitor-owned freshness and holder-bound fold bookkeeping. Preserve mock restart/unlink/relink, seven-shape split controls and canonical fold provenance. Recreate the exact final source tree of rejected candidate d73e56086f4da1e0325c1e7311cdea1625f62a5d over the recorded assignment base. Its imported proposal commits lacked maintainer committer provenance. Original proposal history remains unchanged and retained as evidence; no tests or source-proof receipts are rebound to this new commit. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 23 ++ .../v6/internal/subsystems/monitoring.md | 50 +++ .../internal/subsystems/unified-resources.md | 210 +++++++++- internal/mock/platform_fixtures.go | 24 +- internal/mock/platform_fixtures_test.go | 97 +++++ .../monitoring/canonical_guardrails_test.go | 2 +- internal/monitoring/monitor.go | 73 +++- .../monitoring/monitor_host_agents_test.go | 197 +++++++++ .../unifiedresources/canonical_id_pins.go | 3 +- internal/unifiedresources/manual_links.go | 30 ++ .../presentation_exclusions.go | 22 +- .../unifiedresources/proxmox_node_links.go | 291 +++++++++++++ internal/unifiedresources/registry.go | 67 ++- .../registry_merge_policy_test.go | 385 ++++++++++++++++++ internal/unifiedresources/registry_test.go | 63 +++ .../registry_views_branchcov0722pm_test.go | 2 +- 16 files changed, 1504 insertions(+), 35 deletions(-) create mode 100644 internal/unifiedresources/manual_links.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 6f138aad56..3bcb22c6fa 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -9247,6 +9247,18 @@ carrying a prefixed attribute or namespace declaration named like a SAML attribute is refused. No agent registration, enrolment, install, update or removal path changed. +### Mock-mode read views fold linked fixture agents + +`internal/monitoring/monitor.go` changed only so the mock-mode unified view +applies the operator's manual links (unified-resources contract, "Mock-mode +unified view applies operator links"). A fixture agent the operator linked +into a guest now folds into that guest in mock read views, as a linked agent +already does in live mode, so host snapshots built from them +(`HostsSnapshot`) stop listing it while the guest carries its facet. Agent +registration, enrolment, report admission, removal, continuity and the +fixture agent set itself are unchanged, and the view writes nothing to the +resource store. + ### Demo write guard ignores websocket upgrade headers `internal/api/demo_middleware.go` changed only so a `POST`, `PUT`, `PATCH` or @@ -9309,3 +9321,14 @@ machine setting. Cleanup restores only the captured prior setting, even when owned server teardown fails; an unconfirmed stop fails and retains the handle rather than reporting clean teardown. Real Windows execution and full terminal cleanup evidence still have to be returned by the existing native CI job. + +### Reconciled node-agent relink fold bookkeeping + +A declared node-agent join can consume one split row before its newer +operator relink is applied. The relink records that side's ID and source in +the current holder-bound `ManualLinkFold` index, so identity-pin succession +keeps the operator decision and the resources API seed retains the same fold +for report-merge. No synthetic telemetry, identity key, enrollment or service +link is added. `TestOperatorSplitOverridesProxmoxNodeAgentLink` checks that +the monitor and its re-ingested API seed retain the fold across all split +shapes, then survive pin persistence and a repeated split. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 980d773ff0..620b86fad4 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -3252,6 +3252,38 @@ overlap/covered-write/liveness/resumption dependency. Preserve the monitoring and alert outage precaution until that operational outcome is verified. +### Mock-mode unified view applies operator links + +The mock branch of `currentUnifiedStateView` builds the view the broadcast, +`/api/state`, `GetUnifiedReadStateOrSnapshot` and `UnifiedResourceSnapshot` +serve in mock mode from the fixture graph, which knows no link store. It now +builds the graph with the links the monitor's resource store applied at its +current generation (`resourceStoreManualLinks`, +`mock.UnifiedResourceSnapshotWithLinks`; unified-resources contract +"Mock-mode unified view applies operator links"), so a link reaches the mock +broadcast when the resource store's next rebuild loads it, as in live mode; in +mock mode that is normally the broadcast's own read-path refresh once the +generation is older than `readPathRegistryFreshness`. The cached view is keyed +on the fixture data version and the link list, so a link change rebuilds it +without a fixture tick; an unchanged list keeps the shared build. Freshness, +which the resources API keys its registry cache on, is assigned in +publication order under `mockUnifiedViewMu`: a view built from the same +fixture freshness and links as the one it replaces keeps its freshness, and +any other view gets a later one than every view before it, so a slow build +that publishes late cannot reuse a value. With no links the view still comes +from the shared memoized fixture snapshot, so the demo's allocation profile is +unchanged; with links each monitor builds its own linked snapshot whenever +the fixture data or links change (concurrent misses can each build one). The fence view behind `currentModeReadState` still omits links (see +the mock-mode fence above): it stays a plain projection of current state for +the window before a rebuild in the new epoch, after which the registry applies +the links again. `TestMockUnifiedViewAppliesOperatorManualLinks` in +`internal/monitoring/monitor_host_agents_test.go` pins the broadcast, the +freshness advance and the cache, `TestFixtureGraphAppliesManualLinksAtEachIngestStage` +and `TestUnifiedResourceSnapshotWithLinksLeavesTheSharedSnapshotUnlinked` in +`internal/mock/platform_fixtures_test.go` pin the linked fixture build, and +`TestMockUnifiedStateViewUsesCanonicalMockFixtureGraph` in +`internal/monitoring/canonical_guardrails_test.go` pins the call. + ### Saved quiet-hours policy before queue activation `Monitor.New` binds `alerts.Manager.QuietHoursNotificationPolicy` to the @@ -6457,3 +6489,21 @@ and REST row counts with and without the split, and `TestBroadcastPresentationCoalescePrefersTheListingReadState` pins the listing read state over an older store generation and the store's exclusions for a mock-style view. + +### Reconciled node-agent relink fold bookkeeping + +A declared node-agent join can consume one split row before its newer +operator relink is applied. The relink records that side's ID and source in +the current holder-bound `ManualLinkFold` index, so identity-pin succession +keeps the operator decision and the resources API seed retains the same fold +for report-merge. No synthetic telemetry, identity key, enrollment or service +link is added. `TestOperatorSplitOverridesProxmoxNodeAgentLink` checks that +the monitor and its re-ingested API seed retain the fold across all split +shapes, then survive pin persistence and a repeated split. + +Mock-link lifecycle validation drives the owning ingest/rebuild boundary +after unlink and relink: read hydration intentionally reuses a registry +generation younger than two seconds. The decision takes effect in the next +loaded generation, not retroactively in the listing already served. The +fixture seed keeps default thresholds while the configured live adapter +retains its four-minute Proxmox threshold. diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 1f54825eb2..63dfd161c4 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -6683,9 +6683,9 @@ snapshot and resource ingest applied links, the rebuilt registry behind the websocket broadcast and `/api/state` kept both rows. The resources API, which seeds through resource ingest, showed them merged, and so did the published inventory whenever an unrelated absent machine made continuity hydration -clone the registry. Mock mode builds its unified view from fixtures without -the link store, so links do not reach the mock broadcast; that boundary is -separate. +clone the registry. Mock mode builds its unified view from fixtures and +applies the same links there ("Mock-mode unified view applies operator +links" below). - The monitor adapter passes its configured stale thresholds to record ingest, so a link joined there judges each side's metrics the way a @@ -6889,6 +6889,67 @@ resource seeding. (`TestResourcesAPIAndPatrolAgreeOnContinuityAgentLinkedIntoGuest`) pins the resources API's write and Patrol's read on the same store. +### Mock-mode unified view applies operator links + +In mock mode the websocket broadcast, `/api/state`, and the monitor's +`GetUnifiedReadStateOrSnapshot` and `UnifiedResourceSnapshot` readers serve a +view of the fixture graph, not the monitor's resource store, and the fixture +graph was built without the link store. The resources API seeds its registry +from that same list through a registry backed by the link store, so a manual +link showed merged there while the broadcast kept both rows. The fixture graph +is now built with the links the monitor adapter's current registry generation +loaded from its store (`MonitorAdapter.ManualLinks`), in a registry that has +no store at all (`NewRegistryWithManualLinks`, +`mock.UnifiedResourceSnapshotWithLinks`). Nothing that build ingests can write +identity pins, canonical-ID successions or change records, and the view's read +state forwards no history to the operator's store, so fixture data stays out +of durable state through it. + +- The links apply at each ingest stage, as in the live rebuild: the snapshot, + then the TrueNAS and vSphere records, then availability. An availability + check that names the folded resource by source reference resolves through + the fold and projects onto the primary. Folding the finished list instead + dropped the check with the folded row. A check that names the folded + resource by canonical ID still loses its target, as live (fold history + above). +- A link reaches the view when the adapter's next rebuild loads it, the same + point at which the live broadcast picks it up. The view's cache is keyed on + the fixture data version and on the links' resource pairs and primaries, so + a link change rebuilds it without waiting for a fixture tick. Freshness is + assigned in publication order: a view built from the same fixture + freshness and links as the one it replaces keeps its freshness, and any + other view gets a later one than every view published before it, even when + a slower build publishes after a newer one. The resources API reuses its + registry while that freshness is unchanged, so it follows the published + view and does not keep a list other links folded, for example after a + link's primary is reversed. Fixture freshness stands in for the fixture + data, as it already did for that cache. +- Without links the view is built from the shared memoized fixture snapshot, + as before. With links each monitor builds its own linked snapshot whenever + the fixture data or links change (concurrent cache misses can each build + one). +- Exclusions are not carried by the fixture registry, as before. The + broadcast's presentation coalescing consults the monitor adapter's + exclusions, while the resources API's `ListForPresentation` consults its + own store-backed registry's exclusions. Neither can undo an automatic + identity merge already made inside the store-free fixture graph. +- An agent linked into a guest leaves the agent listings (`Hosts()` lists + agent-type rows only) and the guest carries its facet, as in live mode. + +`TestMockUnifiedViewAppliesOperatorManualLinks` in +`internal/monitoring/monitor_host_agents_test.go` pins the broadcast payload, +the rebuild and freshness advance on a link change with the fixture data +version held, the cached build afterwards, the same rows as a store-backed +registry seeded through resource ingest the way the resources API seeds, and +no change rows reaching the store through the view. +`TestFixtureGraphAppliesManualLinksAtEachIngestStage` and +`TestUnifiedResourceSnapshotWithLinksLeavesTheSharedSnapshotUnlinked` in +`internal/mock/platform_fixtures_test.go` pin the staged availability +projection and the untouched shared snapshot, and +`TestRegistryWithManualLinksFoldsLikeTheStoreBackedRegistry` in +`internal/unifiedresources/registry_test.go` pins the store-less registry +against the store-backed one and the per-generation `ManualLinks` snapshot. + ### Provider link network corroboration An existing reciprocal host/provider link remains authoritative, including a @@ -7114,18 +7175,138 @@ folded into the agent by ID, a path no exclusion check sees, so the pair stayed merged with the link row gone. A pin whose canonical ID the operator excluded from the ID the record takes without the pin (its `chooseNewID` or source-specific ID) now completes nothing. This covers pins already written -that way, not only new ones. Merges through an agent's own declared node link -(`resolveLinkedResource`) still ignore exclusions. +that way, not only new ones. A Proxmox node's link to an agent honours the +split as well; the next paragraphs cover it. + +An operator's split overrides a Proxmox node's link to a pulse-agent. That +link is normally Pulse's own inference rather than a resource-store +decision: monitoring links a node to the one agent whose endpoint address or +hostname matches it (the agents API can also set it by hand), and the +registry joined the pair in `resolveLinkedResource` (a link both sides +declare, or a node-side link the agent's hostname corroborates) before +`findMatch` ran. The linked node had already taken the agent's identity in +`resourceFromProxmoxNode`, and with an agent reporting a machine key it +minted the agent's machine-derived canonical ID before the agent was +ingested. Neither path read exclusions, so the drawer's Split merged +resource (report-merge) on a node and its agent answered "Resource split +applied" and left the pair merged on every rebuild, in the monitor and in +the resources API. Cloned machines and reused short hostnames are where that +link is wrong, and report-merge exists for wrong merges, so the split wins. + +`nodeAgentSplit` (`proxmox_node_links.go`) decides the pair: the operator +split the node off when an exclusion names one of the node's own IDs (its +source-specific ID, or the ID its identity derives without a machine key, +which a Proxmox node never reports itself) against one of the agent's (its +source-specific ID, the ID its identity derives, the ID it holds, or the ID +the joined pair derives). Unlink and report-merge name the merged ID and the +`SourceTargets` candidate IDs, and the split node can hold an ID it did not +hold while joined, so the decision is read across those sets, not one exact +pair. Where the pair merged under the agent's machine-derived ID, an +exclusion naming only the agent's candidate against that ID does not count: +report-merge records the same pair when it splits another source, such as a +Docker host, off the agent. Where it merged under the node's own ID (an agent +reporting no machine key), that exclusion splits the agent off the node. +`splitProxmoxNodeLink` keeps a split node from taking the agent's identity in +`ingestSnapshot` and remembers the agent for identity-pin completion, which +refuses the agent's machine key from a pin written while they were joined +when the node is split from the agent its own link or an agent row holding +the pin's ID names, or from the remembered agent where the pin can be that +agent's (no machine key or DMI UUID both report differs), so also while the +agent is missing from the snapshot or stops reporting its machine ID; +`resolveLinkedResource` refuses the link from either side. The registry adds +no `LinkedAgentID` to the node: guests, agent deployment and service +discovery act on a node's linked agent. The node then holds the ID its own +identity derives, with only its hostname and endpoint address, which +identity matching scores at 0.80 at most, under the 0.85 auto-merge floor. +The agent holds the ID its own identity derives once pins complete it (its +machine-derived ID when it has a machine key). + +The latest decision across the two sets wins, so a link between them that +postdates the newest exclusion joins the pair again even where it names +another pair than the exclusions do, as a relink of the two rows a +report-merge left does (a registry now keeps each exclusion's `CreatedAt`); +a tie keeps the pair apart. Once a relink rejoins them, the declared link +folds one of the rows it names into the other's ID, and `applyManualLinks` +records that ID with the IDs manual links fold (it recognises the node's own +IDs from the joined row's Proxmox facet, and the agent's source-specific +ID), so pin succession keeps the link instead of re-keying it onto the +joined ID and handing the decision back to the older exclusions. The paths +that join rows after ingest apply the same decision: `applyManualLinks` +skips a link between a node row and an agent row split after it, and the +presentation filter (`ListForPresentation` and the broadcast) keeps such +rows apart when no single exclusion names their IDs; an exclusion naming +the two rows exactly keeps them apart there even against a newer link +across the sets, which only a listing older than the registry's rebuild can +present as two rows. It shares the +presentation limit the broadcast paragraph below names: a third same-host +row (another agent reporting no machine ID, say) that merges with one side +first carries the pair into one row. `TestOperatorSplitOverridesProxmoxNodeAgentLink` +pins report-merge, unlink naming the node's candidate or the split rows, an +exclusion between the two candidates and one naming only the agent's +candidate, for seven link shapes (declared both ways, node-side only and +agent-side only, unclustered and clustered, an agent hostname in a different +form, and an agent reporting no machine key), through three rebuilds of a +SQLite-backed monitor adapter in the monitor listing, the broadcast coalesce +and the resources API's registry seeded from the listing or the snapshot, +after agent-first ingests (with and without the host in the snapshot) taken +while the joined-era pins still stand; then a relink of the rows (also +followed by an agent-first rebuild that persists its pins) and the same +request again. An exclusion naming the merged ID +against an unrelated ID leaves the pair merged, and the split node names the +agent only where its monitor record does. +`TestOperatorSplitFromOneAgentKeepsAnotherAgentsPin` keeps a node's own +agent's pin while another agent it was split from declares it, and +`TestOperatorSplitHoldsWhenTheAgentStopsReportingItsMachineID` keeps a split +whose agent comes back without its machine ID. + +What the split does not reach: the agents API's node link +(`/api/agents/agent/link`) records a monitor-side link intent, not a store +decision, so it does not override a resource split; rejoin through +`POST /api/resources/{id}/link`. The monitor's own node record keeps its +link where it has one, so what reads it still follows the agent: the +agent's SMART inventory as the node's disk fallback when the Proxmox disk +query fails, the agent's disk-exclude patterns on the node's disks, and the +node's linked agent that guest discovery, agent deployment and service +discovery act on. +Physical disks inside one host still join through +`resolveLinkedPhysicalDisk` without reading exclusions; a node split puts the +two sides' disks under different parents, so they no longer meet there. +Cluster- and hostname-derived IDs carry no source, so an exclusion recorded +against another resource that derives the node's own ID (a Docker Swarm host +reporting no machine key, named like the node in a swarm named like its +cluster) also splits that node from its agent, and one splitting the node's +candidate off its own ID while another source was merged there (a Docker +host linked into it) also splits it from a later agent reporting no machine +key, which joins under that ID. A rebuild that sees the agent in neither the +snapshot nor the registry, with no node link naming it, cannot tie an +exclusion naming the agent only by its candidate ID to the agent's key on a +joined-era pin, so until the agent reports again the node takes that key +and the agent's ID. The other way round, a node split from an agent that +reports no machine key and declares it refuses any joined-era pin while its +own linked agent is missing, keeping its own ID until that agent reports. An exclusion keyed by the +agent's machine-derived ID stops naming the agent if the agent's effective +canonical key changes (it stops reporting a machine ID and no pin restores +it), as every exclusion keyed by a canonical ID does. The browser's realtime +coalesce (`mergeCanonicalResourceSnapshot` in `resourceStateAdapters.ts`, +used by the websocket store and `useUnifiedResources`) receives no +exclusions and joins a Proxmox-only row and an agent-only row that share a +hostname unless their machine IDs or DMI UUIDs disagree, their Proxmox +facets name different clusters or provider scopes, or several Proxmox nodes +share the hostname and the node does not name the agent, so the UI still +renders such a split pair, from this change or from an identity-match split, +as one row. The resources API seeds its registry from the monitor's listing, which already has the link applied, so it cannot split the pair itself: an unlink shows on REST, `/api/state` and the websocket from the monitor's next registry rebuild, the same generation in which a new link reaches the broadcast. REST applies a new link at once, because its own registry applies -the store's links on top of the seed. The mock-mode view applies no operator -links, so in mock mode REST splits a manually linked pair as soon as the -unlink is recorded. Mock fixtures that identity matching merged stay merged -on every surface: the fixture graph is built without the store's exclusions. +the store's links on top of the seed. The mock-mode view now applies the +current monitor generation's operator links too ("Mock-mode unified view +applies operator links" above), so a mock unlink likewise needs that +monitor rebuild to remove the fold from the seed. Mock fixtures that identity +matching merged stay merged on every surface: the fixture graph is built +without the store's exclusions. The websocket broadcast coalesced host views with `CoalescePresentationHostResources`, which honours no exclusions, while the @@ -7234,3 +7415,14 @@ system into a VM across repeated record ingests, and `TestManualLinkFoldRepeatedPairKeepsEarlierSources` refolds a side with fewer sources than it first brought. `TestResourceAPIReportMergeExcludesRegistryLinkFolds` keeps the handler on the registry's fold record. + +### Reconciled node-agent relink fold bookkeeping + +A declared node-agent join can consume one split row before its newer +operator relink is applied. The relink records that side's ID and source in +the current holder-bound `ManualLinkFold` index, so identity-pin succession +keeps the operator decision and the resources API seed retains the same fold +for report-merge. No synthetic telemetry, identity key, enrollment or service +link is added. `TestOperatorSplitOverridesProxmoxNodeAgentLink` checks that +the monitor and its re-ingested API seed retain the fold across all split +shapes, then survive pin persistence and a repeated split. diff --git a/internal/mock/platform_fixtures.go b/internal/mock/platform_fixtures.go index 19bb49cf2b..e479469b65 100644 --- a/internal/mock/platform_fixtures.go +++ b/internal/mock/platform_fixtures.go @@ -189,8 +189,30 @@ func UnifiedResourceSnapshot() ([]unifiedresources.Resource, time.Time) { return resources, freshness } +// UnifiedResourceSnapshotWithLinks returns the current mock world as unified +// resources with the operator's manual links applied at each ingest stage, as +// the live monitor's registry rebuild applies them: a link whose side arrives +// as a supplemental record, or whose folded side an availability check names +// by source, folds as it would live. Folding the finished list instead loses +// both. The registry has no store, so nothing the fixtures ingest is +// persisted. Without links this is UnifiedResourceSnapshot; with links it +// builds afresh on every call, so callers cache it per fixture data version +// and link set, and its result is the caller's own. +func UnifiedResourceSnapshotWithLinks(links []unifiedresources.ResourceLink) ([]unifiedresources.Resource, time.Time) { + if len(links) == 0 { + return UnifiedResourceSnapshot() + } + if !IsMockEnabled() { + return nil, time.Time{} + } + return CurrentFixtureGraph().unifiedResourceSnapshot(unifiedresources.NewRegistryWithManualLinks(links)) +} + func (g FixtureGraph) UnifiedResourceSnapshot() ([]unifiedresources.Resource, time.Time) { - registry := unifiedresources.NewRegistry(nil) + return g.unifiedResourceSnapshot(unifiedresources.NewRegistry(nil)) +} + +func (g FixtureGraph) unifiedResourceSnapshot(registry *unifiedresources.ResourceRegistry) ([]unifiedresources.Resource, time.Time) { registry.IngestSnapshot(unifiedresources.SnapshotWithoutSources(g.State, SupplementalOwnedSources())) for _, source := range SupplementalOwnedSources() { diff --git a/internal/mock/platform_fixtures_test.go b/internal/mock/platform_fixtures_test.go index d7243b8704..7f25908432 100644 --- a/internal/mock/platform_fixtures_test.go +++ b/internal/mock/platform_fixtures_test.go @@ -459,6 +459,103 @@ func TestFixtureGraphAttachesServiceAvailabilityFixturesToServiceResources(t *te } } +// TestFixtureGraphAppliesManualLinksAtEachIngestStage pins the linked build +// behind the mock-mode unified view. Folding the curated Docker service into a +// VM must move its availability check onto the VM, as the live rebuild's +// availability ingest resolves the check's source reference through the fold; +// folding the finished list dropped the facet with the folded row. +func TestFixtureGraphAppliesManualLinksAtEachIngestStage(t *testing.T) { + now := time.Date(2026, time.May, 6, 12, 0, 0, 0, time.UTC) + graph := buildFixtureGraph(DefaultConfig, now) + unlinked, _ := graph.UnifiedResourceSnapshot() + + const checkID = "mock-availability-docker-frontend-service" + var serviceID, vmID string + for _, resource := range unlinked { + switch { + case resource.Type == unifiedresources.ResourceTypeDockerService && resource.Availability != nil && + resource.Availability.TargetID == checkID: + serviceID = resource.ID + case vmID == "" && resource.Type == unifiedresources.ResourceTypeVM && resource.Proxmox != nil: + vmID = resource.ID + } + } + if serviceID == "" || vmID == "" { + t.Fatalf("fixture graph needs the checked Docker service and a Proxmox VM, got service=%q vm=%q", serviceID, vmID) + } + + links := []unifiedresources.ResourceLink{{ResourceA: vmID, ResourceB: serviceID, PrimaryID: vmID}} + linked, _ := graph.unifiedResourceSnapshot(unifiedresources.NewRegistryWithManualLinks(links)) + var vm, endpoint *unifiedresources.Resource + for i := range linked { + switch { + case linked[i].ID == serviceID: + t.Fatalf("linked build still lists the folded Docker service %s", serviceID) + case linked[i].ID == vmID: + vm = &linked[i] + case linked[i].Type == unifiedresources.ResourceTypeNetworkEndpoint && linked[i].Availability != nil && + linked[i].Availability.TargetID == checkID: + endpoint = &linked[i] + } + } + if vm == nil || vm.Docker == nil { + t.Fatalf("linked VM %s = %+v, want it to carry the folded service's Docker facet", vmID, vm) + } + if vm.Availability == nil || vm.Availability.TargetID != checkID { + t.Fatalf("linked VM availability = %+v, want the folded service's check projected onto it", vm.Availability) + } + if !hasChecksEdgeTo(endpoint, vmID) { + t.Fatalf("check relationships = %+v, want the checks edge to follow the fold to %s", endpoint, vmID) + } +} + +// TestUnifiedResourceSnapshotWithLinksLeavesTheSharedSnapshotUnlinked pins the +// memo boundary: a linked build belongs to its caller (one tenant's links), +// and the memoized snapshot every other reader shares stays unlinked. +func TestUnifiedResourceSnapshotWithLinksLeavesTheSharedSnapshotUnlinked(t *testing.T) { + previous := IsMockEnabled() + previousConfig := GetConfig() + if previous { + mustSetEnabled(t, false) + } + testConfig := previousConfig + testConfig.UpdateInterval = 5 * time.Minute + SetMockConfig(testConfig) + mustSetEnabled(t, true) + t.Cleanup(func() { + mustSetEnabled(t, false) + SetMockConfig(previousConfig) + mustSetEnabled(t, previous) + }) + + shared, _ := UnifiedResourceSnapshot() + if same, _ := UnifiedResourceSnapshotWithLinks(nil); len(same) == 0 || &same[0] != &shared[0] { + t.Fatal("UnifiedResourceSnapshotWithLinks without links must serve the memoized snapshot") + } + var vmID, agentID string + for _, resource := range shared { + switch { + case vmID == "" && resource.Type == unifiedresources.ResourceTypeVM && resource.Agent == nil: + vmID = resource.ID + case agentID == "" && resource.Type == unifiedresources.ResourceTypeAgent && resource.Agent != nil && + len(resource.Sources) == 1 && resource.Sources[0] == unifiedresources.SourceAgent: + agentID = resource.ID + } + } + if vmID == "" || agentID == "" { + t.Fatalf("fixture graph needs an agentless VM and a standalone agent, got vm=%q agent=%q", vmID, agentID) + } + + linked, _ := UnifiedResourceSnapshotWithLinks([]unifiedresources.ResourceLink{{ResourceA: vmID, ResourceB: agentID, PrimaryID: vmID}}) + if slices.ContainsFunc(linked, func(r unifiedresources.Resource) bool { return r.ID == agentID }) { + t.Fatalf("linked build still lists the folded agent %s", agentID) + } + again, _ := UnifiedResourceSnapshot() + if !slices.ContainsFunc(again, func(r unifiedresources.Resource) bool { return r.ID == agentID }) { + t.Fatalf("the shared snapshot lost the agent %s after a linked build", agentID) + } +} + // hasChecksEdgeTo reports whether the source-owned check row carries the // outgoing checks relationship to the resource it matched. The check row owns // that edge; the matched resource carries only the projected facet. diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index b8e69690b4..76f102724c 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2006,7 +2006,7 @@ func TestMockUnifiedStateViewUsesCanonicalMockFixtureGraph(t *testing.T) { requiredSnippets := []string{ "version := mock.FixtureDataVersion()", "m.mockUnifiedViewValid && m.mockUnifiedViewVersion == version", - "resources, freshness := mock.UnifiedResourceSnapshot()", + "resources, freshness := mock.UnifiedResourceSnapshotWithLinks(links)", "view := monitorUnifiedStateViewFromResources(resources, freshness)", "m.mockUnifiedView = view", "m.mockUnifiedViewVersion = version", diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 2edcc12812..317d4130dc 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1216,6 +1216,8 @@ type Monitor struct { mockUnifiedViewMu sync.Mutex mockUnifiedView monitorUnifiedStateView mockUnifiedViewVersion uint64 + mockUnifiedViewLinks []unifiedresources.ResourceLink + mockUnifiedViewFixtureAt time.Time // fixture freshness the published view was built from mockUnifiedViewValid bool pveClients map[string]PVEClientInterface pbsClients map[string]*pbs.Client @@ -5205,27 +5207,55 @@ func (m *Monitor) currentUnifiedStateView() monitorUnifiedStateView { } if mock.IsMockEnabled() { - // Read the version before the snapshot so a tick landing in between - // caches newer data under an older token (harmless rebuild next - // call) rather than ever serving stale data under a newer one. + // Read the version and links before the snapshot so a tick or + // rebuild landing in between caches newer data under an older + // token (harmless rebuild next call) rather than ever serving stale + // data under a newer one. version := mock.FixtureDataVersion() + links := m.resourceStoreManualLinks() m.mockUnifiedViewMu.Lock() - if m.mockUnifiedViewValid && m.mockUnifiedViewVersion == version { + if m.mockUnifiedViewValid && m.mockUnifiedViewVersion == version && sameManualLinks(m.mockUnifiedViewLinks, links) { view := m.mockUnifiedView m.mockUnifiedViewMu.Unlock() return view } m.mockUnifiedViewMu.Unlock() - resources, freshness := mock.UnifiedResourceSnapshot() + // The fixture graph knows no link store, so the operator's links, + // as the resource store's current generation applied them, are + // applied while it is built, at each ingest stage like the live + // rebuild. Its registry has no store: fixture data never reaches + // durable state through it. + resources, freshness := mock.UnifiedResourceSnapshotWithLinks(links) if len(resources) > 0 || !freshness.IsZero() { // Consumers share this view between ticks, mirroring the // sharing semantics the persistent-store ReadState path has // always had in real mode: views are read-only. view := monitorUnifiedStateViewFromResources(resources, freshness) m.mockUnifiedViewMu.Lock() + // Readers key caches on freshness: the resources API reuses + // its registry, seeded from this list, while freshness is + // equal. A link change alters the view without moving any + // fixture timestamp, and a slow build can publish after a + // newer one, so freshness is assigned here, in publication + // order: a view built from the fixture freshness and links of + // the one it replaces keeps its freshness, and any other view + // gets one later than every view published before it. Fixture + // freshness stands in for the fixture data here, as it already + // does for that cache. + fixtureFreshness := view.freshness + if m.mockUnifiedViewValid { + previous := m.mockUnifiedView.freshness + if fixtureFreshness.Equal(m.mockUnifiedViewFixtureAt) && sameManualLinks(m.mockUnifiedViewLinks, links) { + view.freshness = previous + } else if !view.freshness.After(previous) { + view.freshness = previous.Add(time.Nanosecond) + } + } m.mockUnifiedView = view + m.mockUnifiedViewFixtureAt = fixtureFreshness m.mockUnifiedViewVersion = version + m.mockUnifiedViewLinks = links m.mockUnifiedViewValid = true m.mockUnifiedViewMu.Unlock() return view @@ -5261,6 +5291,33 @@ func (m *Monitor) currentUnifiedStateView() monitorUnifiedStateView { return m.unifiedStateViewWithStandaloneHostContinuity(monitorUnifiedStateViewFromSnapshot(m.GetState())) } +// manualLinkResourceStore is implemented by resource stores that apply the +// operator's manual links (MonitorAdapter). +type manualLinkResourceStore interface { + ManualLinks() []unifiedresources.ResourceLink +} + +// resourceStoreManualLinks returns the operator links the resource store's +// current generation applied, or nil when the store applies none. +func (m *Monitor) resourceStoreManualLinks() []unifiedresources.ResourceLink { + m.mu.RLock() + store := m.resourceStore + m.mu.RUnlock() + linkStore, ok := store.(manualLinkResourceStore) + if !ok { + return nil + } + return linkStore.ManualLinks() +} + +// sameManualLinks reports whether two link lists fold the same resources in +// the same order. Link metadata (reason, author, time) does not change a fold. +func sameManualLinks(a, b []unifiedresources.ResourceLink) bool { + return slices.EqualFunc(a, b, func(x, y unifiedresources.ResourceLink) bool { + return x.ResourceA == y.ResourceA && x.ResourceB == y.ResourceB && x.PrimaryID == y.PrimaryID + }) +} + func (m *Monitor) currentUnifiedResourceFreshness() time.Time { if m == nil { return time.Time{} @@ -5345,7 +5402,11 @@ func (m *Monitor) GetUnifiedReadStateOrSnapshot() unifiedresources.ReadState { // the new epoch replaces it, so they get a view of current state meanwhile. // The view lists the state snapshot without provider-owned supplemental // resources or persisted manual links and copies the whole estate, so it is -// only for callers outside per-resource loops. +// only for callers outside per-resource loops. The links stay out because the +// view is a plain projection of current state for a short window: the links +// still apply in the registry once a rebuild in the new epoch lands. Meanwhile +// typed lookups see linked resources unmerged; the previous-state carry, for +// one, loses a manually linked agent's memory. func (m *Monitor) currentModeReadState() unifiedresources.ReadState { if m == nil { return nil diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index a0a25ba99b..e2919b8a8f 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -7,6 +7,7 @@ import ( "fmt" "net" "os" + "slices" "strings" "sync/atomic" "testing" @@ -5795,6 +5796,202 @@ func TestMockModeDropsRealHostContinuity(t *testing.T) { } } +// TestMockUnifiedViewAppliesOperatorManualLinks pins operator links on the +// mock-mode unified view. The fixture graph is built without the link store, +// so a standalone agent linked into a VM showed merged in the resources API, +// which re-seeds through the store-backed registry, but stayed a separate row +// in the websocket broadcast and /api/state. The view applies the links the +// monitor's resource store last loaded, rebuilds when they change without +// waiting for a fixture tick, and never forwards fixture data to the store. +func TestMockUnifiedViewAppliesOperatorManualLinks(t *testing.T) { + previous := mock.IsMockEnabled() + previousConfig := mock.GetConfig() + if previous { + mustSetMockEnabled(t, false) + } + // Hold the fixture graph still, so only the link can change the view. + testConfig := previousConfig + testConfig.UpdateInterval = 5 * time.Minute + mock.SetMockConfig(testConfig) + mustSetMockEnabled(t, true) + t.Cleanup(func() { + mustSetMockEnabled(t, false) + mock.SetMockConfig(previousConfig) + mustSetMockEnabled(t, previous) + }) + + store := unifiedresources.NewMemoryStore() + m := &Monitor{ + state: models.NewState(), + resourceStore: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store)), + alertManager: alerts.NewManager(), + } + t.Cleanup(m.alertManager.Stop) + + unlinked := m.currentUnifiedStateView() + var vmID, agentID string + for _, resource := range unlinked.resources { + switch { + case vmID == "" && resource.Type == unifiedresources.ResourceTypeVM && resource.Agent == nil: + vmID = resource.ID + case agentID == "" && resource.Type == unifiedresources.ResourceTypeAgent && resource.Agent != nil && + len(resource.Sources) == 1 && resource.Sources[0] == unifiedresources.SourceAgent: + agentID = resource.ID + } + } + if vmID == "" || agentID == "" { + t.Fatalf("fixture graph needs an agentless VM and a standalone agent, got vm=%q agent=%q", vmID, agentID) + } + if err := store.AddLink(unifiedresources.ResourceLink{ResourceA: vmID, ResourceB: agentID, PrimaryID: vmID}); err != nil { + t.Fatalf("AddLink: %v", err) + } + version := mock.FixtureDataVersion() + + // The broadcast's read-path refresh rebuilds the resource store, which + // loads the link; the mock view must follow it within the same build. + frontend := m.BuildBroadcastFrontendState() + if got := mock.FixtureDataVersion(); got != version { + t.Fatalf("fixture data version moved from %d to %d; the link alone must refresh the view", version, got) + } + var vmRow *models.ResourceFrontend + for i := range frontend.Resources { + switch frontend.Resources[i].ID { + case vmID: + vmRow = &frontend.Resources[i] + case agentID: + t.Fatalf("broadcast still lists the linked agent %s as its own row", agentID) + } + } + if vmRow == nil || len(vmRow.Agent) == 0 { + t.Fatalf("broadcast VM row %s = %+v, want it to carry the linked agent's facet", vmID, vmRow) + } + + linked := m.currentUnifiedStateView() + if linked.readState == unlinked.readState { + t.Fatal("expected the view to rebuild when the operator links changed") + } + // The resources API keys its registry cache on this freshness, so a + // link change must advance it even though no fixture tick did. + if !linked.freshness.After(unlinked.freshness) { + t.Fatalf("view freshness %v did not advance past %v on the link change", linked.freshness, unlinked.freshness) + } + if again := m.currentUnifiedStateView(); again.readState != linked.readState { + t.Fatal("expected the linked view to stay cached while neither the fixtures nor the links change") + } + + // The resources API re-seeds the same snapshot through a registry backed + // by the link store. Both payloads must list the same rows. + seed, _ := m.UnifiedResourceSnapshot() + rest := unifiedresources.NewRegistry(store) + rest.IngestResources(seed) + viewIDs := make([]string, 0, len(linked.resources)) + for _, resource := range linked.resources { + viewIDs = append(viewIDs, resource.ID) + } + restIDs := make([]string, 0, len(seed)) + for _, resource := range rest.List() { + restIDs = append(restIDs, resource.ID) + } + slices.Sort(viewIDs) + slices.Sort(restIDs) + if !slices.Equal(viewIDs, restIDs) { + t.Fatalf("mock view and resources API disagree:\nview %v\nrest %v", viewIDs, restIDs) + } + + // The view's registry has no store: history recorded through the read + // state it hands out must not land fixture rows in the operator's store. + recorder, ok := linked.readState.(interface { + RecordChange(unifiedresources.ResourceChange) error + }) + if !ok { + t.Fatalf("mock view read state %T does not record changes", linked.readState) + } + before, err := store.GetRecentChanges(vmID, time.Time{}, 1000) + if err != nil { + t.Fatalf("GetRecentChanges: %v", err) + } + if err := recorder.RecordChange(unifiedresources.ResourceChange{ + ID: "mock-view-probe", + ObservedAt: time.Now().UTC(), + ResourceID: vmID, + Kind: unifiedresources.ChangeStateTransition, + }); err != nil { + t.Fatalf("RecordChange: %v", err) + } + after, err := store.GetRecentChanges(vmID, time.Time{}, 1000) + if err != nil { + t.Fatalf("GetRecentChanges: %v", err) + } + if len(after) != len(before) { + t.Fatalf("mock view wrote %d change rows into the durable store", len(after)-len(before)) + } + + // Reconcile the mock link proposal with the current latest-decision + // rule: a fresh monitor must restore the link, and unlink/relink must + // invalidate the seed cache without a fixture tick. Mock seeds retain + // their own default thresholds, not the live adapter's configured ones. + restarted := &Monitor{ + state: models.NewState(), + config: &config.Config{PVEPollingInterval: 2 * time.Minute}, + resourceStore: unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(store)), + alertManager: alerts.NewManager(), + } + t.Cleanup(restarted.alertManager.Stop) + adapter := restarted.resourceStore.(*unifiedresources.MonitorAdapter) + assertMockLink := func(wantLinked bool) time.Time { + t.Helper() + frontend := restarted.BuildBroadcastFrontendState() + var vmFound, agentFound, vmHasAgent bool + for _, resource := range frontend.Resources { + if resource.ID == vmID { + vmFound, vmHasAgent = true, len(resource.Agent) > 0 + } + if resource.ID == agentID { + agentFound = true + } + } + if !vmFound || vmHasAgent != wantLinked || agentFound == wantLinked { + t.Fatalf("mock link=%t: vm=%t vmAgent=%t standaloneAgent=%t", wantLinked, vmFound, vmHasAgent, agentFound) + } + seed, freshness, thresholds := restarted.UnifiedResourceSnapshotWithStaleThresholds() + if len(thresholds) != 0 { + t.Fatalf("mock seed borrowed live thresholds: %v", thresholds) + } + if got := adapter.StaleThresholds()[unifiedresources.SourceProxmox]; got != 4*time.Minute { + t.Fatalf("live adapter threshold = %v, want configured four minutes", got) + } + rest := unifiedresources.NewRegistryWithStaleThresholds(store, thresholds) + rest.IngestResources(seed) + _, hasStandaloneAgent := rest.Get(agentID) + if hasStandaloneAgent == wantLinked { + t.Fatalf("REST seed link=%t: standalone agent=%t", wantLinked, hasStandaloneAgent) + } + if got := mock.FixtureDataVersion(); got != version { + t.Fatalf("fixture version moved from %d to %d during operator lifecycle", version, got) + } + return freshness + } + restartFreshness := assertMockLink(true) + if err := store.AddExclusion(unifiedresources.ResourceExclusion{ResourceA: vmID, ResourceB: agentID}); err != nil { + t.Fatalf("unlink: %v", err) + } + // Reads deliberately reuse a generation younger than two seconds. + // Apply the owning ingest/rebuild boundary instead of waiting for that + // throttle or claiming a recorded decision changes a previous listing. + restarted.updateResourceStore(restarted.currentStateWithScope()) + unlinkFreshness := assertMockLink(false) + if !unlinkFreshness.After(restartFreshness) { + t.Fatal("unlink must advance the mock seed's freshness") + } + if err := store.AddLink(unifiedresources.ResourceLink{ResourceA: agentID, ResourceB: vmID, PrimaryID: vmID}); err != nil { + t.Fatalf("relink: %v", err) + } + restarted.updateResourceStore(restarted.currentStateWithScope()) + if relinkFreshness := assertMockLink(true); !relinkFreshness.After(unlinkFreshness) { + t.Fatal("relink must advance the mock seed's freshness") + } +} + // TestMockModeDiscardsRealHostReports pins the push-side of the mock clean // room. Mock mode already suspends pull-based PVE/PBS/PMG collection, but a // real agent keeps POSTing regardless; ingesting those reports lands a real diff --git a/internal/unifiedresources/canonical_id_pins.go b/internal/unifiedresources/canonical_id_pins.go index 847380b1bd..c48dffd3dc 100644 --- a/internal/unifiedresources/canonical_id_pins.go +++ b/internal/unifiedresources/canonical_id_pins.go @@ -265,7 +265,8 @@ func (rr *ResourceRegistry) completeIdentityFromPins(source DataSource, sourceID // The exclusion names the IDs the record holds without the pin. if len(rr.exclusions) > 0 && (rr.isExcluded(pin.CanonicalID, rr.chooseNewID(resource.Type, identity, source, sourceID)) || - rr.isExcluded(pin.CanonicalID, rr.sourceSpecificID(resource.Type, source, sourceID))) { + rr.isExcluded(pin.CanonicalID, rr.sourceSpecificID(resource.Type, source, sourceID)) || + rr.proxmoxNodePinOfSplitAgentLocked(source, sourceID, resource, identity, pin)) { return identity } if strings.TrimSpace(identity.MachineID) == "" { diff --git a/internal/unifiedresources/manual_links.go b/internal/unifiedresources/manual_links.go new file mode 100644 index 0000000000..0d4de9097c --- /dev/null +++ b/internal/unifiedresources/manual_links.go @@ -0,0 +1,30 @@ +package unifiedresources + +import "slices" + +// NewRegistryWithManualLinks creates a registry that applies the given +// operator links on every ingest but has no store behind it. Read views built +// from data that never passed through the durable registry, such as the mock +// fixture graph, use it to show the operator's links the way the store-backed +// registry does, while nothing they ingest can persist identity pins, +// canonical-ID successions or change records. It carries no exclusions, so +// its identity matching and ListForPresentation see none. +func NewRegistryWithManualLinks(links []ResourceLink) *ResourceRegistry { + rr := NewRegistry(nil) + rr.links = slices.Clone(links) + return rr +} + +// ManualLinks returns the operator's manual links that the adapter's current +// registry generation applied, as loaded from its store when that generation +// was built. A link added to the store since then appears after the next +// rebuild, which is when the adapter's own listing folds it too. +func (a *MonitorAdapter) ManualLinks() []ResourceLink { + registry := a.currentRegistry() + if registry == nil { + return nil + } + registry.mu.RLock() + defer registry.mu.RUnlock() + return slices.Clone(registry.links) +} diff --git a/internal/unifiedresources/presentation_exclusions.go b/internal/unifiedresources/presentation_exclusions.go index b00f0dd4c2..dde1f8325b 100644 --- a/internal/unifiedresources/presentation_exclusions.go +++ b/internal/unifiedresources/presentation_exclusions.go @@ -4,17 +4,23 @@ import "time" // presentationExclusionFilter reports whether the presentation host coalesce // must keep two resources apart because the operator split the pair (unlink -// or report-merge). It reads a copy of the registry's exclusions, so callers -// coalesce without holding the registry lock; nil means no pair is excluded. +// or report-merge): an exclusion names the two rows, or they are a Proxmox +// node and an agent split across the IDs they can hold (nodeAgentSplit). It +// reads a copy of the registry's decisions, so callers coalesce without +// holding the registry lock; nil means no pair is excluded. func (rr *ResourceRegistry) presentationExclusionFilter() func(left, right Resource) bool { rr.mu.RLock() if len(rr.exclusions) == 0 { rr.mu.RUnlock() return nil } - exclusions := make(map[string]struct{}, len(rr.exclusions)) - for key := range rr.exclusions { - exclusions[key] = struct{}{} + // The link index is built once per registry and never written again. + decisions := operatorPairDecisions{ + exclusions: make(map[string]time.Time, len(rr.exclusions)), + linksByID: rr.linksByID, + } + for key, at := range rr.exclusions { + decisions.exclusions[key] = at } rr.mu.RUnlock() @@ -24,8 +30,10 @@ func (rr *ResourceRegistry) presentationExclusionFilter() func(left, right Resou if leftID == "" || rightID == "" { return false } - _, ok := exclusions[exclusionKey(leftID, rightID)] - return ok + if _, ok := decisions.exclusions[exclusionKey(leftID, rightID)]; ok { + return true + } + return rr.nodeAgentRowsSplit(decisions, &left, &right) } } diff --git a/internal/unifiedresources/proxmox_node_links.go b/internal/unifiedresources/proxmox_node_links.go index e38e0d2de2..a290abd831 100644 --- a/internal/unifiedresources/proxmox_node_links.go +++ b/internal/unifiedresources/proxmox_node_links.go @@ -2,11 +2,302 @@ package unifiedresources import ( "net" + "slices" "strings" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/models" ) +// A Proxmox node's link to a pulse-agent is Pulse's own inference, not an +// operator decision: monitoring links a node to the one agent whose endpoint +// address or hostname matches it, and inferLinkedHostsForProxmoxNodes and +// resolveLinkedResource corroborate that link. Cloned machines and reused +// short hostnames make it wrong at times, which is what the resources API's +// unlink and report-merge are for, so an operator's split overrides the link +// as it overrides identity matching in findMatch. A later POST +// /api/resources/{id}/link between the two rows joins them again. + +// nodeAgentLinkSide is one side of a Proxmox node<->agent pair: the record's +// source ID, the identity it derives IDs from and the resource ID it holds in +// the registry, if any. +type nodeAgentLinkSide struct { + sourceID string + identity ResourceIdentity + heldID string +} + +// operatorPairDecisions is the operator's manual decisions a node<->agent +// split is read from: a registry's own, or the copy the presentation filter +// holds outside the registry lock. +type operatorPairDecisions struct { + exclusions map[string]time.Time + linksByID map[string][]ResourceLink +} + +func (rr *ResourceRegistry) operatorPairDecisionsLocked() operatorPairDecisions { + return operatorPairDecisions{exclusions: rr.exclusions, linksByID: rr.linksByID} +} + +// indexLinksByID lists each manual link under both of its IDs. +func indexLinksByID(links []ResourceLink) map[string][]ResourceLink { + if len(links) == 0 { + return nil + } + index := make(map[string][]ResourceLink, 2*len(links)) + for _, link := range links { + a, b := CanonicalResourceID(link.ResourceA), CanonicalResourceID(link.ResourceB) + index[a] = append(index[a], link) + if b != a { + index[b] = append(index[b], link) + } + } + return index +} + +// nodeAgentSplit reports whether the operator split the node off a +// node<->agent pair: an exclusion names one of the node's own IDs against one +// of the agent's IDs, and no link between those IDs postdates it (a tie keeps +// the pair apart, as it does for one pair in effectiveManualPairDecisions). +// +// Unlink and report-merge name the merged resource's ID and the source-specific +// candidate IDs its SourceTargets list, and a split moves the node onto an ID +// it did not hold while joined, so decisions are read across sets rather +// than one exact pair. The node's own IDs are its source-specific ID and the +// ID its identity derives without a machine key. The agent's are its +// source-specific ID, the ID its identity derives, the ID it holds, and the +// ID the joined pair derives (the agent's machine-derived ID once a node +// takes its identity). An exclusion naming only the agent's candidate against +// the joined ID does not count: report-merge records that same pair when it +// splits another source, such as a Docker host, off the agent's +// machine-derived ID. +// +// A link between the two sets that postdates the newest exclusion wins: a +// relink names the rows a split left, a different pair than the exclusions +// report-merge recorded against the merged ID, which the store's +// one-decision-per-pair rule cannot match. +func (rr *ResourceRegistry) nodeAgentSplit(decisions operatorPairDecisions, node, agent nodeAgentLinkSide) bool { + if len(decisions.exclusions) == 0 { + return false + } + nodeIDs, agentIDs := rr.nodeAgentPairIDs(node, agent) + var splitAt time.Time + split := false + for _, nodeID := range nodeIDs { + for _, agentID := range agentIDs { + if nodeID == agentID { + continue + } + if at, ok := decisions.exclusions[exclusionKey(nodeID, agentID)]; ok && (!split || at.After(splitAt)) { + split, splitAt = true, at + } + } + } + if !split { + return false + } + for _, nodeID := range nodeIDs { + for _, link := range decisions.linksByID[nodeID] { + other := CanonicalResourceID(link.ResourceB) + if other == nodeID { + other = CanonicalResourceID(link.ResourceA) + } + if other != nodeID && slices.Contains(agentIDs, other) && link.CreatedAt.After(splitAt) { + return false + } + } + } + return true +} + +// nodeAgentPairIDs returns the node's own IDs and the agent's IDs a split is +// read across. A Proxmox node reports no machine ID or DMI UUID of its own: +// either one on its identity came from a linked agent or an identity pin, so +// the node's own ID is derived without them. The two IDs are then the same +// whether or not the node took the agent's identity in this ingest. +func (rr *ResourceRegistry) nodeAgentPairIDs(node, agent nodeAgentLinkSide) (nodeIDs, agentIDs []string) { + nodeIDs = rr.proxmoxNodeOwnIDs(node.sourceID, node.identity) + if agent.sourceID != "" { + agentIDs = append(agentIDs, rr.sourceSpecificID(ResourceTypeAgent, SourceAgent, agent.sourceID)) + } + if agent.sourceID != "" || agent.identity.MachineID != "" || agent.identity.DMIUUID != "" { + agentIDs = append(agentIDs, rr.chooseNewID(ResourceTypeAgent, agent.identity, SourceAgent, agent.sourceID)) + } + if held := CanonicalResourceID(agent.heldID); held != "" { + agentIDs = append(agentIDs, held) + } + // The joined ID counts as the agent's even where it is also one of the + // node's (an agent reporting no machine key joins under the node's own + // ID): an exclusion splitting the node's candidate off it then splits + // the node from the agent, and the equal pair is skipped. + nodeIdentity := node.identity + nodeIdentity.MachineID, nodeIdentity.DMIUUID = "", "" + agentIDs = append(agentIDs, rr.chooseNewID(ResourceTypeAgent, mergeIdentity(nodeIdentity, agent.identity), SourceProxmox, node.sourceID)) + return nodeIDs, agentIDs +} + +// proxmoxNodeOwnIDs returns the IDs a Proxmox node record holds without any +// agent: its source-specific ID and the ID its identity derives with the +// machine keys it cannot report itself removed. +func (rr *ResourceRegistry) proxmoxNodeOwnIDs(sourceID string, identity ResourceIdentity) []string { + identity.MachineID, identity.DMIUUID = "", "" + return []string{ + rr.sourceSpecificID(ResourceTypeAgent, SourceProxmox, sourceID), + rr.chooseNewID(ResourceTypeAgent, identity, SourceProxmox, sourceID), + } +} + +// proxmoxNodeAgentSplitLocked applies nodeAgentSplit to the registry's own +// decisions. Callers hold rr.mu. +func (rr *ResourceRegistry) proxmoxNodeAgentSplitLocked(node, agent nodeAgentLinkSide) bool { + return rr.nodeAgentSplit(rr.operatorPairDecisionsLocked(), node, agent) +} + +// proxmoxNodePinOfSplitAgentLocked reports whether an identity pin a Proxmox +// node record would complete its machine keys from belongs to an agent the +// operator split the node from. A node reports no machine key of its own, so +// a pin's is an agent's, written while the pair was joined, and completing +// it would mint that agent's machine-derived ID and fold the node into the +// agent by ID or by identity. The pin's canonical ID and keys stand for the +// agent the node's link or an agent row holding the pin's ID names, so this +// holds while the agent is missing from the snapshot. The agent this +// snapshot's split dropped (rr.splitAgents) can be another agent than the +// pin's, so it refuses only a pin that can be its own. +func (rr *ResourceRegistry) proxmoxNodePinOfSplitAgentLocked(source DataSource, sourceID string, resource Resource, identity ResourceIdentity, pin ResourceIdentityPin) bool { + if source != SourceProxmox || resource.Proxmox == nil || len(rr.exclusions) == 0 { + return false + } + node := nodeAgentLinkSide{sourceID: sourceID, identity: identity} + pinAgent := nodeAgentLinkSide{ + sourceID: normalizeSourceID(resource.Proxmox.LinkedAgentID), + identity: ResourceIdentity{MachineID: pin.MachineID, DMIUUID: pin.DMIUUID}, + heldID: pin.CanonicalID, + } + if rr.proxmoxNodeAgentSplitLocked(node, pinAgent) { + return true + } + if held := agentOnlyRow(rr.resources[CanonicalResourceID(pin.CanonicalID)]); held != nil { + if heldSourceID := normalizeSourceID(held.Agent.AgentID); heldSourceID != pinAgent.sourceID { + pinAgent.sourceID = heldSourceID + if rr.proxmoxNodeAgentSplitLocked(node, pinAgent) { + return true + } + } + } + split, ok := rr.splitAgents[normalizeSourceID(sourceID)] + return ok && pinBelongsToAgent(pin, split.identity) && rr.proxmoxNodeAgentSplitLocked(node, split) +} + +// pinBelongsToAgent reports whether an identity pin can be the agent's: only +// a key both report that differs shows the pin is another agent's. An agent +// that stops reporting its machine ID keeps its split that way, while another +// agent with its own key does not claim the pin. +func pinBelongsToAgent(pin ResourceIdentityPin, agent ResourceIdentity) bool { + if pinKey, agentKey := strings.TrimSpace(pin.MachineID), strings.TrimSpace(agent.MachineID); pinKey != "" && agentKey != "" { + return strings.EqualFold(pinKey, agentKey) + } + if pinKey, agentKey := strings.TrimSpace(pin.DMIUUID), strings.TrimSpace(agent.DMIUUID); pinKey != "" && agentKey != "" { + return strings.EqualFold(pinKey, agentKey) + } + return true +} + +// joinedPairOwnsIDLocked reports whether id is an ID one side of a joined +// Proxmox node+agent row holds on its own: the node's (derived from the +// row's Proxmox facet, not its merged identity, whose first hostname can be +// the agent's) or the agent's source-specific ID, which an agent reporting +// no machine key holds when split. +func (rr *ResourceRegistry) joinedPairOwnsIDLocked(joined *Resource, id string) bool { + if joined == nil || joined.Proxmox == nil || joined.Agent == nil || CanonicalResourceType(joined.Type) != ResourceTypeAgent { + return false + } + id = CanonicalResourceID(id) + if agentID := normalizeSourceID(joined.Agent.AgentID); agentID != "" && id == rr.sourceSpecificID(ResourceTypeAgent, SourceAgent, agentID) { + return true + } + sourceID := normalizeSourceID(joined.Proxmox.SourceID) + if sourceID == "" { + return false + } + _, identity := resourceFromProxmoxNode(models.Node{ + ID: sourceID, + Name: joined.Proxmox.NodeName, + NativeNameAliases: joined.Proxmox.NodeAliases, + Instance: joined.Proxmox.Instance, + ClusterName: joined.Proxmox.ClusterName, + }, nil) + return slices.Contains(rr.proxmoxNodeOwnIDs(sourceID, identity), id) +} + +// nodeAgentRowsSplit reports whether two listed rows are a Proxmox node and +// an agent the operator split, for the paths that join rows after ingest: +// manual links (an older link between the rows a split left loses to a newer +// report-merge) and the presentation host coalesce (the rows can hold IDs no +// single exclusion names, a clustered node's cluster-derived ID for one). +func (rr *ResourceRegistry) nodeAgentRowsSplit(decisions operatorPairDecisions, left, right *Resource) bool { + node, agent := proxmoxNodeRow(left), agentOnlyRow(right) + if node == nil || agent == nil { + node, agent = proxmoxNodeRow(right), agentOnlyRow(left) + } + if node == nil || agent == nil { + return false + } + return rr.nodeAgentSplit(decisions, + nodeAgentLinkSide{sourceID: normalizeSourceID(node.Proxmox.SourceID), identity: node.Identity}, + nodeAgentLinkSide{sourceID: normalizeSourceID(agent.Agent.AgentID), identity: agent.Identity, heldID: agent.ID}, + ) +} + +func proxmoxNodeRow(resource *Resource) *Resource { + if resource == nil || CanonicalResourceType(resource.Type) != ResourceTypeAgent || + resource.Proxmox == nil || resource.Agent != nil || strings.TrimSpace(resource.Proxmox.SourceID) == "" { + return nil + } + return resource +} + +func agentOnlyRow(resource *Resource) *Resource { + if resource == nil || CanonicalResourceType(resource.Type) != ResourceTypeAgent || + resource.Agent == nil || resource.Proxmox != nil || strings.TrimSpace(resource.Agent.AgentID) == "" { + return nil + } + return resource +} + +// splitProxmoxNodeLink returns the node record and the linked host it takes +// identity from, dropping the host for a node the operator split from that +// host's agent. A linked node takes the agent's identity, and with it the +// agent's machine-derived canonical ID, before the agent is ingested, so +// refusing only the later merge would leave the node holding the agent's ID +// and machine key, and identity matching would join the agent to it again. +// Without the agent's identity the node keeps its own ID, hostname and +// endpoint address, which identity matching never merges on. rr.splitAgents +// remembers the agent for the pin guard, which reads the split from it where +// the node record itself does not name that agent; it is rewritten for every +// node on every snapshot, and nothing else reads it. +func (rr *ResourceRegistry) splitProxmoxNodeLink(node models.Node, host *models.Host) (models.Node, *models.Host) { + nodeID := normalizeSourceID(node.ID) + rr.mu.Lock() + defer rr.mu.Unlock() + delete(rr.splitAgents, nodeID) + if host == nil || len(rr.exclusions) == 0 { + return node, host + } + _, nodeIdentity := resourceFromProxmoxNode(node, nil) + _, hostIdentity := resourceFromHost(*host) + if !rr.proxmoxNodeAgentSplitLocked( + nodeAgentLinkSide{sourceID: nodeID, identity: nodeIdentity}, + nodeAgentLinkSide{sourceID: normalizeSourceID(host.ID), identity: hostIdentity}, + ) { + return node, host + } + if rr.splitAgents == nil { + rr.splitAgents = make(map[string]nodeAgentLinkSide) + } + rr.splitAgents[nodeID] = nodeAgentLinkSide{sourceID: normalizeSourceID(host.ID), identity: hostIdentity} + return node, nil +} + func inferLinkedHostsForProxmoxNodes(nodes []models.Node, hostByID map[string]*models.Host) map[string]*models.Host { if len(nodes) == 0 || len(hostByID) == 0 { return nil diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index e68a22f221..a74cc9774d 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -93,7 +93,9 @@ type ResourceRegistry struct { matcher *IdentityMatcher store ResourceStore links []ResourceLink - exclusions map[string]struct{} + exclusions map[string]time.Time // exclusionKey -> when the operator last split the pair + linksByID map[string][]ResourceLink + splitAgents map[string]nodeAgentLinkSide // node source ID -> the linked agent the operator split it from identityPins *identityPinIndex pbsBackups []models.PBSBackup // linkFoldIndex maps each canonical ID a manual link folded into another @@ -233,7 +235,7 @@ func NewRegistryWithStaleThresholds(store ResourceStore, thresholds map[DataSour bySource: make(map[DataSource]map[string]string), matcher: NewIdentityMatcher(), store: store, - exclusions: make(map[string]struct{}), + exclusions: make(map[string]time.Time), canonicalMetadataDirty: true, staleThresholds: cloneStaleThresholds(thresholds), } @@ -268,9 +270,12 @@ func (rr *ResourceRegistry) loadOverrides() { // An unlink or link landing between the two reads would load both // decisions for its pair (manual_link_decisions.go). rr.links, exclusions = effectiveManualPairDecisions(rr.links, exclusions) + rr.linksByID = indexLinksByID(rr.links) for _, exclusion := range exclusions { key := exclusionKey(exclusion.ResourceA, exclusion.ResourceB) - rr.exclusions[key] = struct{}{} + if at, ok := rr.exclusions[key]; !ok || exclusion.CreatedAt.After(at) { + rr.exclusions[key] = exclusion.CreatedAt + } } } else { log.Printf("unifiedresources: failed to load manual exclusions from store: %v", err) @@ -325,7 +330,7 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh } } } - rr.ingestProxmoxNode(node, inferredLinkedHostByNodeID[strings.TrimSpace(node.ID)]) + rr.ingestProxmoxNode(rr.splitProxmoxNodeLink(node, inferredLinkedHostByNodeID[strings.TrimSpace(node.ID)])) } for _, host := range snapshot.Hosts { rr.ingestHost(host) @@ -3451,12 +3456,17 @@ func (rr *ResourceRegistry) resolveLinkedResource(source DataSource, sourceID st return "" } linkedNodeID := strings.TrimSpace(existing.Agent.LinkedNodeID) - if linkedNodeID == sourceID { - return id + if linkedNodeID != sourceID && + (linkedNodeID != "" || !identitiesShareHostname(existing.Identity, resource.Identity)) { + return "" } - if linkedNodeID == "" && identitiesShareHostname(existing.Identity, resource.Identity) { - return id + if rr.proxmoxNodeAgentSplitLocked( + nodeAgentLinkSide{sourceID: sourceID, identity: resource.Identity}, + nodeAgentLinkSide{sourceID: strings.TrimSpace(resource.Proxmox.LinkedAgentID), identity: existing.Identity, heldID: id}, + ) { + return "" } + return id } } case SourceAgent: @@ -3470,6 +3480,12 @@ func (rr *ResourceRegistry) resolveLinkedResource(source DataSource, sourceID st if linkedHostID == "" || linkedHostID != sourceID { return "" } + if rr.proxmoxNodeAgentSplitLocked( + nodeAgentLinkSide{sourceID: strings.TrimSpace(resource.Agent.LinkedNodeID), identity: existing.Identity, heldID: id}, + nodeAgentLinkSide{sourceID: sourceID, identity: resource.Identity}, + ) { + return "" + } return id } } @@ -3908,7 +3924,8 @@ func (rr *ResourceRegistry) findCorroboratedOneSidedProxmoxLink( } matchID := "" - for _, resourceID := range rr.bySource[SourceProxmox] { + var matchNodeIDs []string + for nodeID, resourceID := range rr.bySource[SourceProxmox] { existing := rr.resources[resourceID] if existing == nil || existing.Proxmox == nil { continue @@ -3923,6 +3940,15 @@ func (rr *ResourceRegistry) findCorroboratedOneSidedProxmoxLink( return "" } matchID = resourceID + matchNodeIDs = append(matchNodeIDs, nodeID) + } + for _, nodeID := range matchNodeIDs { + if rr.proxmoxNodeAgentSplitLocked( + nodeAgentLinkSide{sourceID: nodeID, identity: rr.resources[matchID].Identity, heldID: matchID}, + nodeAgentLinkSide{sourceID: hostAgentID, identity: identity}, + ) { + return "" + } } return matchID @@ -4958,6 +4984,23 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura otherID = link.ResourceA } other := rr.resources[otherID] + if other == nil && otherID != primaryID && rr.joinedPairOwnsIDLocked(primary, otherID) { + // A relink of a split node and its agent names the two rows, + // and the declared link folds one of them into the other's ID + // once the relink rejoins them. Treat the ID as folded by this + // link so pin succession keeps the link rather than re-keying + // it onto the joined ID, which would let the split decide again. + // Current succession and API report-merge use holder-bound fold + // records, not the old registry-wide merged-ID set. The missing + // side was already joined by the declared link; record only its + // source and ID here, never invented telemetry or identity keys. + foldedSource := SourceProxmox + if otherID == rr.sourceSpecificID(ResourceTypeAgent, SourceAgent, normalizeSourceID(primary.Agent.AgentID)) { + foldedSource = SourceAgent + } + recordManualLinkFold(primary, primaryID, &Resource{Sources: []DataSource{foldedSource}}, otherID) + rr.indexLinkFoldsLocked(primary) + } if other == nil || otherID == primaryID { continue } @@ -4967,6 +5010,12 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura if isAvailabilityOwnedResource(*primary) || isAvailabilityOwnedResource(*other) { continue } + // A link between a Proxmox node and an agent loses to a newer split + // of the two read across their IDs: report-merge cannot delete a + // link naming the rows an earlier split left (nodeAgentSplit). + if rr.nodeAgentRowsSplit(rr.operatorPairDecisionsLocked(), primary, other) { + continue + } // A manual link records operator intent to unify identities, but its // direction must not change the semantic resource shape. In particular, diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index 8d2cba07cc..3ca84279b2 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -144,6 +144,391 @@ func onlyResourceOfType(t *testing.T, rr *ResourceRegistry, resourceType Resourc return resources[0] } +// An operator who splits a Proxmox node from the pulse-agent Pulse linked it +// to (the drawer's report-merge, or unlink naming the node's side) must see +// the split on every monitor rebuild and in the resources API. The +// node<->agent link merged the pair in resolveLinkedResource before +// findMatch, and the linked node had already taken the agent's machine +// identity, so the pair stayed merged although the request answered 200. +// Each step is judged over three rebuilds of a store-backed adapter (pins +// persist on one rebuild and steer the next) and in every view: the +// monitor's listing, the broadcast coalesce, and the resources API's +// registry seeded from that listing or from the snapshot. +func TestOperatorSplitOverridesProxmoxNodeAgentLink(t *testing.T) { + now := time.Now().UTC() + for _, link := range []struct { + name string + nodeLinked bool + agentLinked bool + cluster string + agentHostname string + machineID string + }{ + {"declared-both-ways", true, true, "", "pve1", "0123456789abcdef"}, + // A node-side link the agent does not declare joins only when the + // two share a hostname (findCorroboratedOneSidedProxmoxLink). + {"node-side-only", true, false, "", "pve1", "0123456789abcdef"}, + {"agent-side-only", false, true, "", "pve1", "0123456789abcdef"}, + // A clustered node's own ID is cluster-derived, not one the + // candidate exclusions name, and its identity pin is found by + // cluster and hostname, carrying the agent's machine key. + {"clustered-node", true, true, "lab", "pve1", "0123456789abcdef"}, + {"clustered-agent-side-only", false, true, "lab", "pve1", "0123456789abcdef"}, + // The joined row's first hostname is the agent's when the agent + // is ingested first. + {"clustered-agent-fqdn", true, true, "lab", "pve1.example", "0123456789abcdef"}, + // An agent reporting no machine key joins under the node's own ID. + {"clustered-machine-keyless-agent", true, true, "lab", "pve1", ""}, + } { + for _, request := range []struct { + name string + // exclusions returns the pairs the request records. + exclusions func(ids splitRequestIDs) [][2]string + split func(ids splitRequestIDs) bool + }{ + {"report-merge", func(ids splitRequestIDs) [][2]string { + return [][2]string{{ids.merged, ids.nodeCandidate}, {ids.merged, ids.agentCandidate}} + }, splitAlways}, + {"unlink-naming-node", func(ids splitRequestIDs) [][2]string { + return [][2]string{{ids.merged, ids.nodeCandidate}} + }, splitAlways}, + // The ID the node row holds once split (cluster-derived for a + // clustered node), as an unlink of the two rows records it. + {"unlink-naming-node-row", func(ids splitRequestIDs) [][2]string { + return [][2]string{{ids.nodeOwn, ids.agentOwn}} + }, func(ids splitRequestIDs) bool { return ids.nodeOwn != ids.agentOwn }}, + // Neither row's ID: a relink of the rows replaces no exclusion + // row, so the newer decision across the pair's IDs must decide. + {"unlink-naming-both-candidates", func(ids splitRequestIDs) [][2]string { + return [][2]string{{ids.agentCandidate, ids.nodeCandidate}} + }, splitAlways}, + // The agent's candidate against the agent's machine-derived ID, + // under which the pair merged, is also what report-merge records + // when it splits another source (a Docker host, say) off the + // agent, so it does not split the node from the agent. Merged + // under the node's own ID, it splits the agent off the node. + {"exclusion-naming-only-the-agent", func(ids splitRequestIDs) [][2]string { + return [][2]string{{ids.merged, ids.agentCandidate}} + }, func(ids splitRequestIDs) bool { return ids.merged == ids.nodeOwn }}, + {"unrelated-exclusion", func(ids splitRequestIDs) [][2]string { + return [][2]string{{ids.merged, "agent-00000000deadbeef"}} + }, func(splitRequestIDs) bool { return false }}, + } { + t.Run(link.name+"/"+request.name, func(t *testing.T) { + store, err := NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + + node := models.Node{ID: "lab-pve1", Name: "pve1", Instance: "lab", ClusterName: link.cluster, Host: "https://10.0.0.5:8006", Status: "online", LastSeen: now} + host := models.Host{ + ID: "host-pve1", + Hostname: link.agentHostname, + MachineID: link.machineID, + Status: "online", + LastSeen: now, + // The node's endpoint address and a MAC: what a correct + // link shares, and enough for identity matching to merge + // the pair again if the node kept the agent's identity. + NetworkInterfaces: []models.HostNetworkInterface{{Name: "vmbr0", MAC: "aa:bb:cc:dd:ee:01", Addresses: []string{"10.0.0.5/24"}}}, + } + if link.nodeLinked { + node.LinkedAgentID = host.ID + } + if link.agentLinked { + host.LinkedNodeID = node.ID + } + snapshot := models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}, Hosts: []models.Host{host}} + // The IDs each side holds on its own, with nothing to link to. + alone := NewRegistry(nil) + alone.IngestSnapshot(models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}}) + nodeAlone := hostRowsByFacet(t, alone.List()).node + alone = NewRegistry(nil) + alone.IngestSnapshot(models.StateSnapshot{LastUpdate: now, Hosts: []models.Host{host}}) + agentAlone := hostRowsByFacet(t, alone.List()).agent + adapter := NewMonitorAdapter(NewRegistry(store)) + + adapter.PopulateFromSnapshot(snapshot) + merged := hostRowsByFacet(t, adapter.GetAll()) + if merged.joined == "" || nodeAlone == "" || agentAlone == "" { + t.Fatalf("fixture did not link the node and the agent: %+v (alone: node %q, agent %q)", merged, nodeAlone, agentAlone) + } + reportMerge := func(mergedID string) (nodeCandidate, agentCandidate string) { + t.Helper() + for _, target := range adapter.currentRegistry().SourceTargets(mergedID) { + switch target.Source { + case SourceProxmox: + nodeCandidate = target.CandidateID + case SourceAgent: + agentCandidate = target.CandidateID + } + } + if nodeCandidate == "" || agentCandidate == "" || nodeCandidate == mergedID || agentCandidate == mergedID { + t.Fatalf("merged %s lists candidates node=%q agent=%q", mergedID, nodeCandidate, agentCandidate) + } + return nodeCandidate, agentCandidate + } + nodeCandidate, agentCandidate := reportMerge(merged.joined) + ids := splitRequestIDs{merged: merged.joined, nodeCandidate: nodeCandidate, agentCandidate: agentCandidate, nodeOwn: nodeAlone, agentOwn: agentAlone} + for _, pair := range request.exclusions(ids) { + if err := store.AddExclusion(ResourceExclusion{ResourceA: pair[0], ResourceB: pair[1], CreatedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + } + + // Split, each side returns to the ID it holds alone (for an + // agent reporting a machine key, the ID the pair merged + // under); otherwise the pair stays merged under its ID. + split := request.split(ids) + want := nodeAgentRows{joined: merged.joined} + if split { + want = nodeAgentRows{node: nodeAlone, agent: agentAlone} + } + assertViews := func(step string, rebuilds int, want nodeAgentRows) { + t.Helper() + for rebuild := 1; rebuild <= rebuilds; rebuild++ { + adapter.PopulateFromSnapshot(snapshot) + listed, _, thresholds := adapter.GetAllWithMetricsTargetsAndStaleThresholds() + broadcast, ok := adapter.CoalesceForPresentation(listed, thresholds) + if !ok { + t.Fatal("store-backed adapter did not coalesce with its exclusions") + } + rest := NewRegistry(store) + rest.IngestResources(listed) + restFromSnapshot := NewRegistry(store) + restFromSnapshot.IngestSnapshot(snapshot) + for view, resources := range map[string][]Resource{ + "monitor": listed, + "broadcast": broadcast, + "rest": rest.List(), + "rest presentation": rest.ListForPresentation(), + "rest from snapshot": restFromSnapshot.List(), + "rest presentation snapshot": restFromSnapshot.ListForPresentation(), + } { + if got := hostRowsByFacet(t, resources); got != want { + t.Fatalf("%s rebuild %d %s: got %+v, want %+v", step, rebuild, view, got, want) + } + } + } + } + // Snapshots ingest nodes first. A registry that meets the agent + // first, which already holds its machine-derived ID, splits the + // same way, also when the snapshot then lacks the host: before + // any split rebuild rewrites the pins, the node's pin from the + // joined era alone carries the agent's machine key. + if split { + withoutHost := snapshot + withoutHost.Hosts = nil + for name, nodes := range map[string]models.StateSnapshot{"with host": snapshot, "without host": withoutHost} { + agentFirst := NewRegistry(store) + agentFirst.IngestRecords(SourceAgent, []IngestRecord{HostIngestRecord(host)}) + agentFirst.IngestSnapshot(nodes) + for view, resources := range map[string][]Resource{"list": agentFirst.List(), "presentation": agentFirst.ListForPresentation()} { + if got := hostRowsByFacet(t, resources); got != want { + t.Fatalf("agent ingested first, snapshot %s, %s: got %+v, want %+v", name, view, got, want) + } + } + } + } + assertViews("after the request", 3, want) + if !split { + return + } + // The split node names the agent only where the monitor's own + // node record does: guest actions, deploy and service + // discovery act on a node's linked agent. + for _, resource := range adapter.GetAll() { + if resource.ID == want.node && resource.Proxmox.LinkedAgentID != node.LinkedAgentID { + t.Fatalf("split node links agent %q, its record links %q", resource.Proxmox.LinkedAgentID, node.LinkedAgentID) + } + } + + // Linking the rows the split left joins the pair again, although + // report-merge recorded exclusions against other IDs. + if err := store.AddLink(ResourceLink{ResourceA: want.agent, ResourceB: want.node, PrimaryID: want.agent, CreatedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + assertViews("after relink", 2, nodeAgentRows{joined: merged.joined}) + // The inferred join can consume the node row before manual + // links run. Its fold must still reach current succession and + // the API seed, rather than an obsolete registry-wide ID set. + seeded := NewRegistry(store) + seeded.IngestResources(adapter.GetAll()) + for name, registry := range map[string]*ResourceRegistry{"monitor": adapter.currentRegistry(), "API seed": seeded} { + found := false + for _, fold := range registry.ManualLinkFolds(merged.joined) { + nodeFold := fold.HolderID == want.agent && fold.FoldedID == want.node && slices.Contains(fold.Sources, SourceProxmox) + agentFold := fold.HolderID == want.node && fold.FoldedID == want.agent && slices.Contains(fold.Sources, SourceAgent) + // A machine-keyless agent rejoins under the node's ID, + // so the same pair is folded in the opposite direction. + if nodeFold || agentFold { + found = true + } + } + if !found { + t.Fatalf("%s lost the relink's pair fold: %+v", name, registry.ManualLinkFolds(merged.joined)) + } + } + // A rebuild that meets the agent first orders the joined row's + // hostnames the agent's way; persisting its pins must not + // re-key the relink onto the joined ID either. + reordered := NewRegistry(store) + reordered.IngestRecords(SourceAgent, []IngestRecord{HostIngestRecord(host)}) + reordered.IngestSnapshot(snapshot) + reordered.PersistIdentityPins() + assertViews("after an agent-first rebuild persisted its pins", 2, nodeAgentRows{joined: merged.joined}) + + // The same request after that relink splits them again. Where + // it names a pair other than the relinked rows, the store keeps + // the older link, which must lose to the newer split. + nodeCandidate, agentCandidate = reportMerge(merged.joined) + ids = splitRequestIDs{merged: merged.joined, nodeCandidate: nodeCandidate, agentCandidate: agentCandidate, nodeOwn: nodeAlone, agentOwn: agentAlone} + for _, pair := range request.exclusions(ids) { + if err := store.AddExclusion(ResourceExclusion{ResourceA: pair[0], ResourceB: pair[1], CreatedAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } + } + assertViews("after splitting again", 2, want) + }) + } + } +} + +// A node split from one agent keeps another agent's pin. The node links +// agent A, whose joined-era pin carries A's machine key for the node's +// cluster and hostname; while A is missing from a rebuild, agent B declares +// the node, and the operator has split the node from B. The pin is A's, so +// the node still completes A's key and keeps the ID it holds with A, as any +// boot window before A checks in does. +func TestOperatorSplitFromOneAgentKeepsAnotherAgentsPin(t *testing.T) { + now := time.Now().UTC() + store, err := NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + node := models.Node{ID: "lab-pve1", Name: "pve1", Instance: "lab", ClusterName: "lab", Host: "https://10.0.0.5:8006", Status: "online", LastSeen: now, LinkedAgentID: "host-a"} + agentA := models.Host{ID: "host-a", Hostname: "pve1", MachineID: "aaaaaaaaaaaaaaaa", LinkedNodeID: node.ID, Status: "online", LastSeen: now} + agentB := models.Host{ID: "host-b", Hostname: "pve1", MachineID: "bbbbbbbbbbbbbbbb", LinkedNodeID: node.ID, Status: "online", LastSeen: now} + + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateFromSnapshot(models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}, Hosts: []models.Host{agentA, agentB}}) + var joinedWithA string + for _, resource := range adapter.GetAll() { + if resource.Proxmox != nil && resource.Agent != nil && resource.Agent.AgentID == agentA.ID { + joinedWithA = resource.ID + } + } + if joinedWithA == "" { + t.Fatalf("fixture did not join the node with agent A: %+v", adapter.GetAll()) + } + splitFromB := ResourceExclusion{ + ResourceA: SourceSpecificID(ResourceTypeAgent, SourceProxmox, node.ID), + ResourceB: SourceSpecificID(ResourceTypeAgent, SourceAgent, agentB.ID), + CreatedAt: time.Now().UTC(), + } + if err := store.AddExclusion(splitFromB); err != nil { + t.Fatal(err) + } + + withoutA := models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}, Hosts: []models.Host{agentB}} + rebuilt := NewRegistry(store) + rebuilt.IngestSnapshot(withoutA) + for _, resource := range rebuilt.List() { + if resource.Proxmox == nil { + continue + } + if resource.ID != joinedWithA { + t.Fatalf("node took %s while agent A was missing, want %s from A's pin", resource.ID, joinedWithA) + } + if resource.Agent != nil && resource.Agent.AgentID == agentB.ID { + t.Fatalf("node joined agent B, which the operator split it from") + } + } +} + +// An agent that stops reporting its machine ID stays split from the node. +// The joined-era pin carries the agent's old machine key for the clustered +// node's cluster and hostname, and the node record names no agent (only the +// agent declares the node), so nothing but the split drop ties the pin to +// the agent: its keyless report must not count as another agent's. +func TestOperatorSplitHoldsWhenTheAgentStopsReportingItsMachineID(t *testing.T) { + now := time.Now().UTC() + store, err := NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + node := models.Node{ID: "lab-pve1", Name: "pve1", Instance: "lab", ClusterName: "lab", Host: "https://10.0.0.5:8006", Status: "online", LastSeen: now} + agent := models.Host{ID: "host-pve1", Hostname: "pve1", MachineID: "0123456789abcdef", LinkedNodeID: node.ID, Status: "online", LastSeen: now} + + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateFromSnapshot(models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}, Hosts: []models.Host{agent}}) + merged := hostRowsByFacet(t, adapter.GetAll()) + if merged.joined == "" { + t.Fatalf("fixture did not join the node and the agent: %+v", merged) + } + alone := NewRegistry(nil) + alone.IngestSnapshot(models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}}) + nodeAlone := hostRowsByFacet(t, alone.List()).node + if err := store.AddExclusion(ResourceExclusion{ + ResourceA: SourceSpecificID(ResourceTypeAgent, SourceProxmox, node.ID), + ResourceB: SourceSpecificID(ResourceTypeAgent, SourceAgent, agent.ID), + CreatedAt: time.Now().UTC(), + }); err != nil { + t.Fatal(err) + } + + agent.MachineID = "" + rebuilt := NewRegistry(store) + rebuilt.IngestSnapshot(models.StateSnapshot{LastUpdate: now, Nodes: []models.Node{node}, Hosts: []models.Host{agent}}) + got := hostRowsByFacet(t, rebuilt.List()) + if got.joined != "" || got.node != nodeAlone || got.agent == "" { + t.Fatalf("keyless agent rebuild: got %+v, want the node apart on %s", got, nodeAlone) + } +} + +// splitRequestIDs are the IDs an unlink or report-merge of a joined node and +// agent can name: the merged row, its SourceTargets candidates, and the IDs +// the two rows hold once split. +type splitRequestIDs struct { + merged, nodeCandidate, agentCandidate, nodeOwn, agentOwn string +} + +func splitAlways(splitRequestIDs) bool { return true } + +type nodeAgentRows struct { + node string // a row with the Proxmox facet only + agent string // a row with the agent facet only + joined string // a row with both +} + +func hostRowsByFacet(t *testing.T, resources []Resource) nodeAgentRows { + t.Helper() + var rows nodeAgentRows + set := func(slot *string, id string) { + if *slot != "" { + t.Fatalf("two host rows of one kind: %s and %s", *slot, id) + } + *slot = id + } + for _, resource := range resources { + if resource.Type != ResourceTypeAgent { + continue + } + switch { + case resource.Proxmox != nil && resource.Agent != nil: + set(&rows.joined, resource.ID) + case resource.Proxmox != nil: + set(&rows.node, resource.ID) + case resource.Agent != nil: + set(&rows.agent, resource.ID) + } + } + return rows +} + // A host agent past its reporting lease is offline. Its source sighting is // stale too, and the stale pass used to rank that above offline, so the // machine reached the frontend as a warning with its last report rendered diff --git a/internal/unifiedresources/registry_test.go b/internal/unifiedresources/registry_test.go index 185a04c1b5..8fcdf65ac0 100644 --- a/internal/unifiedresources/registry_test.go +++ b/internal/unifiedresources/registry_test.go @@ -2310,6 +2310,69 @@ func TestResourceRegistry_ManualNodeMergeRewritesProxmoxGuestParentAndActionAgen } } +// TestRegistryWithManualLinksFoldsLikeTheStoreBackedRegistry pins the +// store-less registry the mock-mode unified view seeds: given the links an +// adapter generation loaded, it folds already-unified resources exactly as a +// registry backed by the link store does. The adapter reports the links of its +// current generation and picks up a new one on its next rebuild. +func TestRegistryWithManualLinksFoldsLikeTheStoreBackedRegistry(t *testing.T) { + now := time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) + resources := []Resource{ + { + ID: "vm-checkout", + Type: ResourceTypeVM, + Name: "checkout-web-01", + Status: StatusOnline, + LastSeen: now, + Sources: []DataSource{SourceProxmox}, + Proxmox: &ProxmoxData{SourceID: "pve:node1:101", NodeName: "node1", Instance: "pve", VMID: 101}, + }, + { + ID: "agent-apollo", + Type: ResourceTypeAgent, + Name: "apollo-114", + Status: StatusOnline, + LastSeen: now, + Sources: []DataSource{SourceAgent}, + Agent: &AgentData{AgentID: "agent-apollo", Hostname: "apollo-114"}, + }, + } + link := ResourceLink{ResourceA: "vm-checkout", ResourceB: "agent-apollo", PrimaryID: "vm-checkout"} + + store := NewMemoryStore() + adapter := NewMonitorAdapter(NewRegistry(store)) + if err := store.AddLink(link); err != nil { + t.Fatalf("AddLink: %v", err) + } + if got := adapter.ManualLinks(); len(got) != 0 { + t.Fatalf("ManualLinks before a rebuild = %+v, want the generation's empty set", got) + } + adapter.PopulateFromSnapshot(models.StateSnapshot{}) + links := adapter.ManualLinks() + if len(links) != 1 || links[0].ResourceA != link.ResourceA || links[0].ResourceB != link.ResourceB || links[0].PrimaryID != link.PrimaryID { + t.Fatalf("ManualLinks after a rebuild = %+v, want %+v", links, link) + } + links[0].PrimaryID = "agent-apollo" + if got := adapter.ManualLinks(); got[0].PrimaryID != "vm-checkout" { + t.Fatalf("ManualLinks handed out the registry's own slice: %+v", got) + } + links[0].PrimaryID = "vm-checkout" + + storeBacked := NewRegistry(store) + storeBacked.IngestResources(resources) + storeLess := NewRegistryWithManualLinks(links) + // The registry keeps its own copy of the caller's links. + links[0].ResourceB = "agent-other" + storeLess.IngestResources(resources) + + for name, rr := range map[string]*ResourceRegistry{"store-backed": storeBacked, "store-less": storeLess} { + got := rr.List() + if len(got) != 1 || got[0].ID != "vm-checkout" || got[0].Agent == nil || got[0].Proxmox == nil { + t.Fatalf("%s registry listed %v, want the VM carrying the linked agent", name, resourceIDs(got)) + } + } +} + func TestResourceRegistry_ProxmoxGuestParentPrefersAgentBackedNodeDuplicate(t *testing.T) { rr := NewRegistry(nil) now := time.Date(2026, 5, 14, 10, 0, 0, 0, time.UTC) diff --git a/internal/unifiedresources/registry_views_branchcov0722pm_test.go b/internal/unifiedresources/registry_views_branchcov0722pm_test.go index e54b00e62b..2f874e37e9 100644 --- a/internal/unifiedresources/registry_views_branchcov0722pm_test.go +++ b/internal/unifiedresources/registry_views_branchcov0722pm_test.go @@ -203,7 +203,7 @@ func TestBranchcov0722PM(t *testing.T) { } // Register the exclusion between the two IDs and re-run; the // closure's lookup must now return true and block the merge. - rr.exclusions[exclusionKey("agent-proxmox-pve1", "agent-runtime-pve1")] = struct{}{} + rr.exclusions[exclusionKey("agent-proxmox-pve1", "agent-runtime-pve1")] = time.Now() got := rr.ListForPresentation() if len(got) != 2 { t.Fatalf("expected exclusion to keep both host views split, got %d: %+v", len(got), got) From d525ef6583edfc3eaaf3f6611cc51db7636b0a8d Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:03:46 +0100 Subject: [PATCH 2/3] Resolve TrueNAS Patrol disk heat through its alert policy Reconcile the absent runtime portion of PR2693 aaf539848099c35f5eb62dd02961d56e3d36f842 on current assigned source. TrueNAS disk heat follows its disk override, TrueNAS-wide default and inherited per-type policy rather than the system synthetic agent ID. Preserve agent-host overrides, Off, clear thresholds, alert switches, retained-reading limits and canonical identity. Include discriminating resolver, adapter, Patrol and Assistant tests; leave the frontend stack to Web. Change-source: pulse-maintainer --- .../v6/internal/subsystems/ai-runtime.md | 21 +- .../v6/internal/subsystems/alerts.md | 24 ++ .../internal/subsystems/unified-resources.md | 13 + internal/ai/alert_threshold_adapter.go | 20 ++ internal/ai/alert_threshold_adapter_test.go | 33 +++ internal/ai/mock_test.go | 6 + internal/ai/patrol.go | 6 + internal/ai/patrol_ai.go | 2 +- internal/ai/patrol_disk_temperature.go | 52 +++- internal/ai/patrol_disk_temperature_test.go | 145 +++++++++++ internal/ai/patrol_thresholds_test.go | 3 + internal/ai/resource_context.go | 2 +- .../threshold_resolution_shared_test.go | 237 ++++++++++++++++++ internal/alerts/truenas_disk_temperature.go | 42 ++++ internal/api/ai_handlers_concurrency_test.go | 3 + internal/api/ai_handlers_more_test.go | 3 + 16 files changed, 602 insertions(+), 10 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index fa473ad85e..beb9334f0b 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -382,6 +382,17 @@ linked node or guest) replaces the per-type trigger, and an override or agent default that switches the host's alerts off leaves no heat to judge. A disk no agent reports gets the per-type policy. The global alerts switch and the agent alert-type switch stop `CheckHost` without changing this policy, as before. +A TrueNAS disk (`alerts.IsTrueNASDiskResource`, the disks the unified evaluator +alerts on as `truenas-disk`) skips that walk, because its TrueNAS system carries +a synthetic `Agent.AgentID` that would pass for a host agent: +`physicalDiskTemperatureLimits` resolves it through +`ThresholdProvider.GetTrueNASDiskTemperatureThreshold(resourceID, diskType)` +(the alert manager's `TrueNASDiskTemperatureThreshold`, the `truenas-disk` +tiers of `effectiveAlertPolicyNoLock`), so the disk's own override, then the +TrueNAS-wide TrueNAS Disks value, then the per-type policy decide, as for its +temperature alert, and a disabled override or TrueNAS Disks default leaves no +heat. A host override under the system's synthetic ID no longer reaches its +disks, and Disable all TrueNAS silences the alert without changing this policy. The run state carries the provider (`patrolRuntimeState.thresholdProvider`). With no provider the factory alert configuration applies. A disk is hot from its trigger: a triage warning, a disk issue and an AI chat "needing attention" entry, as its temperature alert and @@ -392,9 +403,13 @@ verification unknown rather than recovered. Scoped runs keep the provider, and a provider set before Patrol starts is handed to it. Below the trigger Patrol flags nothing, so an NVMe at 63C stays quiet and a SATA disk at 56C is flagged. Proofs: `internal/ai/patrol_disk_temperature_test.go` (a real registry with one -overridden agent in `TestPatrolJudgesAgentDisksByTheAgentDiskTemperatureOverride`), -`TestAlertThresholdAdapter_DiskTemperatureFollowsAlertPolicy` and -`TestAlertThresholdAdapter_DiskTemperatureHonoursHostOverrides` in +overridden agent in `TestPatrolJudgesAgentDisksByTheAgentDiskTemperatureOverride`, +and TrueNAS fixture disks under a per-disk override, a TrueNAS-wide value and a +host override on the system ID, one with a retained reading, in +`TestPatrolJudgesTrueNASDisksByTheirAlertTiers`), +`TestAlertThresholdAdapter_DiskTemperatureFollowsAlertPolicy`, +`TestAlertThresholdAdapter_DiskTemperatureHonoursHostOverrides` and +`TestAlertThresholdAdapter_TrueNASDiskTemperatureFollowsItsAlertTiers` in `internal/ai/alert_threshold_adapter_test.go`. Retained summaries disclose that point and bucket timestamps describe returned diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 94480c7ae9..2c498bb9ac 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -4475,3 +4475,27 @@ resubmission and concurrent response encoding/update under the race detector. Existing snapshot, persistence, partial-update and threshold tests remain required. This is source ownership acceptance, not installed notification or guest-agent recovery evidence. + +### TrueNAS disk heat policy for Patrol and Assistant + +`Manager.TrueNASDiskTemperatureThreshold(resourceID, diskType)` resolves the +`truenas-disk` tiers of `effectiveAlertPolicyNoLock`: the disk's canonical or +identity-resolved override, then TrueNAS Disks defaults, then disk-type policy. +A disabled disk/default or non-positive trigger leaves no heat policy. +Global/platform alert switches silence alert evaluation without changing this +policy, matching existing agent-disk semantics. Callers receive a copy; config +normalisation, saved Off/custom/inherited values and alert severity are unchanged. +`IsTrueNASDiskResource` uses the evaluator's classification, including merged +agent/TrueNAS disks. Patrol/Assistant do not walk those disks to a synthetic +TrueNAS system agent and accidentally apply an unrelated host override. +Retained temperature readings remain unjudged, not affirmative recovery. + +Verification: `TestTrueNASDiskTemperatureThresholdMatchesTrueNASDiskAlerts` +and `TestIsTrueNASDiskResourceMatchesTheUnifiedEvaluator` in +`internal/alerts/threshold_resolution_shared_test.go` cover tier, identity, +Off/disabled, untyped, clone and nil-manager compatibility. +`TestAlertThresholdAdapter_TrueNASDiskTemperatureFollowsItsAlertTiers` and +`TestPatrolJudgesTrueNASDisksByTheirAlertTiers` cover adapter, Patrol triage, +verification and Assistant context while preserving agent-host overrides. +Frontend table/drawer tiers remain a separate Web-owned change; this backend +repair does not establish browser or native appliance acceptance. diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 20b764f7fa..698bd28cc0 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -7499,3 +7499,16 @@ for report-merge. No synthetic telemetry, identity key, enrollment or service link is added. `TestOperatorSplitOverridesProxmoxNodeAgentLink` checks that the monitor and its re-ingested API seed retain the fold across all split shapes, then survive pin persistence and a repeated split. + +### TrueNAS disk policy consumers use the disk's alert identity + +The AI/Patrol physical-disk consumers classify TrueNAS disks through +`alerts.IsTrueNASDiskResource`, the same classifier as unified alert evaluation. +They resolve temperature policy under the disk's canonical resource ID, not its +parent system's synthetic agent ID. Merged agent/TrueNAS disks retain that +classification; ordinary agent disks still follow their reporting host's policy. +Registry identity, source precedence and retained-reading semantics are unchanged. +`TestIsTrueNASDiskResourceMatchesTheUnifiedEvaluator` and +`TestPatrolJudgesTrueNASDisksByTheirAlertTiers` exercise the unified registry +projection and its consumer boundaries. Table/drawer frontend acceptance remains +with Web, not established by these backend tests. diff --git a/internal/ai/alert_threshold_adapter.go b/internal/ai/alert_threshold_adapter.go index 53163e2f3b..143144ea12 100644 --- a/internal/ai/alert_threshold_adapter.go +++ b/internal/ai/alert_threshold_adapter.go @@ -79,6 +79,26 @@ func (a *AlertThresholdAdapter) GetDiskTemperatureThreshold(host alerts.DiskTemp } else { threshold = alerts.DefaultDiskTemperatureThreshold(diskType) } + return diskTemperatureTriggerAndClear(threshold) +} + +// GetTrueNASDiskTemperatureThreshold returns the temperature alert trigger and +// clear values for one TrueNAS disk, resolved exactly as its temperature alert +// resolves them (alerts.Manager.TrueNASDiskTemperatureThreshold): the disk's +// own override, then the TrueNAS-wide value, then the per-type policy. +func (a *AlertThresholdAdapter) GetTrueNASDiskTemperatureThreshold(resourceID, diskType string) (float64, float64) { + var threshold *alerts.HysteresisThreshold + if a.manager != nil { + threshold = a.manager.TrueNASDiskTemperatureThreshold(resourceID, diskType) + } else { + threshold = alerts.DefaultDiskTemperatureThreshold(diskType) + } + return diskTemperatureTriggerAndClear(threshold) +} + +// diskTemperatureTriggerAndClear unpacks a disk temperature threshold, with a +// nil or non-positive trigger meaning off (0, 0). +func diskTemperatureTriggerAndClear(threshold *alerts.HysteresisThreshold) (float64, float64) { if threshold == nil || threshold.Trigger <= 0 { return 0, 0 } diff --git a/internal/ai/alert_threshold_adapter_test.go b/internal/ai/alert_threshold_adapter_test.go index 67347911f6..dd231267d2 100644 --- a/internal/ai/alert_threshold_adapter_test.go +++ b/internal/ai/alert_threshold_adapter_test.go @@ -108,6 +108,39 @@ func TestAlertThresholdAdapter_DiskTemperatureFollowsAlertPolicy(t *testing.T) { } } +// A TrueNAS disk reaches Patrol through its own alert tiers: its override, +// then the TrueNAS-wide value, then the per-type policy. +func TestAlertThresholdAdapter_TrueNASDiskTemperatureFollowsItsAlertTiers(t *testing.T) { + if trigger, clear := NewAlertThresholdAdapter(nil).GetTrueNASDiskTemperatureThreshold("disk-any", "nvme"); trigger != 70 || clear != 65 { + t.Fatalf("no manager: nvme = %v/%v, want the factory 70/65", trigger, clear) + } + + mgr := alerts.NewManager() + t.Cleanup(mgr.Stop) + a := NewAlertThresholdAdapter(mgr) + if trigger, clear := a.GetTrueNASDiskTemperatureThreshold("disk-plain", "sata"); trigger != 55 || clear != 50 { + t.Fatalf("no TrueNAS-wide value: sata = %v/%v, want the per-type 55/50", trigger, clear) + } + + cfg := mgr.GetConfig() + cfg.TrueNASDiskDefaults.Temperature = &alerts.HysteresisThreshold{Trigger: 62, Clear: 57} + cfg.Overrides = map[string]alerts.ThresholdConfig{ + "disk-raised": {Temperature: &alerts.HysteresisThreshold{Trigger: 75, Clear: 70}}, + "disk-off": {Disabled: true}, + } + mgr.UpdateConfig(cfg) + for id, want := range map[string][2]float64{ + "disk-raised": {75, 70}, + "disk-plain": {62, 57}, + "disk-off": {0, 0}, + } { + trigger, clear := a.GetTrueNASDiskTemperatureThreshold(id, "nvme") + if trigger != want[0] || clear != want[1] { + t.Errorf("%s nvme = %v/%v, want %v/%v", id, trigger, clear, want[0], want[1]) + } + } +} + // A host's Disk Temp override reaches Patrol for that host's disks only, and a // host whose alerts are switched off has no disk heat policy at all. func TestAlertThresholdAdapter_DiskTemperatureHonoursHostOverrides(t *testing.T) { diff --git a/internal/ai/mock_test.go b/internal/ai/mock_test.go index 6fb53b6a55..6672cb39f5 100644 --- a/internal/ai/mock_test.go +++ b/internal/ai/mock_test.go @@ -73,6 +73,12 @@ func (m *mockThresholdProvider) GetDiskTemperatureThreshold(_ alerts.DiskTempera return 0, 0 } +// GetTrueNASDiskTemperatureThreshold judges a TrueNAS disk by the same +// per-type table, as a TrueNAS disk with no override or TrueNAS-wide value is. +func (m *mockThresholdProvider) GetTrueNASDiskTemperatureThreshold(_ string, diskType string) (float64, float64) { + return m.GetDiskTemperatureThreshold(alerts.DiskTemperatureHost{}, diskType) +} + type mockUnifiedResourceProvider struct { UnifiedResourceProvider getAllFunc func() []unifiedresources.Resource diff --git a/internal/ai/patrol.go b/internal/ai/patrol.go index d349e60417..b8f07600ee 100644 --- a/internal/ai/patrol.go +++ b/internal/ai/patrol.go @@ -71,6 +71,12 @@ type ThresholdProvider interface { // agent reports. A host with no ID means no agent reports the disk. A zero // trigger means disk temperature alerting is off for it. GetDiskTemperatureThreshold(host alerts.DiskTemperatureHost, diskType string) (trigger, clear float64) + // GetTrueNASDiskTemperatureThreshold returns the disk temperature alert + // trigger and clear values (°C) for one TrueNAS disk, given its resource + // ID and type: the disk's own override, then the TrueNAS-wide value, then + // the per-type policy, as its temperature alert resolves them. A zero + // trigger means disk temperature alerting is off for it. + GetTrueNASDiskTemperatureThreshold(resourceID, diskType string) (trigger, clear float64) } // AlertResolver provides the ability to review and resolve alerts diff --git a/internal/ai/patrol_ai.go b/internal/ai/patrol_ai.go index e8462ae8f8..5b62edfe3b 100644 --- a/internal/ai/patrol_ai.go +++ b/internal/ai/patrol_ai.go @@ -2484,7 +2484,7 @@ func patrolPhysicalDiskRows(snap patrolRuntimeState, scopedSet map[string]bool) temperature: tools.SplitDiskTemperature(r.PhysicalDisk.Temperature, r.PhysicalDisk.Collection), smartEvidence: unifiedPhysicalDiskSMARTIssueParts(r.PhysicalDisk.SMART), } - row.temperatureLimits = snap.diskTemperatureLimits(physicalDiskTemperatureHost(r, owners), row.diskType) + row.temperatureLimits = physicalDiskTemperatureLimits(snap.thresholdProvider, r, owners) rows = append(rows, row) } return rows diff --git a/internal/ai/patrol_disk_temperature.go b/internal/ai/patrol_disk_temperature.go index 1af32bb0f4..d5857d6782 100644 --- a/internal/ai/patrol_disk_temperature.go +++ b/internal/ai/patrol_disk_temperature.go @@ -22,12 +22,53 @@ type diskTemperatureLimits struct { // that host's Disk Temp override applies, or from the factory alert // configuration when Patrol has no threshold provider. func diskTemperatureLimitsFor(provider ThresholdProvider, host alerts.DiskTemperatureHost, diskType string) diskTemperatureLimits { - var trigger, clear float64 if provider != nil { - trigger, clear = provider.GetDiskTemperatureThreshold(host, diskType) - } else if threshold := alerts.DefaultDiskTemperatureThreshold(diskType); threshold != nil { - trigger, clear = threshold.Trigger, threshold.Clear + return newDiskTemperatureLimits(provider.GetDiskTemperatureThreshold(host, diskType)) } + return factoryDiskTemperatureLimits(diskType) +} + +// trueNASDiskTemperatureLimitsFor resolves the policy for one TrueNAS disk the +// way its temperature alert does, from the user's alert configuration, so the +// disk's own override and the TrueNAS-wide value apply, or from the factory +// alert configuration when Patrol has no threshold provider. +func trueNASDiskTemperatureLimitsFor(provider ThresholdProvider, resourceID, diskType string) diskTemperatureLimits { + if provider != nil { + return newDiskTemperatureLimits(provider.GetTrueNASDiskTemperatureThreshold(resourceID, diskType)) + } + return factoryDiskTemperatureLimits(diskType) +} + +// physicalDiskTemperatureLimits resolves the policy for one physical disk +// resource under the alert that judges its heat. A TrueNAS disk has its own +// temperature alert, judged by the disk's override, the TrueNAS-wide value and +// the per-type policy; its TrueNAS system carries a synthetic agent ID, so the +// host walk would wrongly treat that system as a host agent. Any other disk is +// judged as a SMART disk of the host agent that reports it. +func physicalDiskTemperatureLimits(provider ThresholdProvider, disk unifiedresources.Resource, owners map[string]unifiedresources.Resource) diskTemperatureLimits { + diskType := "" + if disk.PhysicalDisk != nil { + diskType = disk.PhysicalDisk.DiskType + } + if alerts.IsTrueNASDiskResource(disk) { + return trueNASDiskTemperatureLimitsFor(provider, disk.ID, diskType) + } + return diskTemperatureLimitsFor(provider, physicalDiskTemperatureHost(disk, owners), diskType) +} + +// factoryDiskTemperatureLimits is the policy for a disk of the given type +// under the factory alert configuration. +func factoryDiskTemperatureLimits(diskType string) diskTemperatureLimits { + if threshold := alerts.DefaultDiskTemperatureThreshold(diskType); threshold != nil { + return newDiskTemperatureLimits(threshold.Trigger, threshold.Clear) + } + return diskTemperatureLimits{} +} + +// newDiskTemperatureLimits builds limits from an alert trigger and clear +// value. A non-positive trigger is off; a clear value missing or above the +// trigger leaves no band below it. +func newDiskTemperatureLimits(trigger, clear float64) diskTemperatureLimits { if trigger <= 0 { return diskTemperatureLimits{} } @@ -72,7 +113,8 @@ func physicalDiskOwnerIndex(urp UnifiedResourceProvider) map[string]unifiedresou // disk, the host its disk temperature alerts are evaluated under: the agent on // the machine the disk is parented to or, for a disk parented to a storage // pool (Unraid array and cache disks), the agent on the pool's machine. A disk -// no agent reports gets the hostless policy. +// no agent reports gets the hostless policy. TrueNAS disks do not come here; +// physicalDiskTemperatureLimits judges them by their own alert. func physicalDiskTemperatureHost(disk unifiedresources.Resource, owners map[string]unifiedresources.Resource) alerts.DiskTemperatureHost { parentID := disk.ParentID for hops := 0; parentID != nil && hops < 2; hops++ { diff --git a/internal/ai/patrol_disk_temperature_test.go b/internal/ai/patrol_disk_temperature_test.go index dc3260378f..8516dc7d89 100644 --- a/internal/ai/patrol_disk_temperature_test.go +++ b/internal/ai/patrol_disk_temperature_test.go @@ -8,6 +8,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/truenas" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/aicontracts" "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" @@ -343,3 +344,147 @@ func TestPhysicalDiskTemperatureHostFollowsTheParentChain(t *testing.T) { t.Errorf("parentless disk: host = %+v, want hostless", got) } } + +// trueNASHeatRegistry is a TrueNAS system whose pool holds four disks beside +// the two overridden host agents of overriddenAgentDisksRegistry: nvme0n1 at +// 72C, nvme1n1 at 63C, SATA sda at 60C and SATA sdb, whose 80C reading is +// retained from before standby. The user saved a TrueNAS-wide 62C (clear 57), +// raised nvme0n1's own trigger to 75C (clear 70), and left a 50C Disk Temp +// override under the TrueNAS system's synthetic agent ID. It returns the disk +// resources by name. +func trueNASHeatRegistry(t *testing.T) (*unifiedresources.ResourceRegistry, *AlertThresholdAdapter, map[string]unifiedresources.Resource) { + t.Helper() + registry, _ := overriddenAgentDisksRegistry(t) + now := time.Now() + disk := func(name, transport string, temperature int) truenas.Disk { + return truenas.Disk{ + ID: "disk-" + name, Name: name, Pool: "tank", Status: "ONLINE", Model: "Model " + name, + Serial: "SERIAL-" + strings.ToUpper(name), Temperature: temperature, Transport: transport, + } + } + records := truenas.FixtureRecords(truenas.FixtureSnapshot{ + CollectedAt: now, + System: truenas.SystemInfo{Hostname: "truenas-main", Healthy: true, CollectedAt: now}, + Pools: []truenas.Pool{{ID: "pool-tank", Name: "tank", Status: "ONLINE"}}, + Disks: []truenas.Disk{ + disk("nvme0n1", "nvme", 72), disk("nvme1n1", "nvme", 63), + disk("sda", "sata", 60), disk("sdb", "sata", 80), + }, + }) + for i := range records { + if meta := records[i].Resource.PhysicalDisk; meta != nil && records[i].Resource.Name == "sdb" { + meta.Collection = &diskinventory.CollectionStatus{Temperature: diskinventory.Unavailable("truenas", "disk is in standby")} + } + } + registry.IngestRecords(unifiedresources.SourceTrueNAS, records) + + adapter := unifiedresources.NewUnifiedAIAdapter(registry) + disks := make(map[string]unifiedresources.Resource) + for _, r := range adapter.GetByType(unifiedresources.ResourceTypePhysicalDisk) { + if alerts.IsTrueNASDiskResource(r) { + disks[r.Name] = r + } + } + systemAgentID := "" + for _, r := range adapter.GetAll() { + if r.Agent != nil && r.Agent.Platform == "truenas" { + systemAgentID = r.Agent.AgentID + } + } + if len(disks) != 4 || systemAgentID == "" { + t.Fatalf("TrueNAS disks = %v, system agent ID = %q; want four disks under one system", disks, systemAgentID) + } + + mgr := alerts.NewManager() + t.Cleanup(mgr.Stop) + cfg := mgr.GetConfig() + cfg.TrueNASDiskDefaults.Temperature = &alerts.HysteresisThreshold{Trigger: 62, Clear: 57} + cfg.Overrides = map[string]alerts.ThresholdConfig{ + "agent-cool": {DiskTemperature: &alerts.HysteresisThreshold{Trigger: 80, Clear: 75}}, + systemAgentID: {DiskTemperature: &alerts.HysteresisThreshold{Trigger: 50, Clear: 45}}, + disks["nvme0n1"].ID: {Temperature: &alerts.HysteresisThreshold{Trigger: 75, Clear: 70}}, + } + mgr.UpdateConfig(cfg) + return registry, NewAlertThresholdAdapter(mgr), disks +} + +// Patrol judges a TrueNAS disk by its own temperature alert's tiers, not as a +// disk of the TrueNAS system's synthetic host agent: nvme0n1's override keeps +// it quiet at 72C, the TrueNAS-wide 62C flags nvme1n1 at 63C (under the NVMe +// trigger of 70C) and spares SATA sda at 60C (over the SATA trigger of 55C), +// and the 50C host override under the system ID reaches none of them. A +// retained reading stays unjudged. Agent disks keep their host's override. +func TestPatrolJudgesTrueNASDisksByTheirAlertTiers(t *testing.T) { + registry, provider, disks := trueNASHeatRegistry(t) + state := newPatrolRuntimeStateWithProviders(models.StateSnapshot{}, nil, unifiedresources.NewUnifiedAIAdapter(registry)) + state.thresholdProvider = provider + + type want struct { + limits diskTemperatureLimits + issue bool + } + expected := map[string]want{ + disks["nvme0n1"].ID: {limits: diskTemperatureLimits{trigger: 75, clear: 70}}, + disks["nvme1n1"].ID: {limits: diskTemperatureLimits{trigger: 62, clear: 57}, issue: true}, + disks["sda"].ID: {limits: diskTemperatureLimits{trigger: 62, clear: 57}}, + disks["sdb"].ID: {limits: diskTemperatureLimits{trigger: 62, clear: 57}}, + } + rows := patrolPhysicalDiskRows(state, nil) + if len(rows) != 6 { + t.Fatalf("rows = %+v, want four TrueNAS disks and two agent disks", rows) + } + for _, row := range rows { + w, trueNAS := expected[row.id] + if !trueNAS { + // overriddenAgentDisksRegistry: agent-cool's 76C NVMe under its 80C + // override, agent-plain's 72C NVMe over the NVMe trigger. + w = want{limits: diskTemperatureLimits{trigger: 70, clear: 65}, issue: true} + if row.temperature.Collected == 76 { + w = want{limits: diskTemperatureLimits{trigger: 80, clear: 75}} + } + } + if row.temperatureLimits != w.limits { + t.Errorf("disk %s at %dC: limits = %+v, want %+v", row.id, row.temperature.Collected, row.temperatureLimits, w.limits) + } + if got := patrolPhysicalDiskHealthIssue(row); got != w.issue { + t.Errorf("disk %s at %dC: health issue = %v, want %v", row.id, row.temperature.Collected, got, w.issue) + } + } + + flags := triageDiskHealthChecksState(state, nil) + for _, name := range []string{"nvme0n1", "sda", "sdb"} { + if flag := triageFindFlag(flags, func(f TriageFlag) bool { + return f.ResourceID == disks[name].ID && strings.Contains(f.Reason, "Disk temperature") + }); flag != nil { + t.Errorf("%s: unexpected temperature flag %+v", name, *flag) + } + } + if flag := triageFindFlag(flags, func(f TriageFlag) bool { + return f.ResourceID == disks["nvme1n1"].ID && strings.Contains(f.Reason, "Disk temperature") + }); flag == nil || flag.Threshold != 62 { + t.Errorf("nvme1n1: temperature flag = %+v, want one at the TrueNAS-wide 62C", flag) + } + + if _, err := verifyMetricRecoveredState(state, PatrolThresholds{}, "disk-high", disks["sdb"].ID, "physical_disk"); !errors.Is(err, aicontracts.ErrVerificationUnknown) { + t.Errorf("retained 80C: err = %v, want ErrVerificationUnknown", err) + } + + s := &Service{} + s.SetUnifiedResourceProvider(unifiedresources.NewUnifiedAIAdapter(registry)) + s.SetPatrolThresholdProvider(provider) + _, section, _ := strings.Cut(s.buildUnifiedResourceContext(), "Physical Disks Needing Attention") + section, _, _ = strings.Cut(section, "\n\n") + for _, reading := range []string{"Temp: 63C", "Temp: 72C"} { + if !strings.Contains(section, reading) { + t.Errorf("attention section = %q, want the disk at %s", section, reading) + } + } + for _, reading := range []string{"Temp: 60C", "Temp: 76C", "Temp: 80C"} { + if strings.Contains(section, reading) { + t.Errorf("attention section = %q, lists a disk at %s", section, reading) + } + } + if strings.Count(section, "Temp: 72C") != 1 { + t.Errorf("attention section = %q, want only agent-plain's 72C disk, not TrueNAS nvme0n1 under its 75C override", section) + } +} diff --git a/internal/ai/patrol_thresholds_test.go b/internal/ai/patrol_thresholds_test.go index 883795f24d..71145845ea 100644 --- a/internal/ai/patrol_thresholds_test.go +++ b/internal/ai/patrol_thresholds_test.go @@ -24,6 +24,9 @@ func (m MockThresholdProvider) GetStorageThreshold() float64 { return m.Stor func (m MockThresholdProvider) GetDiskTemperatureThreshold(alerts.DiskTemperatureHost, string) (float64, float64) { return 0, 0 } +func (m MockThresholdProvider) GetTrueNASDiskTemperatureThreshold(string, string) (float64, float64) { + return 0, 0 +} func TestCalculatePatrolThresholds_Default(t *testing.T) { // Test default behavior (exact mode) diff --git a/internal/ai/resource_context.go b/internal/ai/resource_context.go index 5d7778688e..0f6df0c27b 100644 --- a/internal/ai/resource_context.go +++ b/internal/ai/resource_context.go @@ -417,7 +417,7 @@ func (s *Service) buildUnifiedResourceContextForModel(destinationModel string) s health = strings.ToUpper(strings.TrimSpace(disk.PhysicalDisk.Health)) // A retained last-known reading is not current heat. collected := tools.SplitDiskTemperature(disk.PhysicalDisk.Temperature, disk.PhysicalDisk.Collection).Collected - hot = diskTemperatureLimitsFor(thresholdProvider, physicalDiskTemperatureHost(disk, owners), disk.PhysicalDisk.DiskType).hot(collected) + hot = physicalDiskTemperatureLimits(thresholdProvider, disk, owners).hot(collected) } if disk.Status != unifiedresources.StatusOnline || (health != "" && health != "PASSED" && health != "UNKNOWN") || hot { attention = append(attention, disk) diff --git a/internal/alerts/threshold_resolution_shared_test.go b/internal/alerts/threshold_resolution_shared_test.go index 77f6dfd1d2..8d246c3c5f 100644 --- a/internal/alerts/threshold_resolution_shared_test.go +++ b/internal/alerts/threshold_resolution_shared_test.go @@ -12,6 +12,7 @@ import ( alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/recovery" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) func boolPtr(v bool) *bool { @@ -2135,3 +2136,239 @@ func TestHostDiskTemperatureThresholdMatchesCheckHostOverrides(t *testing.T) { t.Fatalf("nil manager host policy = %+v, want the factory nvme 70", got) } } + +// TrueNASDiskTemperatureThreshold is the disk heat policy for a TrueNAS disk, +// so it must agree with the disk's temperature alert on every tier: the +// disk's own override, then the TrueNAS-wide value, then the per-type policy. +// A host override under the TrueNAS system's synthetic agent ID is not one of +// them. Disable all TrueNAS silences the alert without changing the policy, +// as the agent switches do for agent disks. +func TestTrueNASDiskTemperatureThresholdMatchesTrueNASDiskAlerts(t *testing.T) { + disk := trueNASTemperatureDisk("nvme0n1", "nvme", 0) + + const canonicalDiskID = "physical-disk:canonical-nvme0n1" + for _, tc := range []struct { + name string + mutate func(*AlertConfig) + setup func(*Manager) + want *HysteresisThreshold + off bool + silenced bool + }{ + {name: "per-type policy", want: &HysteresisThreshold{Trigger: 70, Clear: 65}}, + { + name: "raised per-type trigger", + mutate: func(cfg *AlertConfig) { + cfg.DiskTempByType = map[string]HysteresisThreshold{"nvme": {Trigger: 75, Clear: 70}} + }, + want: &HysteresisThreshold{Trigger: 75, Clear: 70}, + }, + { + name: "TrueNAS-wide value", + mutate: func(cfg *AlertConfig) { + cfg.TrueNASDiskDefaults.Temperature = &HysteresisThreshold{Trigger: 62, Clear: 57} + }, + want: &HysteresisThreshold{Trigger: 62, Clear: 57}, + }, + { + name: "per-disk override beats the TrueNAS-wide value", + mutate: func(cfg *AlertConfig) { + cfg.TrueNASDiskDefaults.Temperature = &HysteresisThreshold{Trigger: 62, Clear: 57} + cfg.Overrides = map[string]ThresholdConfig{disk.ID: {Temperature: &HysteresisThreshold{Trigger: 75, Clear: 70}}} + }, + want: &HysteresisThreshold{Trigger: 75, Clear: 70}, + }, + { + name: "per-disk override without a clear value", + mutate: func(cfg *AlertConfig) { + cfg.Overrides = map[string]ThresholdConfig{disk.ID: {Temperature: &HysteresisThreshold{Trigger: 75}}} + }, + want: &HysteresisThreshold{Trigger: 75, Clear: 70}, + }, + { + name: "per-disk override under the canonical identity", + mutate: func(cfg *AlertConfig) { + cfg.Overrides = map[string]ThresholdConfig{canonicalDiskID: {Temperature: &HysteresisThreshold{Trigger: 78, Clear: 73}}} + }, + setup: func(m *Manager) { + m.SetResourceIntentIdentityResolver(func(resourceID string) (string, bool) { + return canonicalDiskID, resourceID == disk.ID + }) + }, + want: &HysteresisThreshold{Trigger: 78, Clear: 73}, + }, + { + name: "host override under the TrueNAS system ID does not reach its disks", + mutate: func(cfg *AlertConfig) { + cfg.Overrides = map[string]ThresholdConfig{"truenas-main": {DiskTemperature: &HysteresisThreshold{Trigger: 50, Clear: 45}}} + }, + want: &HysteresisThreshold{Trigger: 70, Clear: 65}, + }, + { + name: "per-disk override switches the disk off", + mutate: func(cfg *AlertConfig) { + cfg.Overrides = map[string]ThresholdConfig{disk.ID: {Temperature: &HysteresisThreshold{}}} + }, + off: true, + }, + { + name: "per-disk override disables the disk's alerts", + mutate: func(cfg *AlertConfig) { + cfg.Overrides = map[string]ThresholdConfig{disk.ID: {Disabled: true}} + }, + }, + { + name: "TrueNAS-wide off", + mutate: func(cfg *AlertConfig) { + cfg.TrueNASDiskDefaults.Temperature = &HysteresisThreshold{} + }, + off: true, + }, + { + name: "TrueNAS Disks defaults disabled", + mutate: func(cfg *AlertConfig) { + cfg.TrueNASDiskDefaults.Disabled = true + }, + }, + { + name: "agent Disk Temp default off", + mutate: func(cfg *AlertConfig) { + cfg.AgentDefaults.DiskTemperature = &HysteresisThreshold{} + }, + off: true, + }, + { + name: "Disable all TrueNAS keeps the policy", + mutate: func(cfg *AlertConfig) { + cfg.DisableAllTrueNAS = true + }, + want: &HysteresisThreshold{Trigger: 70, Clear: 65}, + silenced: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + m := newTestManager(t) + cfg := unifiedEvalBaseConfig() + cfg.TrueNASDiskDefaults = ThresholdConfig{} + if tc.mutate != nil { + tc.mutate(&cfg) + } + configureUnifiedEvalManager(t, m, cfg) + if tc.setup != nil { + tc.setup(m) + } + + got := m.TrueNASDiskTemperatureThreshold(" "+disk.ID+" ", " NVMe") + switch { + case tc.off: + if got == nil || got.Trigger > 0 { + t.Fatalf("TrueNASDiskTemperatureThreshold = %+v, want a switched-off threshold", got) + } + case (got == nil) != (tc.want == nil) || (got != nil && *got != *tc.want): + t.Fatalf("TrueNASDiskTemperatureThreshold = %+v, want %+v", got, tc.want) + } + + var alert *Alert + fires := func(temperature int) bool { + reading := disk + meta := *disk.PhysicalDisk + meta.Temperature = temperature + reading.PhysicalDisk = &meta + checkTrueNASTemperatureDisk(t, m, reading) + var firing bool + alert, firing = trueNASDiskTemperatureAlert(t, m, reading) + return firing + } + if tc.silenced { + if fires(99) { + t.Fatalf("alert fired at 99C with TrueNAS alerts disabled, active: %v", alertKeys(m)) + } + return + } + if got == nil || got.Trigger <= 0 { + if fires(99) { + t.Fatalf("alert fired at 99C on a disk the policy judges no heat for, active: %v", alertKeys(m)) + } + return + } + trigger := int(got.Trigger) + if fires(trigger - 1) { + t.Fatalf("alert fired at %dC, under the policy trigger %v", trigger-1, got.Trigger) + } + if !fires(trigger) { + t.Fatalf("alert stayed quiet at the policy trigger %dC, active: %v", trigger, alertKeys(m)) + } + // The alert recovers at the clear value Patrol reads. + if alert.Threshold != got.Trigger || alert.Metadata["clearThreshold"] != got.Clear { + t.Fatalf("alert trigger/clear = %v/%v, want the policy %v/%v", alert.Threshold, alert.Metadata["clearThreshold"], got.Trigger, got.Clear) + } + }) + } + + // A disk whose type TrueNAS never reported follows the agent Disk Temp + // default, as its alert does. + t.Run("untyped disk", func(t *testing.T) { + m := newTestManager(t) + cfg := unifiedEvalBaseConfig() + cfg.TrueNASDiskDefaults = ThresholdConfig{} + cfg.AgentDefaults.DiskTemperature = &HysteresisThreshold{Trigger: 58, Clear: 53} + configureUnifiedEvalManager(t, m, cfg) + untypedDisk := trueNASTemperatureDisk("sdx", "", 58) + if got := m.TrueNASDiskTemperatureThreshold(untypedDisk.ID, ""); got == nil || got.Trigger != 58 || got.Clear != 53 { + t.Fatalf("untyped disk = %+v, want the agent default 58/53", got) + } + checkTrueNASTemperatureDisk(t, m, untypedDisk) + if _, firing := trueNASDiskTemperatureAlert(t, m, untypedDisk); !firing { + t.Fatalf("untyped disk at the agent default 58C raised no alert, active: %v", alertKeys(m)) + } + }) + + m := configureDiskTempTypeHostManager(t) + m.TrueNASDiskTemperatureThreshold(disk.ID, "nvme").Trigger = 1 + if got := m.TrueNASDiskTemperatureThreshold(disk.ID, "nvme"); got == nil || got.Trigger != 70 { + t.Fatalf("caller mutation reached the config: %+v", got) + } + var nilManager *Manager + if got := nilManager.TrueNASDiskTemperatureThreshold(disk.ID, "sata"); got == nil || got.Trigger != 55 { + t.Fatalf("nil manager TrueNAS disk policy = %+v, want the factory sata 55", got) + } +} + +// IsTrueNASDiskResource names the disks the unified evaluator judges as +// TrueNAS disks, the ones TrueNASDiskTemperatureThreshold resolves. +func TestIsTrueNASDiskResourceMatchesTheUnifiedEvaluator(t *testing.T) { + agentDisk := unifiedresources.Resource{ + ID: "physical-disk:host-1/nvme0n1", Type: unifiedresources.ResourceTypePhysicalDisk, + Sources: []unifiedresources.DataSource{unifiedresources.SourceAgent}, + PhysicalDisk: &unifiedresources.PhysicalDiskMeta{DiskType: "nvme"}, + } + trueNASFacetDisk := agentDisk + trueNASFacetDisk.TrueNAS = &unifiedresources.TrueNASData{} + mergedDisk := agentDisk + mergedDisk.Sources = []unifiedresources.DataSource{unifiedresources.SourceAgent, unifiedresources.SourceTrueNAS} + noMetaDisk := trueNASTemperatureDisk("sdy", "sata", 40) + noMetaDisk.PhysicalDisk = nil + trueNASSystem := unifiedresources.Resource{ + ID: "agent:truenas-main", Type: unifiedresources.ResourceTypeAgent, + Sources: []unifiedresources.DataSource{unifiedresources.SourceTrueNAS}, + } + for name, tc := range map[string]struct { + resource unifiedresources.Resource + want bool + }{ + "TrueNAS-sourced disk": {resource: trueNASTemperatureDisk("sda", "sata", 40), want: true}, + "TrueNAS facet disk": {resource: trueNASFacetDisk, want: true}, + "agent and TrueNAS": {resource: mergedDisk, want: true}, + "no disk metadata": {resource: noMetaDisk, want: true}, + "agent disk": {resource: agentDisk}, + "TrueNAS system": {resource: trueNASSystem}, + } { + if got := IsTrueNASDiskResource(tc.resource); got != tc.want { + t.Errorf("%s: IsTrueNASDiskResource = %v, want %v", name, got, tc.want) + } + input, ok := UnifiedResourceInputFromResource(tc.resource) + if evaluated := ok && input.Type == "truenas-disk"; evaluated != tc.want { + t.Errorf("%s: unified evaluator judges it as a TrueNAS disk = %v, want %v", name, evaluated, tc.want) + } + } +} diff --git a/internal/alerts/truenas_disk_temperature.go b/internal/alerts/truenas_disk_temperature.go index 9934e83233..d7fca0f69a 100644 --- a/internal/alerts/truenas_disk_temperature.go +++ b/internal/alerts/truenas_disk_temperature.go @@ -1,5 +1,11 @@ package alerts +import ( + "strings" + + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + // trueNASDiskTemperatureDefaultNoLock returns the default tier of a TrueNAS // disk's temperature threshold. A TrueNAS-wide value the user saved under // TrueNAS Disks applies to every TrueNAS disk, as a host Disk Temp override @@ -16,3 +22,39 @@ func (m *Manager) trueNASDiskTemperatureDefaultNoLock(diskType string, typeUnkno } return cloneThreshold(m.hostDiskTemperatureThresholdNoLock(m.config.AgentDefaults.DiskTemperature, false, diskType)) } + +// IsTrueNASDiskResource reports whether the unified evaluator judges a +// resource as a TrueNAS disk (truenas-disk), whose temperature alert resolves +// its threshold through TrueNASDiskTemperatureThreshold rather than through +// the host agent that reports it. +func IsTrueNASDiskResource(resource unifiedresources.Resource) bool { + typeKey, ok := unifiedAlertResourceType(resource) + return ok && typeKey == "truenas-disk" +} + +// TrueNASDiskTemperatureThreshold is the disk heat policy for one TrueNAS +// disk, resolved by the same effectiveAlertPolicyNoLock tiers its temperature +// alert uses: an override stored under the disk's resource ID, then the +// TrueNAS-wide TrueNAS Disks value, then the per-type policy. An override or +// TrueNAS Disks default that switches the disk's alerts off leaves no +// threshold (nil); a non-positive trigger is off too. Like +// HostDiskTemperatureThreshold it ignores the global and per-platform alert +// switches, which silence the alert without changing the policy. A nil +// manager resolves the factory configuration, where no TrueNAS-wide value is +// set. +func (m *Manager) TrueNASDiskTemperatureThreshold(resourceID, diskType string) *HysteresisThreshold { + if m == nil { + return DefaultDiskTemperatureThreshold(diskType) + } + m.mu.RLock() + defer m.mu.RUnlock() + thresholds := m.effectiveAlertPolicyNoLock(alertPolicyQuery{ + TypeKey: "truenas-disk", + ResourceID: unifiedresources.CanonicalResourceID(resourceID), + DiskType: strings.ToLower(strings.TrimSpace(diskType)), + }).Thresholds + if thresholds.Disabled { + return nil + } + return cloneThreshold(thresholds.Temperature) +} diff --git a/internal/api/ai_handlers_concurrency_test.go b/internal/api/ai_handlers_concurrency_test.go index 0c5276fa05..db1b4cad27 100644 --- a/internal/api/ai_handlers_concurrency_test.go +++ b/internal/api/ai_handlers_concurrency_test.go @@ -71,6 +71,9 @@ func (concurrencyThresholdProvider) GetStorageThreshold() float64 { return 92 func (concurrencyThresholdProvider) GetDiskTemperatureThreshold(alerts.DiskTemperatureHost, string) (float64, float64) { return 70, 65 } +func (concurrencyThresholdProvider) GetTrueNASDiskTemperatureThreshold(string, string) (float64, float64) { + return 70, 65 +} type concurrencyMetricsHistoryProvider struct{} diff --git a/internal/api/ai_handlers_more_test.go b/internal/api/ai_handlers_more_test.go index fb0397dca8..8f67b0d184 100644 --- a/internal/api/ai_handlers_more_test.go +++ b/internal/api/ai_handlers_more_test.go @@ -162,6 +162,9 @@ func (stubThresholdProvider) GetStorageThreshold() float64 { return 92 } func (stubThresholdProvider) GetDiskTemperatureThreshold(alerts.DiskTemperatureHost, string) (float64, float64) { return 70, 65 } +func (stubThresholdProvider) GetTrueNASDiskTemperatureThreshold(string, string) (float64, float64) { + return 70, 65 +} type stubMetricsHistoryProvider struct{} From 0f368fe04753d969583aada069d99c591d96ac88 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Thu, 8 Oct 2026 11:33:19 +0100 Subject: [PATCH 3/3] Check Docker inventory before repeated-name reports can restore it Extend the existing sequential three-surface identity fixture to an FQDN connection with identical short guest names and only the home agent. Assert initial guest association, safe ambiguity clearing, unchanged raw Docker inventory and stable parent/child identity in API-list and broadcast projections before any fresh Docker report. This closes an acceptance gap; it does not establish native recovery for issue 2681. Contract-Neutral: Tighten regression assertions for the existing cross-installation identity and independent Docker ingest boundaries; no runtime, API or operator-policy delta. Change-source: pulse-maintainer --- ...onitor_cross_installation_identity_test.go | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) diff --git a/internal/monitoring/monitor_cross_installation_identity_test.go b/internal/monitoring/monitor_cross_installation_identity_test.go index 2294e5579b..5b84c0a9c5 100644 --- a/internal/monitoring/monitor_cross_installation_identity_test.go +++ b/internal/monitoring/monitor_cross_installation_identity_test.go @@ -1,6 +1,7 @@ package monitoring import ( + "reflect" "testing" "time" @@ -27,6 +28,8 @@ func TestCrossInstallationIdentitySurvivesStandaloneAddition(t *testing.T) { {"classified member-hostname coincidence", "https://pmx1:8006", true, false}, {"unclassified FQDN connection and distinct guest domains", "https://pmx1.remote.example:8006", false, true}, {"classified FQDN connection and distinct guest domains", "https://pmx1.remote.example:8006", true, true}, + {"unclassified FQDN connection and identical short guest names", "https://pmx1.remote.example:8006", false, false}, + {"classified FQDN connection and identical short guest names", "https://pmx1.remote.example:8006", true, false}, } { name, classified := tc.name, tc.classified clusterName := "" @@ -80,6 +83,45 @@ func TestCrossInstallationIdentitySurvivesStandaloneAddition(t *testing.T) { t.Fatal("initial Docker report absent from read state") } homeCanonicalID := adapter.VMs()[0].ID() + initialHosts := m.state.GetSnapshot().Hosts + if len(initialHosts) != 1 || initialHosts[0].LinkedVMID != homeGuestID { + t.Fatalf("initial automatic guest link = %+v, want %s", initialHosts, homeGuestID) + } + initialDocker := m.state.GetSnapshot().DockerHosts + initialDockerHostID := adapter.DockerHosts()[0].ID() + initialWorkloadID := adapter.DockerContainers()[0].ID() + assertUnchangedDocker := func(stage string) { + t.Helper() + // No new Docker report has arrived: a report after link clearing + // could restore deleted inventory and hide the actual regression. + if got := m.state.GetSnapshot().DockerHosts; !reflect.DeepEqual(got, initialDocker) { + t.Fatalf("%s changed the independent Docker snapshot: got %+v, want %+v", stage, got, initialDocker) + } + if hosts, workloads := adapter.DockerHosts(), adapter.DockerContainers(); len(hosts) != 1 || len(workloads) != 1 || + hosts[0].ID() != initialDockerHostID || workloads[0].ID() != initialWorkloadID { + t.Fatalf("%s lost Docker source identities: hosts=%+v workloads=%+v", stage, hosts, workloads) + } + listed, _, thresholds := adapter.GetAllWithMetricsTargetsAndStaleThresholds() + broadcast, ok := adapter.CoalesceForPresentation(listed, thresholds) + if !ok { + t.Fatal("store-backed Docker presentation was not available") + } + for surface, resources := range map[string][]unifiedresources.Resource{"API list": listed, "broadcast": broadcast} { + var machineID string + var workloads []unifiedresources.Resource + for _, resource := range resources { + if resource.Type == unifiedresources.ResourceTypeAgent && resource.Docker != nil && resource.Docker.HostSourceID == initialDocker[0].ID { + machineID = resource.ID + } + if resource.Type == unifiedresources.ResourceTypeAppContainer && resource.Docker != nil && resource.Docker.ContainerID == "home-workload" { + workloads = append(workloads, resource) + } + } + if machineID == "" || len(workloads) != 1 || workloads[0].ID != initialWorkloadID || workloads[0].ParentID == nil || *workloads[0].ParentID != machineID { + t.Fatalf("%s %s detached Docker workload from its reporting machine: machine=%s workloads=%+v", stage, surface, machineID, workloads) + } + } + } standalone := config.PVEInstance{Name: "Hetzner pmx1", Host: tc.host} m.config.PVEInstances = append(m.config.PVEInstances, standalone) @@ -100,9 +142,19 @@ func TestCrossInstallationIdentitySurvivesStandaloneAddition(t *testing.T) { // not discard the agent, its Docker inventory, or the original VM's data. hostReport.Timestamp = now.Add(time.Second) dockerReport.Timestamp = now.Add(time.Second) + assertUnchangedDocker("provider addition without agent reports") if _, err := m.ApplyHostReport(hostReport, token); err != nil { t.Fatal(err) } + hosts := m.state.GetSnapshot().Hosts + wantLink := "" + if tc.fqdnGuests { + wantLink = homeGuestID + } + if len(hosts) != 1 || hosts[0].LinkedVMID != wantLink || hosts[0].LinkedNodeID != "" || hosts[0].LinkedContainerID != "" { + t.Fatalf("post-addition guest link = %+v, want %q with no node/container guess", hosts, wantLink) + } + assertUnchangedDocker("host-only report after automatic guest-link reconciliation") if _, err := m.ApplyDockerReport(dockerReport, token); err != nil { t.Fatal(err) }