mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 21:20:00 +00:00
Apply operator links in the mock-mode unified view
In mock mode the websocket broadcast and /api/state serve a view of the fixture graph, and the fixture graph was built in a registry with no store, so it never loaded the operator's manual links. The resources API seeds its own store-backed registry from that list and applied them, so a standalone agent linked into a VM showed merged in /api/resources and stayed a separate row in the broadcast, even after a restart. Peers verifying link UI on mock stacks had to block the websocket to see it. The mock view now builds the fixture graph with the links the monitor adapter's current registry generation loaded (MonitorAdapter.ManualLinks), through NewRegistryWithManualLinks: a registry with those links and no store, so fixture ingest cannot persist identity pins, successions or change rows. The links apply at each ingest stage, as in the live rebuild, so an availability check that names the folded resource by source still projects onto the primary. Without links the shared memoized snapshot is used as before. The cached view is keyed on the fixture version and the link list, so a link change rebuilds it without a fixture tick. Its freshness, which the resources API keys its registry cache on, is assigned in publication order: a view with the published fixture freshness and links keeps the published value, and any other view gets a later one than every view before it. The fence view behind currentModeReadState still omits links; it only covers the window after leaving mock mode, before the registry rebuilds.
This commit is contained in:
@@ -8982,6 +8982,18 @@ their own heartbeat cutoff (`fleethealth.AgentStaleThreshold`, which is
|
||||
separate from the monitoring reporting lease). Agent registration, enrolment,
|
||||
install, update, removal and report identity are unchanged.
|
||||
|
||||
### 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
|
||||
|
||||
@@ -2834,6 +2834,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
|
||||
|
||||
@@ -6616,9 +6616,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
|
||||
@@ -6673,6 +6673,65 @@ thresholds on the rebuild, supplemental and overlay paths, and
|
||||
published inventory with no continuity, an unrelated absent machine and the
|
||||
linked agent's own saved enrollment.
|
||||
|
||||
### 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, as before. The broadcast's presentation
|
||||
coalescing does not consult exclusions in either mode, while the resources
|
||||
API's `ListForPresentation` does.
|
||||
- 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
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -2009,7 +2009,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",
|
||||
|
||||
@@ -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
|
||||
@@ -5146,27 +5148,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
|
||||
@@ -5202,6 +5232,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{}
|
||||
@@ -5286,7 +5343,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
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
@@ -5718,6 +5719,137 @@ 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))
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -2238,6 +2238,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)
|
||||
|
||||
Reference in New Issue
Block a user