diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index a399d24dd8..ec0a3957d7 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -9010,3 +9010,12 @@ websocket as REST coalesces it (the filter's limits are recorded under "Unlink replaces the pair's operator link" in the unified-resources contract). Agent registration, enrolment, install, update, removal, report identity and continuity are unchanged. + +### Report-merge splits a linked agent + +`internal/api/resourceapi/resources.go` changed only so report-merge on a +resource an operator linked an agent into (a VM, a node or a Docker host) +replaces that link, splitting the agent back out as unlink does. Agent +registration, enrolment, install, update, removal, report identity and +continuity are unchanged, and an agent's own declared node link still +ignores exclusions. diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index b0b7fb977f..72d213f392 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -11493,3 +11493,17 @@ Router's or the package-global limiters. Limits, windows, tenant selection, revocation and response payloads are unchanged. The connected lifetime and ownership controls are in `router_limiter_lifecycle_test.go`; the integration fixture's cleanup also runs on a failed listener/request assertion. + +### Report-merge undoes operator links + +`POST /api/resources/{id}/report-merge` keeps its request, status codes and +response shape. For a reported source that an operator link brought in, it +now records the exclusion against the link's own pair instead of a candidate +ID derived from the merged resource's type, so the store replaces the link +and the pair splits as after unlink (contract under "Report-merge splits +operator-linked resources" in unified-resources). `exclusions` counts those +pairs with the identity-match exclusions. `400 Resource is not merged` and +`400 No source targets found` now also need a resource with no applied link, +so two linked agents, which share their only source, can be reported. +`TestContract_ResourceReportMergeReplacesOperatorLink` pins the response +shape and the link's removal for a VM and the agent linked into it. diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 4737de1395..000ee8cfd2 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -6925,3 +6925,13 @@ or recovery path is added or moved. hidden routes, so the demo guard on a `DEMO_MODE` instance answers it with `404`. That removes a diagnostic read surface on demo instances only; no storage, retention, backup, migration or recovery path is added or moved. + +### Report-merge link folds + +`internal/unifiedresources/types.go` adds an unexported, in-memory +`linkFolds` record of the operator links a resource absorbed, and +`internal/api/resourceapi/resources.go` makes report-merge exclude those +links' pairs, which deletes the pair's link row in the resource store as an +unlink does. Nothing new is persisted: the record is rebuilt with each +registry generation, and no schema, backup, retention, migration or +recovery path is added or moved. diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index b944cbef74..aed6f0106a 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -6922,10 +6922,9 @@ as SQLite does. Reads keep the newest link per pair, so an older reversed copy cannot win with its primary, and drop exclusions older than it; duplicate exclusions already stored stay, which changes no decision. A registry resolves its `GetLinks` and `GetExclusions` results together in `loadOverrides`, so an unlink committed between the two reads -cannot load both decisions for one generation. Report-merge still records -exclusions against candidate IDs derived from the merged resource's type, -which do not name the folded side of a cross-type manual link, so it does not -undo such a link; unlink does. `TestManualPairDecisionReplacesThePreviousOne` and +cannot load both decisions for one generation. Report-merge undoes a link +the same way (see "Report-merge splits operator-linked resources" below). +`TestManualPairDecisionReplacesThePreviousOne` and `TestManualPairDecisionReadsResolveStoredConflictsByRecency` pin the store rule on SQLite and the in-memory store, `TestUnlinkSplitsHostPairWhosePinTookTheOtherSidesKey` links and unlinks @@ -6939,3 +6938,69 @@ and `TestMonitorAdapterCoalesceForPresentationHonoursExclusions`, `TestBroadcastPresentationCoalesceHonoursMergeExclusions` and `TestBroadcastPresentationCoalescePrefersTheListingReadState` pin broadcast and REST row counts for a split host pair. + +### Report-merge splits operator-linked resources + +`POST /api/resources/{id}/report-merge` records exclusions for the reported +sources. For each reported source mapped to the merged resource it excludes +the merged resource's ID and the candidate ID that source's record takes when +no match is allowed (`ResourceRegistry.SourceTargets`), as before, which +keeps an identity match apart. That candidate is derived from the merged +resource's type, so for a source an operator link brought in it named +neither side of the link: an agent linked into a VM got a `vm-` +candidate, and an agent linked with a node or a Docker host keyed by its +machine ID got a source-specific one. No link row matched, the link stayed +applied, and the API answered "Merge reported" while the drawer's Split +dialog toasted "Resource split applied". Two linked agents share their only +source, so the API refused them as "Resource is not merged"; an applied link +now marks a merge as well (the drawer still offers Split only for a resource +listing two sources). + +`applyManualLinks` now records each link it applies on the merged resource +(`recordManualLinkFold`): the link row's own pair, the resource that took the +other in and the one folded into it, plus the sources the folded side +brought, including anything links had already folded into it, so every link +along a chain keeps its own pair. A pair is recorded once, since record +ingest can recreate a folded side whose source mapping points at a holder of +another type and fold it again, so a repeated pair adds its sources to the +fold already recorded (the holder still carries what the earlier fold +brought). The record is unexported and rides in-memory +clones, so it reaches the resources API's registry, which seeds from the +monitor's already-linked listing and never holds the folded row. +Report-merge excludes the pair of every link whose folded side brought a +reported source category (`ResourceRegistry.ManualLinkFolds`), and the +store's one-decision-per-pair rule deletes that link row, so the pair splits +on every surface from the monitor's next rebuild, as with unlink. The drawer +reports every merged source. Categories are not attributed to individual +contributors: naming one that some folded side carried undoes that link even +where the holder carries it too, and a source that arrived along a chain of +links undoes every link on its way to the merged resource. + +A link joins only the pair its row names. Canonical-ID succession re-keys +link endpoints with `UPDATE OR IGNORE` but moves `primary_id` +unconditionally, so a row whose re-key collided with the successor's own row +kept the retired endpoint and took the successor as its primary. +`applyManualLinks` honoured that primary and merged the successor through a +row no exclusion of the merged pair removes, so an unlink or report-merge of +that pair came back on the next rebuild. A primary that names neither side +of its row now falls back to the row's first side, which leaves such a row +inert while the retired ID is gone. + +`TestResourceReportMergeSplitsOperatorLinkedPair` links and reports through +the API handlers against a monitor adapter on the same SQLite store, rebuilt +three times per step, for VM/agent, node/agent and agent/Docker pairs linked +from either side and for two linked agents. Every two-source case but a node +linked from its agent fails with the previous handler, whose candidate for +that node happened to be the node's own source-specific ID. +`TestResourceReportMergeSplitsPairBehindASuccessionShadowedRow` reports a +pair that a shadowed row also named, +`TestResourceReportMergeSourceFilterSelectsLinks` pins the source filter, +`TestResourceReportMergeSplitsIdentityMergedPair` keeps an identity-merged +agent and Docker host apart through the rebuilds, +`TestManualLinkFoldsNameEachLinkPairAlongAChain` pins a chain's pairs on the +monitor's registry and on one seeded from its listing, and +`TestManualLinkFoldsRecordEachPairOnceAcrossRecordIngests` folds a TrueNAS +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. diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 87b9781699..4dcee90831 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -14428,6 +14428,79 @@ func TestContract_TenantResourcesDoNotFallbackToRawSnapshotSeeding(t *testing.T) } } +// POST /api/resources/{id}/report-merge keeps its response shape, and on a +// pair an operator linked it replaces the link: the store no longer holds it +// and the resource list shows both sides again. +func TestContract_ResourceReportMergeReplacesOperatorLink(t *testing.T) { + now := time.Now().UTC() + h := newActionTestResourceHandlers(t, &config.Config{DataPath: t.TempDir()}) + h.SetStateProvider(resourceStateProvider{snapshot: models.StateSnapshot{ + LastUpdate: now, + VMs: []models.VM{{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now}}, + Hosts: []models.Host{{ID: "host-box", Hostname: "box-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now}}, + }}) + list := func() []unifiedresources.Resource { + t.Helper() + rec := httptest.NewRecorder() + h.HandleListResources(rec, httptest.NewRequest(http.MethodGet, "/api/resources?type=vm,agent", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("/api/resources status = %d, body=%s", rec.Code, rec.Body.String()) + } + var resp ResourcesResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode /api/resources: %v", err) + } + return resp.Data + } + var vmID, agentID string + for _, resource := range list() { + switch resource.Type { + case unifiedresources.ResourceTypeVM: + vmID = resource.ID + case unifiedresources.ResourceTypeAgent: + agentID = resource.ID + } + } + if vmID == "" || agentID == "" { + t.Fatalf("fixture did not list a separate VM and agent: %+v", list()) + } + + linkRec := httptest.NewRecorder() + h.HandleLink(linkRec, httptest.NewRequest(http.MethodPost, "/api/resources/"+vmID+"/link", strings.NewReader(`{"targetId":"`+agentID+`"}`))) + if linkRec.Code != http.StatusOK { + t.Fatalf("link status = %d, body=%s", linkRec.Code, linkRec.Body.String()) + } + if linked := list(); len(linked) != 1 { + t.Fatalf("link listed %d resources, want one", len(linked)) + } + + rec := httptest.NewRecorder() + h.HandleReportMerge(rec, httptest.NewRequest(http.MethodPost, "/api/resources/"+vmID+"/report-merge", strings.NewReader(`{}`))) + if rec.Code != http.StatusOK { + t.Fatalf("report-merge status = %d, body=%s", rec.Code, rec.Body.String()) + } + var resp map[string]any + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode report-merge: %v", err) + } + if len(resp) != 3 || resp["status"] != "ok" || resp["message"] != "Merge reported" { + t.Fatalf("report-merge response = %#v, want status, message and exclusions", resp) + } + if exclusions, ok := resp["exclusions"].(float64); !ok || exclusions < 1 { + t.Fatalf("report-merge exclusions = %#v, want a positive count", resp["exclusions"]) + } + store, err := h.getStore("default") + if err != nil { + t.Fatal(err) + } + if links, err := store.GetLinks(); err != nil || len(links) != 0 { + t.Fatalf("report-merge left links %+v (err %v)", links, err) + } + if split := list(); len(split) != 2 { + t.Fatalf("report-merge listed %d resources, want the VM and the agent apart", len(split)) + } +} + func TestContract_ResourceListPolicyMetadata(t *testing.T) { now := time.Date(2026, 3, 17, 10, 0, 0, 0, time.UTC) h := newActionTestResourceHandlers(t, &config.Config{DataPath: t.TempDir()}) diff --git a/internal/api/resourceapi/resources.go b/internal/api/resourceapi/resources.go index 25365bd6f6..8a0a600690 100644 --- a/internal/api/resourceapi/resources.go +++ b/internal/api/resourceapi/resources.go @@ -964,13 +964,16 @@ func (h *QueryService) HandleReportMerge(w http.ResponseWriter, r *http.Request) return } - if len(resource.Sources) < 2 { + // Two resources an operator linked can share every source (two agents), + // so an applied link marks a merge too. + linkFolds := registry.ManualLinkFolds(path) + if len(resource.Sources) < 2 && len(linkFolds) == 0 { http.Error(w, "Resource is not merged", http.StatusBadRequest) return } sourceTargets := registry.SourceTargets(path) - if len(sourceTargets) == 0 { + if len(sourceTargets) == 0 && len(linkFolds) == 0 { http.Error(w, "No source targets found", http.StatusBadRequest) return } @@ -979,31 +982,56 @@ func (h *QueryService) HandleReportMerge(w http.ResponseWriter, r *http.Request) for _, source := range payload.Sources { filteredSources[strings.ToLower(strings.TrimSpace(source))] = struct{}{} } + selected := func(sources ...unified.DataSource) bool { + if len(filteredSources) == 0 { + return true + } + for _, source := range sources { + if _, ok := filteredSources[strings.ToLower(string(source))]; ok { + return true + } + } + return false + } reason := strings.TrimSpace(payload.Notes) if reason == "" { reason = "reported_incorrect_merge" } - exclusionsAdded := 0 - seen := make(map[string]struct{}) - for _, target := range sourceTargets { - if len(filteredSources) > 0 { - if _, ok := filteredSources[strings.ToLower(string(target.Source))]; !ok { - continue - } + // An operator link that brought a reported source in is undone by + // excluding the link's own pair, which replaces the link in the store. + // Identity merges are kept apart by excluding the candidate ID each + // source's record takes when no match is allowed; for a linked source that + // candidate names neither side of the link. + type exclusionPair struct{ a, b string } + pairs := make([]exclusionPair, 0, len(linkFolds)+len(sourceTargets)) + for _, fold := range linkFolds { + if selected(fold.Sources...) { + pairs = append(pairs, exclusionPair{fold.HolderID, fold.FoldedID}) } - if target.CandidateID == "" || target.CandidateID == path { + } + for _, target := range sourceTargets { + if !selected(target.Source) || target.CandidateID == "" || target.CandidateID == path { continue } - key := target.CandidateID + pairs = append(pairs, exclusionPair{path, target.CandidateID}) + } + + exclusionsAdded := 0 + seen := make(map[string]struct{}) + for _, pair := range pairs { + key := pair.a + "|" + pair.b + if pair.b < pair.a { + key = pair.b + "|" + pair.a + } if _, ok := seen[key]; ok { continue } seen[key] = struct{}{} exclusion := unified.ResourceExclusion{ - ResourceA: path, - ResourceB: target.CandidateID, + ResourceA: pair.a, + ResourceB: pair.b, Reason: reason, CreatedBy: getUserID(r), CreatedAt: time.Now().UTC(), @@ -1013,7 +1041,8 @@ func (h *QueryService) HandleReportMerge(w http.ResponseWriter, r *http.Request) Err(err). Str("orgID", orgID). Str("resourceID", path). - Str("candidateID", target.CandidateID). + Str("resourceA", pair.a). + Str("resourceB", pair.b). Msg("Failed to add resource merge exclusion") http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError) return diff --git a/internal/api/resourceapi/resources_report_merge_test.go b/internal/api/resourceapi/resources_report_merge_test.go new file mode 100644 index 0000000000..0ebea98fad --- /dev/null +++ b/internal/api/resourceapi/resources_report_merge_test.go @@ -0,0 +1,263 @@ +package resourceapi + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +// Report-merge on a resource an operator link folded together splits it on +// every surface, as unlink does. It used to record exclusions only against +// candidate IDs derived from the merged resource's type, which name neither +// side of the link, so the link row survived and the pair stayed merged while +// the API answered "Merge reported" and the drawer toasted "Resource split +// applied". REST seeds from the monitor's already-linked listing, as in +// production, so the folded row is never in its registry. +func TestResourceReportMergeSplitsOperatorLinkedPair(t *testing.T) { + now := time.Now().UTC() + vm := models.VM{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now} + agent := models.Host{ID: "host-box", Hostname: "box-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now} + node := models.Node{ID: "lab-pve1", Name: "pve1", Instance: "lab", Host: "https://10.0.0.5:8006", Status: "online", LastSeen: now} + // The agent and the Docker host take canonical IDs derived from their + // machine IDs, which no source-specific candidate reproduces, so a + // same-type link between them fails the same way a cross-type one does. + docker := models.DockerHost{ID: "docker-box", Hostname: "dock-box", MachineID: "0011223344556677", Status: "online", LastSeen: now} + // Two linked agents share their only source, which the API used to refuse + // as "Resource is not merged". + spare := models.Host{ID: "host-spare", Hostname: "spare-agent", MachineID: "8899aabbccddeeff", Status: "online", LastSeen: now} + + for _, tc := range []struct { + name string + snapshot models.StateSnapshot + // linkFromAgent links from the agent's side, as the agent's drawer + // would; otherwise from the other resource's side. + linkFromAgent bool + }{ + {"vm-agent", models.StateSnapshot{VMs: []models.VM{vm}, Hosts: []models.Host{agent}}, false}, + {"vm-agent-from-agent", models.StateSnapshot{VMs: []models.VM{vm}, Hosts: []models.Host{agent}}, true}, + {"node-agent", models.StateSnapshot{Nodes: []models.Node{node}, Hosts: []models.Host{agent}}, false}, + {"node-agent-from-agent", models.StateSnapshot{Nodes: []models.Node{node}, Hosts: []models.Host{agent}}, true}, + {"docker-agent", models.StateSnapshot{DockerHosts: []models.DockerHost{docker}, Hosts: []models.Host{agent}}, false}, + {"docker-agent-from-agent", models.StateSnapshot{DockerHosts: []models.DockerHost{docker}, Hosts: []models.Host{agent}}, true}, + {"agent-agent", models.StateSnapshot{Hosts: []models.Host{spare, agent}}, false}, + } { + t.Run(tc.name, func(t *testing.T) { + tc.snapshot.LastUpdate = now + h, store, rebuild := newReportMergeHarness(t, tc.snapshot) + resources := rebuild() + if len(resources) != 2 { + t.Fatalf("fixture produced %d resources, want 2 apart: %+v", len(resources), resources) + } + var agentID, otherID string + for _, resource := range resources { + if resource.Name == agent.Hostname { + agentID = resource.ID + } else { + otherID = resource.ID + } + } + if agentID == "" || otherID == "" { + t.Fatalf("fixture lacks a standalone agent: %+v", resources) + } + fromID, targetID := otherID, agentID + if tc.linkFromAgent { + fromID, targetID = agentID, otherID + } + postResourceAction(t, h, fromID, "link", map[string]any{"targetId": targetID}) + linked := rebuild() + if len(linked) != 1 { + t.Fatalf("link left %d monitor resources, want one: %+v", len(linked), linked) + } + assertRESTResourceCount(t, h, 1) + + // Report as the drawer's Split dialog does: every merged source. + sources := make([]string, 0, len(linked[0].Sources)) + for _, source := range linked[0].Sources { + sources = append(sources, string(source)) + } + postResourceAction(t, h, linked[0].ID, "report-merge", map[string]any{"sources": sources}) + if split := rebuild(); len(split) != 2 { + t.Fatalf("report-merge left %d monitor resources, want the pair apart: %+v", len(split), split) + } + assertRESTResourceCount(t, h, 2) + if links, err := store.GetLinks(); err != nil || len(links) != 0 { + t.Fatalf("report-merge left links %+v (err %v)", links, err) + } + exclusions, err := store.GetExclusions() + if err != nil { + t.Fatal(err) + } + excludedPair := false + for _, exclusion := range exclusions { + if (exclusion.ResourceA == agentID && exclusion.ResourceB == otherID) || + (exclusion.ResourceA == otherID && exclusion.ResourceB == agentID) { + excludedPair = true + } + } + if !excludedPair { + t.Fatalf("report-merge did not exclude the linked pair %s/%s: %+v", agentID, otherID, exclusions) + } + }) + } +} + +// A source filter undoes only the links whose folded side brought a named +// source in. +func TestResourceReportMergeSourceFilterSelectsLinks(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + LastUpdate: now, + VMs: []models.VM{{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now}}, + Hosts: []models.Host{{ID: "host-box", Hostname: "box-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now}}, + } + h, store, rebuild := newReportMergeHarness(t, snapshot) + var vmID, agentID string + for _, resource := range rebuild() { + switch resource.Type { + case unified.ResourceTypeVM: + vmID = resource.ID + case unified.ResourceTypeAgent: + agentID = resource.ID + } + } + postResourceAction(t, h, vmID, "link", map[string]any{"targetId": agentID}) + rebuild() + + // The VM's own source was not folded in by the link. + postResourceAction(t, h, vmID, "report-merge", map[string]any{"sources": []string{"proxmox"}}) + if links, err := store.GetLinks(); err != nil || len(links) != 1 { + t.Fatalf("a filter naming only the VM's own source changed the link: %+v (err %v)", links, err) + } + postResourceAction(t, h, vmID, "report-merge", map[string]any{"sources": []string{"agent"}}) + if links, err := store.GetLinks(); err != nil || len(links) != 0 { + t.Fatalf("a filter naming the agent left the link: %+v (err %v)", links, err) + } + if split := rebuild(); len(split) != 2 { + t.Fatalf("report-merge left %d monitor resources, want two: %+v", len(split), split) + } +} + +// Report-merge on a pair identity matching merged keeps it apart through the +// monitor's rebuilds, whose identity pins persist on one generation and steer +// the next. +func TestResourceReportMergeSplitsIdentityMergedPair(t *testing.T) { + now := time.Now().UTC() + interfaces := []models.HostNetworkInterface{{Name: "eth0", MAC: "aa:bb:cc:dd:ee:ff", Addresses: []string{"10.0.0.5"}}} + snapshot := models.StateSnapshot{ + LastUpdate: now, + Hosts: []models.Host{{ID: "host-1", Hostname: "alpha", MachineID: "0123456789abcdef", Status: "online", LastSeen: now, NetworkInterfaces: interfaces}}, + DockerHosts: []models.DockerHost{{ID: "docker-1", Hostname: "alpha", MachineID: "0123456789abcdef", Status: "online", CPUs: 4, LastSeen: now, NetworkInterfaces: interfaces}}, + } + h, _, rebuild := newReportMergeHarness(t, snapshot) + merged := rebuild() + if len(merged) != 1 { + t.Fatalf("fixture produced %d resources, want one merged: %+v", len(merged), merged) + } + sources := make([]string, 0, len(merged[0].Sources)) + for _, source := range merged[0].Sources { + sources = append(sources, string(source)) + } + postResourceAction(t, h, merged[0].ID, "report-merge", map[string]any{"sources": sources}) + if split := rebuild(); len(split) != 2 { + t.Fatalf("report-merge left %d monitor resources, want two: %+v", len(split), split) + } + assertRESTResourceCount(t, h, 2) +} + +// newReportMergeHarness serves the resources API from a monitor adapter on the +// same SQLite store, rebuilt three times per step: identity pins persist on one +// rebuild and steer the next. +func newReportMergeHarness(t *testing.T, snapshot models.StateSnapshot) (*QueryService, unified.ResourceStore, func() []unified.Resource) { + t.Helper() + h := NewQueryService(&config.Config{DataPath: t.TempDir()}) + store, err := h.getStore("default") + if err != nil { + t.Fatalf("getStore: %v", err) + } + adapter := unified.NewMonitorAdapter(unified.NewRegistry(store)) + h.SetStateProvider(&monitorAdapterSeedProvider{adapter: adapter, snapshot: snapshot}) + rebuild := func() []unified.Resource { + t.Helper() + for i := 0; i < 3; i++ { + adapter.PopulateFromSnapshot(snapshot) + } + return adapter.GetAll() + } + return h, store, rebuild +} + +func postResourceAction(t *testing.T, h *QueryService, resourceID, action string, payload map[string]any) { + t.Helper() + body, _ := json.Marshal(payload) + rec := httptest.NewRecorder() + h.HandleResourceRoutes(rec, httptest.NewRequest(http.MethodPost, "/api/resources/"+resourceID+"/"+action, bytes.NewReader(body))) + if rec.Code != http.StatusOK { + t.Fatalf("%s %s status = %d body=%s", action, resourceID, rec.Code, rec.Body.String()) + } +} + +func assertRESTResourceCount(t *testing.T, h *QueryService, want int) { + t.Helper() + rec := httptest.NewRecorder() + h.HandleListResources(rec, httptest.NewRequest(http.MethodGet, "/api/resources", nil)) + if rec.Code != http.StatusOK { + t.Fatalf("list status = %d body=%s", rec.Code, rec.Body.String()) + } + var resp ResourcesResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode list: %v", err) + } + if len(resp.Data) != want { + ids := make([]string, 0, len(resp.Data)) + for _, resource := range resp.Data { + ids = append(ids, resource.ID+"("+string(resource.Type)+")") + } + t.Fatalf("REST lists %d resources %v, want %d", len(resp.Data), ids, want) + } +} + +// Canonical-ID succession re-keys link rows with UPDATE OR IGNORE but moves +// primary_id unconditionally, so a row whose re-key collided with the +// successor's own row keeps the retired endpoint and gains the successor as +// its primary. That row used to merge the successor through a pair it does +// not name, so reporting (or unlinking) the merged pair left it behind and +// the next rebuild merged the pair again. +func TestResourceReportMergeSplitsPairBehindASuccessionShadowedRow(t *testing.T) { + now := time.Now().UTC() + snapshot := models.StateSnapshot{ + LastUpdate: now, + VMs: []models.VM{{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now}}, + Hosts: []models.Host{{ID: "host-box", Hostname: "box-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now}}, + } + h, store, rebuild := newReportMergeHarness(t, snapshot) + var vmID, agentID string + for _, resource := range rebuild() { + switch resource.Type { + case unified.ResourceTypeVM: + vmID = resource.ID + case unified.ResourceTypeAgent: + agentID = resource.ID + } + } + if err := store.AddLink(unified.ResourceLink{ResourceA: "agent-retired0000000", ResourceB: vmID, PrimaryID: agentID, CreatedAt: now.Add(-time.Hour)}); err != nil { + t.Fatal(err) + } + postResourceAction(t, h, vmID, "link", map[string]any{"targetId": agentID}) + linked := rebuild() + if len(linked) != 1 { + t.Fatalf("link left %d monitor resources, want one: %+v", len(linked), linked) + } + + postResourceAction(t, h, linked[0].ID, "report-merge", map[string]any{}) + if split := rebuild(); len(split) != 2 { + t.Fatalf("report-merge left %d monitor resources, want the pair apart: %+v", len(split), split) + } + assertRESTResourceCount(t, h, 2) +} diff --git a/internal/unifiedresources/clone.go b/internal/unifiedresources/clone.go index 0beca03f35..bebac8b929 100644 --- a/internal/unifiedresources/clone.go +++ b/internal/unifiedresources/clone.go @@ -69,6 +69,7 @@ func cloneResourceData(in *Resource) Resource { out.Metrics = cloneResourceMetrics(in.Metrics) out.ParentID = cloneStringPtr(in.ParentID) out.parentBySource = cloneParentBySourceMap(in.parentBySource) + out.linkFolds = cloneManualLinkFolds(in.linkFolds) out.Tags = cloneStringSlice(in.Tags) out.Incidents = cloneResourceIncidentSlice(in.Incidents) out.Relationships = cloneResourceRelationshipSlice(in.Relationships) diff --git a/internal/unifiedresources/clone_test.go b/internal/unifiedresources/clone_test.go index 63b87b54a1..e52c6ff3e6 100644 --- a/internal/unifiedresources/clone_test.go +++ b/internal/unifiedresources/clone_test.go @@ -269,6 +269,25 @@ func TestCloneResource_MutateParentBySource(t *testing.T) { } } +// Operator link folds ride clones, so a registry seeded from another +// registry's listing can name each link's pair, and stay detached from them. +func TestCloneResource_CarriesLinkFolds(t *testing.T) { + original := &Resource{ + ID: "vm-1", + linkFolds: []ManualLinkFold{{HolderID: "vm-1", FoldedID: "agent-1", Sources: []DataSource{SourceAgent}}}, + } + cloned := cloneResource(original) + if len(cloned.linkFolds) != 1 || cloned.linkFolds[0].FoldedID != "agent-1" { + t.Fatalf("clone lost link folds: %+v", cloned.linkFolds) + } + + cloned.linkFolds[0].Sources[0] = SourceDocker + cloned.linkFolds[0].FoldedID = "MUTATED" + if original.linkFolds[0].Sources[0] != SourceAgent || original.linkFolds[0].FoldedID != "agent-1" { + t.Errorf("mutating cloned link folds affected the original: %+v", original.linkFolds) + } +} + func TestCloneResource_MutateAvailabilityTimes(t *testing.T) { checkedAt := time.Date(2026, time.July, 9, 12, 0, 0, 0, time.UTC) succeededAt := checkedAt.Add(-time.Minute) diff --git a/internal/unifiedresources/code_standards_test.go b/internal/unifiedresources/code_standards_test.go index 6bf168aa27..696a6ce507 100644 --- a/internal/unifiedresources/code_standards_test.go +++ b/internal/unifiedresources/code_standards_test.go @@ -729,6 +729,28 @@ func TestResourceAPIUsesCanonicalTenantUnifiedSeed(t *testing.T) { } } +// Report-merge undoes an operator link through the registry's fold record. A +// candidate ID derived from the merged resource's type names neither side of +// a link, and the resources API's registry, seeded from the monitor's +// already-linked listing, never holds the folded row to derive one from. +func TestResourceAPIReportMergeExcludesRegistryLinkFolds(t *testing.T) { + data, err := os.ReadFile(filepath.Join("..", "api", "resourceapi", "resources.go")) + if err != nil { + t.Fatalf("failed to read resources.go: %v", err) + } + source := string(data) + + for _, snippet := range []string{ + "linkFolds := registry.ManualLinkFolds(path)", + "if len(resource.Sources) < 2 && len(linkFolds) == 0 {", + "pairs = append(pairs, exclusionPair{fold.HolderID, fold.FoldedID})", + } { + if !strings.Contains(source, snippet) { + t.Fatalf("internal/api/resourceapi/resources.go report-merge must contain %q", snippet) + } + } +} + func TestAgentlessAvailabilityTargetKindStaysCanonical(t *testing.T) { requiredSnippets := map[string][]string{ filepath.Join("..", "config", "availability.go"): { diff --git a/internal/unifiedresources/manual_link_folds.go b/internal/unifiedresources/manual_link_folds.go new file mode 100644 index 0000000000..4091d9e66a --- /dev/null +++ b/internal/unifiedresources/manual_link_folds.go @@ -0,0 +1,78 @@ +package unifiedresources + +// ManualLinkFold records one operator link that applyManualLinks applied: the +// resource FoldedID names was merged into the one HolderID names. The two IDs +// are the link row's own pair, so an exclusion naming them replaces the link +// (the store keeps one decision per pair). A candidate ID derived from the +// merged resource's type cannot: for an agent linked into a VM it is a vm- ID +// that names neither side. Sources lists what the folded side brought, +// including anything links had already folded into it. +type ManualLinkFold struct { + HolderID string + FoldedID string + Sources []DataSource +} + +// recordManualLinkFold notes on the link's merged resource that it took in +// other through the link pair (primaryID, otherID), together with the folds +// other already held, so each link along a chain keeps its own pair. The +// record is unexported and rides in-memory clones, so a registry seeded from +// another registry's listing, as the resources API's is, sees the folds the +// monitor applied although it never holds the folded rows itself. A pair is +// recorded once: record ingest can recreate a folded side whose source +// mapping points at a holder of another type and fold it again, possibly +// with fewer sources than it brought before, so a repeated pair adds its +// sources to the fold already recorded. The holder still carries what the +// earlier fold brought. +func recordManualLinkFold(primary *Resource, primaryID string, other *Resource, otherID string) { + added := make([]ManualLinkFold, 0, len(other.linkFolds)+1) + added = append(added, other.linkFolds...) + added = append(added, ManualLinkFold{ + HolderID: primaryID, + FoldedID: otherID, + Sources: cloneDataSourceSlice(other.Sources), + }) + folds := make([]ManualLinkFold, 0, len(primary.linkFolds)+len(added)) + recorded := make(map[string]int, len(primary.linkFolds)+len(added)) + record := func(fold ManualLinkFold) { + key := exclusionKey(fold.HolderID, fold.FoldedID) + if i, ok := recorded[key]; ok { + folds[i].Sources = addSources(cloneDataSourceSlice(folds[i].Sources), fold.Sources) + return + } + recorded[key] = len(folds) + folds = append(folds, fold) + } + for _, fold := range primary.linkFolds { + record(fold) + } + for _, fold := range added { + record(fold) + } + primary.linkFolds = folds +} + +// ManualLinkFolds lists the operator links folded into a resource, one per +// link pair: the resource's own earlier folds first, then each folded side's +// folds ahead of the link that took that side in. +func (rr *ResourceRegistry) ManualLinkFolds(resourceID string) []ManualLinkFold { + rr.mu.RLock() + defer rr.mu.RUnlock() + resource := rr.resources[CanonicalResourceID(resourceID)] + if resource == nil { + return nil + } + return cloneManualLinkFolds(resource.linkFolds) +} + +func cloneManualLinkFolds(in []ManualLinkFold) []ManualLinkFold { + if in == nil { + return nil + } + out := make([]ManualLinkFold, len(in)) + for i, fold := range in { + fold.Sources = cloneDataSourceSlice(fold.Sources) + out[i] = fold + } + return out +} diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 029b0dcc47..240074bb13 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -4794,8 +4794,13 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura return } for _, link := range rr.links { + // A link joins only the pair its row names. Canonical-ID succession + // can leave a row whose endpoint re-key collided with the successor's + // own row while its primary moved on to the successor; honouring that + // primary merged the successor through a row naming a retired ID, which + // an unlink or report-merge of the merged pair leaves behind. primaryID := link.PrimaryID - if primaryID == "" { + if primaryID != link.ResourceA && primaryID != link.ResourceB { primaryID = link.ResourceA } primary := rr.resources[primaryID] @@ -4836,6 +4841,7 @@ func (rr *ResourceRegistry) applyManualLinks(thresholds map[DataSource]time.Dura } rr.mergeResourceData(primary, other, thresholds) + recordManualLinkFold(primary, primaryID, other, otherID) delete(rr.resources, otherID) if rr.linkMergedIDs == nil { rr.linkMergedIDs = make(map[string]struct{}) diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index 6f04cb36b7..de62b83d9d 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -2,6 +2,7 @@ package unifiedresources import ( "encoding/json" + "reflect" "slices" "strconv" "strings" @@ -862,6 +863,144 @@ func TestManualLinkKeepsProviderPayloadsThePrimaryLacks(t *testing.T) { } } +// Each operator link folded into a resource is recorded with the link row's +// own pair, also along a chain (a Docker host linked to an agent, the agent +// linked into a VM), and the record rides the monitor's listing into a +// registry seeded from it, which never holds the folded rows. +func TestManualLinkFoldsNameEachLinkPairAlongAChain(t *testing.T) { + now := time.Now().UTC() + store, err := NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + snapshot := models.StateSnapshot{ + LastUpdate: now, + VMs: []models.VM{{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now}}, + Hosts: []models.Host{{ID: "host-box", Hostname: "box-agent", MachineID: "fedcba9876543210", Status: "online", LastSeen: now}}, + DockerHosts: []models.DockerHost{{ID: "docker-box", Hostname: "dock-box", Status: "online", LastSeen: now}}, + } + adapter := NewMonitorAdapter(NewRegistry(store)) + adapter.PopulateFromSnapshot(snapshot) + var vmID, agentID, dockerID string + for _, resource := range adapter.GetAll() { + switch { + case resource.Type == ResourceTypeVM: + vmID = resource.ID + case resource.Agent != nil: + agentID = resource.ID + case resource.Docker != nil: + dockerID = resource.ID + } + if len(resource.linkFolds) != 0 { + t.Fatalf("unlinked resource %s carries link folds %+v", resource.ID, resource.linkFolds) + } + } + if vmID == "" || agentID == "" || dockerID == "" || len(adapter.GetAll()) != 3 { + t.Fatalf("fixture did not produce a separate VM, agent and Docker host: %+v", adapter.GetAll()) + } + + // Links apply in store order, so the agent takes in the Docker host before + // the VM takes in the agent. + if err := store.AddLink(ResourceLink{ResourceA: agentID, ResourceB: dockerID, PrimaryID: agentID, CreatedAt: now}); err != nil { + t.Fatal(err) + } + if err := store.AddLink(ResourceLink{ResourceA: agentID, ResourceB: vmID, PrimaryID: agentID, CreatedAt: now.Add(time.Second)}); err != nil { + t.Fatal(err) + } + adapter.PopulateFromSnapshot(snapshot) + merged := adapter.GetAll() + if len(merged) != 1 || merged[0].ID != vmID { + t.Fatalf("links did not fold the chain into the VM: %+v", merged) + } + + // The agent linked from its own side still folds into the guest it runs + // in, and the fold names the link's pair either way. + want := []ManualLinkFold{ + {HolderID: agentID, FoldedID: dockerID, Sources: []DataSource{SourceDocker}}, + {HolderID: vmID, FoldedID: agentID, Sources: []DataSource{SourceAgent, SourceDocker}}, + } + if got := adapter.currentRegistry().ManualLinkFolds(vmID); !reflect.DeepEqual(got, want) { + t.Fatalf("monitor folds = %+v, want %+v", got, want) + } + + seeded := NewRegistry(store) + seeded.IngestResources(adapter.GetAll()) + if got := seeded.ManualLinkFolds(vmID); !reflect.DeepEqual(got, want) { + t.Fatalf("seeded registry folds = %+v, want %+v", got, want) + } + if got := seeded.ManualLinkFolds(agentID); got != nil { + t.Fatalf("folded agent ID reports folds %+v", got) + } +} + +// Record ingest recreates a folded side whose source mapping points at a +// holder of another type (an agent-type TrueNAS record linked into a VM) and +// the record-ingest epilogue folds it again, so the pair must be recorded +// once however often the records arrive. +func TestManualLinkFoldsRecordEachPairOnceAcrossRecordIngests(t *testing.T) { + now := time.Now().UTC() + store, err := NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + snapshot := models.StateSnapshot{ + LastUpdate: now, + VMs: []models.VM{{ID: "lab:pve1:101", VMID: 101, Name: "web", Node: "pve1", Instance: "lab", Status: "running", Type: "qemu", LastSeen: now}}, + } + records := []IngestRecord{{ + SourceID: "nas-1", + Resource: Resource{Type: ResourceTypeAgent, Name: "nas-1", Status: StatusOnline, LastSeen: now}, + Identity: ResourceIdentity{MachineID: "abcdef0123456789"}, + }} + probe := NewRegistry(store) + probe.IngestSnapshot(snapshot) + probe.IngestRecords(SourceTrueNAS, records) + var vmID, systemID string + for _, resource := range probe.List() { + switch resource.Type { + case ResourceTypeVM: + vmID = resource.ID + case ResourceTypeAgent: + systemID = resource.ID + } + } + if vmID == "" || systemID == "" { + t.Fatalf("fixture did not produce a separate VM and TrueNAS system: %+v", probe.List()) + } + if err := store.AddLink(ResourceLink{ResourceA: vmID, ResourceB: systemID, PrimaryID: vmID, CreatedAt: now}); err != nil { + t.Fatal(err) + } + + rr := NewRegistry(store) + rr.IngestSnapshot(snapshot) + for i := 0; i < 3; i++ { + rr.IngestRecords(SourceTrueNAS, records) + } + if listed := rr.List(); len(listed) != 1 { + t.Fatalf("link did not fold the TrueNAS system into the VM: %+v", listed) + } + want := []ManualLinkFold{{HolderID: vmID, FoldedID: systemID, Sources: []DataSource{SourceTrueNAS}}} + if got := rr.ManualLinkFolds(vmID); !reflect.DeepEqual(got, want) { + t.Fatalf("folds after repeated record ingests = %+v, want %+v", got, want) + } +} + +// A folded side recreated with fewer sources than it brought before (one of +// an identity-merged agent's records replayed) folds again; the holder still +// carries everything the earlier fold brought, so the fold keeps every +// source and a report naming any of them still selects the link. +func TestManualLinkFoldRepeatedPairKeepsEarlierSources(t *testing.T) { + holder := &Resource{ID: "vm-1"} + recordManualLinkFold(holder, "vm-1", &Resource{ID: "agent-1", Sources: []DataSource{SourceAgent, SourceDocker}}, "agent-1") + recordManualLinkFold(holder, "vm-1", &Resource{ID: "agent-1", Sources: []DataSource{SourceAgent}}, "agent-1") + want := []ManualLinkFold{{HolderID: "vm-1", FoldedID: "agent-1", Sources: []DataSource{SourceAgent, SourceDocker}}} + if !reflect.DeepEqual(holder.linkFolds, want) { + t.Fatalf("folds = %+v, want %+v", holder.linkFolds, want) + } +} + func TestLeaseExpiredKubernetesClusterStaysOffline(t *testing.T) { rr := NewRegistry(nil) rr.IngestSnapshot(models.StateSnapshot{ diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index 1e54e94b23..19cd64861b 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -57,6 +57,9 @@ type Resource struct { ParentName string `json:"parentName,omitempty"` ChildCount int `json:"childCount,omitempty"` parentBySource map[DataSource]string + // linkFolds lists the operator links folded into this resource + // (recordManualLinkFold); report-merge excludes their pairs. + linkFolds []ManualLinkFold Tags []string `json:"tags,omitempty"` CustomURL string `json:"customUrl,omitempty"`