mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-07-10 00:14:38 +00:00
fix(resources): pin canonical host IDs to durable identity so they survive restarts
Canonical IDs for merged-source hosts (PVE node + pulse-agent) were minted from whichever identity keys the creating record happened to carry: the agent record knows the machine ID, the Proxmox node record only knows cluster+hostname. The registry rebuilds from scratch every tick, so a boot window where the agent had not checked in yet minted a cluster-keyed ID (agent-7a62... for delly) while steady state minted a machine-keyed one (agent-bdd4...). Every restart re-ran the race, fragmenting the resource_changes journal into per-boot eras (9.4k vs 6.1k rows for the same host) and silently truncating report availability and UI timelines. Fix, in the layer that owns identity: - Persist identity pins (canonical_id <-> machine_id/dmi/cluster/hostname) in the previously schema-only resource_identities table, written by the store-backed registry after monitor-adapter rebuilds, diff-aware so steady-state ticks cost no writes. - Complete weak incoming identities from the pins before matching and ID derivation, so a node-only boot window derives the same machine-keyed canonical ID as steady state. Derivation itself is unchanged; ephemeral nil-store registries behave exactly as before. - Expand change-journal reads (Get/Count families, SQLite and memory) to the full era set recomputed from the pinned identity keys, healing historical journals at query time with no row migration. Reads keyed by a stale era ID resolve to the same merged timeline. Regression tests cover both ingest orders, restart simulation via the monitor adapter, era ID derivation, and era-merged journal reads on both store implementations. Contracts updated: unified-resources obligation 25 (durable identity pins), monitoring obligation 10 (adapter rebuild persistence).
This commit is contained in:
rcourtman
parent
3a51429a58
commit
e8455e84b7
11 changed files with 879 additions and 32 deletions
|
|
@ -311,6 +311,13 @@ truth for live infrastructure data.
|
|||
SMB/NFS shares must enter unified resources as `network-share` records with
|
||||
the TrueNAS share facet, not as generic storage rows or Docker/container
|
||||
compatibility records.
|
||||
10. Keep the monitor-adapter rebuild lifecycle persisting canonical identity
|
||||
pins. `internal/unifiedresources/monitor_adapter.go` calls
|
||||
`PersistIdentityPins` after snapshot rebuilds and supplemental-record
|
||||
ingestion so canonical host IDs stay stable across restarts (see the
|
||||
unified-resources contract's durable identity-pin obligation). Rebuild
|
||||
paths added to the adapter must keep that persistence step; ephemeral
|
||||
snapshot-bridge adapters stay read-only.
|
||||
|
||||
## Current State
|
||||
|
||||
|
|
|
|||
|
|
@ -949,6 +949,29 @@ AI-only summary payloads, or page-local heuristics.
|
|||
`verificationOutcome.status` presentation for `verified`,
|
||||
`unverified`, `failed`, and `unknown` so resource history, future
|
||||
action audit consumers, and tests share one bounded vocabulary.
|
||||
25. Keep canonical host IDs stable across restarts through durable
|
||||
identity pins. Canonical ID derivation hashes the strongest identity
|
||||
key available at mint time, and merged-source hosts (Proxmox node +
|
||||
pulse-agent) expose different identity subsets depending on which
|
||||
records are present when a registry rebuilds: agent records carry the
|
||||
machine ID, node records only cluster+hostname. The store-backed
|
||||
registry therefore persists identity pins
|
||||
(`canonical_id` ↔ machine ID / DMI UUID / cluster / hostname) in the
|
||||
`resource_identities` table (`ResourceStore.UpsertResourceIdentityPins`
|
||||
/ `ListResourceIdentityPins`, written diff-aware after monitor-adapter
|
||||
rebuilds) and completes weak incoming identities from those pins
|
||||
before matching and ID derivation
|
||||
(`internal/unifiedresources/canonical_id_pins.go`). Change-journal
|
||||
reads (`GetRecentChanges*` / `CountRecentChanges*`) must expand a
|
||||
canonical ID to the era set recomputable from its pinned identity keys
|
||||
(`ResourceIdentityPin.EraIDs`) so journal rows recorded under an
|
||||
earlier era's ID stay part of the resource's timeline; reads keyed by
|
||||
a stale era ID resolve to the same merged timeline. New ingest paths
|
||||
must not mint host canonical IDs from snapshot-content-dependent
|
||||
identity subsets without consulting the pins, and pin writes stay on
|
||||
the durable store-backed registry (ephemeral per-request registries
|
||||
consult, never write). Regression coverage:
|
||||
`internal/unifiedresources/canonical_id_pins_test.go`.
|
||||
|
||||
## Current State
|
||||
|
||||
|
|
|
|||
189
internal/unifiedresources/canonical_id_pins.go
Normal file
189
internal/unifiedresources/canonical_id_pins.go
Normal file
|
|
@ -0,0 +1,189 @@
|
|||
package unifiedresources
|
||||
|
||||
import (
|
||||
"log"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// identityPinIndex is the in-memory lookup over the durable identity pins.
|
||||
// It answers one question at ingest time: "have we ever durably assigned a
|
||||
// canonical ID to the physical host this (possibly weak) identity refers to,
|
||||
// and what strong identity keys did we record for it?"
|
||||
type identityPinIndex struct {
|
||||
byCanonicalID map[string]ResourceIdentityPin
|
||||
byMachineID map[string]ResourceIdentityPin
|
||||
byDMIUUID map[string]ResourceIdentityPin
|
||||
byClusterHost map[string]ResourceIdentityPin
|
||||
// byHostname holds pins per normalized hostname. Hostnames are not
|
||||
// unique across machines, so lookups through this map require the
|
||||
// bucket to be unambiguous.
|
||||
byHostname map[string][]ResourceIdentityPin
|
||||
}
|
||||
|
||||
func newIdentityPinIndex(pins []ResourceIdentityPin) *identityPinIndex {
|
||||
index := &identityPinIndex{
|
||||
byCanonicalID: make(map[string]ResourceIdentityPin, len(pins)),
|
||||
byMachineID: make(map[string]ResourceIdentityPin),
|
||||
byDMIUUID: make(map[string]ResourceIdentityPin),
|
||||
byClusterHost: make(map[string]ResourceIdentityPin),
|
||||
byHostname: make(map[string][]ResourceIdentityPin),
|
||||
}
|
||||
for _, pin := range pins {
|
||||
pin = pin.normalized()
|
||||
if pin.CanonicalID == "" || !pin.hasStrongKey() {
|
||||
continue
|
||||
}
|
||||
index.byCanonicalID[pin.CanonicalID] = pin
|
||||
if pin.MachineID != "" {
|
||||
index.byMachineID[pin.MachineID] = pin
|
||||
}
|
||||
if pin.DMIUUID != "" {
|
||||
index.byDMIUUID[pin.DMIUUID] = pin
|
||||
}
|
||||
if pin.ClusterName != "" && pin.Hostname != "" {
|
||||
index.byClusterHost[clusterHostPinKey(pin.ClusterName, pin.Hostname)] = pin
|
||||
}
|
||||
if pin.Hostname != "" {
|
||||
index.byHostname[pin.Hostname] = append(index.byHostname[pin.Hostname], pin)
|
||||
}
|
||||
}
|
||||
return index
|
||||
}
|
||||
|
||||
func clusterHostPinKey(clusterName, hostname string) string {
|
||||
return strings.ToLower(strings.TrimSpace(clusterName)) + "\x00" + NormalizeHostname(hostname)
|
||||
}
|
||||
|
||||
// find resolves the pin for an incoming identity, strongest key first. A pin
|
||||
// found through a weaker key is rejected when the incoming identity carries a
|
||||
// strong key that contradicts the pin (a different machine claiming the same
|
||||
// cluster slot or hostname must mint fresh, not absorb the old host).
|
||||
func (index *identityPinIndex) find(identity ResourceIdentity) (ResourceIdentityPin, bool) {
|
||||
if index == nil {
|
||||
return ResourceIdentityPin{}, false
|
||||
}
|
||||
machineID := strings.TrimSpace(identity.MachineID)
|
||||
dmiUUID := strings.TrimSpace(identity.DMIUUID)
|
||||
if machineID != "" {
|
||||
if pin, ok := index.byMachineID[machineID]; ok {
|
||||
return pin, true
|
||||
}
|
||||
}
|
||||
if dmiUUID != "" {
|
||||
if pin, ok := index.byDMIUUID[dmiUUID]; ok && pinCompatible(pin, machineID, "") {
|
||||
return pin, true
|
||||
}
|
||||
}
|
||||
clusterName := strings.TrimSpace(identity.ClusterName)
|
||||
for _, hostname := range identity.Hostnames {
|
||||
normalized := NormalizeHostname(hostname)
|
||||
if normalized == "" {
|
||||
continue
|
||||
}
|
||||
if clusterName != "" {
|
||||
if pin, ok := index.byClusterHost[clusterHostPinKey(clusterName, normalized)]; ok && pinCompatible(pin, machineID, dmiUUID) {
|
||||
return pin, true
|
||||
}
|
||||
}
|
||||
bucket := index.byHostname[normalized]
|
||||
if len(bucket) == 1 && pinCompatible(bucket[0], machineID, dmiUUID) {
|
||||
return bucket[0], true
|
||||
}
|
||||
}
|
||||
return ResourceIdentityPin{}, false
|
||||
}
|
||||
|
||||
// pinCompatible reports whether an incoming identity's strong keys are
|
||||
// consistent with the pin. Empty incoming keys never contradict; the pin's
|
||||
// own empty keys never contradict either.
|
||||
func pinCompatible(pin ResourceIdentityPin, machineID, dmiUUID string) bool {
|
||||
if machineID != "" && pin.MachineID != "" && machineID != pin.MachineID {
|
||||
return false
|
||||
}
|
||||
if dmiUUID != "" && pin.DMIUUID != "" && dmiUUID != pin.DMIUUID {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (rr *ResourceRegistry) loadIdentityPins() {
|
||||
rr.identityPins = newIdentityPinIndex(nil)
|
||||
if rr.store == nil {
|
||||
return
|
||||
}
|
||||
pins, err := rr.store.ListResourceIdentityPins()
|
||||
if err != nil {
|
||||
log.Printf("unifiedresources: failed to load identity pins from store: %v", err)
|
||||
return
|
||||
}
|
||||
rr.identityPins = newIdentityPinIndex(pins)
|
||||
}
|
||||
|
||||
// completeIdentityFromPins fills the machine-level identity keys a weak
|
||||
// incoming identity lacks, from the durable pin for the same physical host.
|
||||
// This runs before identity matching and canonical-ID derivation, so a boot
|
||||
// window where the agent has not checked in yet (the Proxmox node record only
|
||||
// knows cluster+hostname) still derives the same canonical ID as a steady
|
||||
// state rebuild that knows the machine ID. Only missing fields are filled; an
|
||||
// incoming identity that already carries a machine ID is never overridden.
|
||||
func (rr *ResourceRegistry) completeIdentityFromPins(resourceType ResourceType, identity ResourceIdentity) ResourceIdentity {
|
||||
if CanonicalResourceType(resourceType) != ResourceTypeAgent {
|
||||
return identity
|
||||
}
|
||||
if strings.TrimSpace(identity.MachineID) != "" && strings.TrimSpace(identity.DMIUUID) != "" {
|
||||
return identity
|
||||
}
|
||||
pin, ok := rr.identityPins.find(identity)
|
||||
if !ok {
|
||||
return identity
|
||||
}
|
||||
if strings.TrimSpace(identity.MachineID) == "" {
|
||||
identity.MachineID = pin.MachineID
|
||||
}
|
||||
if strings.TrimSpace(identity.DMIUUID) == "" {
|
||||
identity.DMIUUID = pin.DMIUUID
|
||||
}
|
||||
return identity
|
||||
}
|
||||
|
||||
// PersistIdentityPins writes the identity pins for the registry's current
|
||||
// host resources into the resource store. Only new or changed pins are
|
||||
// written, so steady-state rebuild ticks cost no writes. Call this after a
|
||||
// rebuild on the durable store-backed registry; ephemeral per-request
|
||||
// registries consult pins but do not write them.
|
||||
func (rr *ResourceRegistry) PersistIdentityPins() {
|
||||
if rr.store == nil {
|
||||
return
|
||||
}
|
||||
|
||||
rr.mu.RLock()
|
||||
var pins []ResourceIdentityPin
|
||||
for _, resource := range rr.resources {
|
||||
pin, ok := identityPinForResource(resource)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if existing, known := rr.identityPins.byCanonicalID[pin.CanonicalID]; known && existing == pin {
|
||||
continue
|
||||
}
|
||||
pins = append(pins, pin)
|
||||
}
|
||||
rr.mu.RUnlock()
|
||||
|
||||
if len(pins) == 0 {
|
||||
return
|
||||
}
|
||||
if err := rr.store.UpsertResourceIdentityPins(pins); err != nil {
|
||||
log.Printf("unifiedresources: failed to persist identity pins: %v", err)
|
||||
return
|
||||
}
|
||||
refreshed, err := rr.store.ListResourceIdentityPins()
|
||||
if err != nil {
|
||||
log.Printf("unifiedresources: failed to reload identity pins after persist: %v", err)
|
||||
return
|
||||
}
|
||||
index := newIdentityPinIndex(refreshed)
|
||||
rr.mu.Lock()
|
||||
rr.identityPins = index
|
||||
rr.mu.Unlock()
|
||||
}
|
||||
|
|
@ -32,3 +32,83 @@ func CanonicalResourceID(id string) string {
|
|||
}
|
||||
return trimmed
|
||||
}
|
||||
|
||||
// ResourceIdentityPin is the durable identity→canonical-ID record persisted in
|
||||
// the resource store (resource_identities table). It exists because canonical
|
||||
// ID derivation keys off the strongest identity field known at mint time, and
|
||||
// merged-source hosts (PVE node + pulse-agent) expose different identity
|
||||
// subsets depending on which records are present when the registry rebuilds:
|
||||
// the agent record knows the machine ID, the Proxmox node record only knows
|
||||
// cluster+hostname. Pins let a rebuild that only sees the weak subset recover
|
||||
// the strong keys, so the same physical host derives the same canonical ID in
|
||||
// every boot, in every registry instance.
|
||||
type ResourceIdentityPin struct {
|
||||
CanonicalID string
|
||||
ResourceType ResourceType
|
||||
MachineID string
|
||||
DMIUUID string
|
||||
ClusterName string
|
||||
// Hostname is the normalized primary hostname (NormalizeHostname).
|
||||
Hostname string
|
||||
}
|
||||
|
||||
func (p ResourceIdentityPin) normalized() ResourceIdentityPin {
|
||||
p.CanonicalID = CanonicalResourceID(p.CanonicalID)
|
||||
p.ResourceType = CanonicalResourceType(p.ResourceType)
|
||||
p.MachineID = strings.TrimSpace(p.MachineID)
|
||||
p.DMIUUID = strings.TrimSpace(p.DMIUUID)
|
||||
p.ClusterName = strings.TrimSpace(p.ClusterName)
|
||||
p.Hostname = NormalizeHostname(p.Hostname)
|
||||
return p
|
||||
}
|
||||
|
||||
func (p ResourceIdentityPin) hasStrongKey() bool {
|
||||
return p.MachineID != "" || p.DMIUUID != "" || (p.ClusterName != "" && p.Hostname != "")
|
||||
}
|
||||
|
||||
// EraIDs returns every canonical ID this pin's identity keys derive under the
|
||||
// historical chooseNewID ladder (machine > dmi > cluster+hostname > hostname).
|
||||
// Change-journal rows written in boots that only knew a weaker key sit under
|
||||
// the weaker key's hash; expanding a read to the full era set merges those
|
||||
// journal eras without rewriting history.
|
||||
func (p ResourceIdentityPin) EraIDs() []string {
|
||||
p = p.normalized()
|
||||
ids := make([]string, 0, 5)
|
||||
if p.CanonicalID != "" {
|
||||
ids = append(ids, p.CanonicalID)
|
||||
}
|
||||
if p.MachineID != "" {
|
||||
ids = append(ids, buildHashID(p.ResourceType, "machine:"+p.MachineID))
|
||||
}
|
||||
if p.DMIUUID != "" {
|
||||
ids = append(ids, buildHashID(p.ResourceType, "dmi:"+p.DMIUUID))
|
||||
}
|
||||
if p.Hostname != "" {
|
||||
if p.ClusterName != "" {
|
||||
ids = append(ids, buildHashID(p.ResourceType, fmt.Sprintf("cluster:%s:%s", p.ClusterName, p.Hostname)))
|
||||
}
|
||||
ids = append(ids, buildHashID(p.ResourceType, "hostname:"+p.Hostname))
|
||||
}
|
||||
return uniqueTrimmed(ids...)
|
||||
}
|
||||
|
||||
// identityPinForResource derives the persistable pin for a canonical resource.
|
||||
// Only host resources with at least one strong identity key are pinned; a
|
||||
// hostname-only host has a single possible derivation and needs no pin.
|
||||
func identityPinForResource(resource *Resource) (ResourceIdentityPin, bool) {
|
||||
if resource == nil || CanonicalResourceType(resource.Type) != ResourceTypeAgent {
|
||||
return ResourceIdentityPin{}, false
|
||||
}
|
||||
pin := ResourceIdentityPin{
|
||||
CanonicalID: resource.ID,
|
||||
ResourceType: ResourceTypeAgent,
|
||||
MachineID: resource.Identity.MachineID,
|
||||
DMIUUID: resource.Identity.DMIUUID,
|
||||
ClusterName: resource.Identity.ClusterName,
|
||||
Hostname: firstIdentityHostname(resource.Identity),
|
||||
}.normalized()
|
||||
if pin.CanonicalID == "" || !pin.hasStrongKey() {
|
||||
return ResourceIdentityPin{}, false
|
||||
}
|
||||
return pin, true
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,3 +40,50 @@ func TestSourceSpecificIDCanonicalizesSourceIDWhitespace(t *testing.T) {
|
|||
t.Fatalf("SourceSpecificID should trim source ID whitespace: got %q want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResourceIdentityPinEraIDs(t *testing.T) {
|
||||
pin := ResourceIdentityPin{
|
||||
CanonicalID: buildHashID(ResourceTypeAgent, "machine:machine-1"),
|
||||
ResourceType: ResourceTypeAgent,
|
||||
MachineID: "machine-1",
|
||||
DMIUUID: "dmi-1",
|
||||
ClusterName: "homelab",
|
||||
Hostname: "delly.lan",
|
||||
}
|
||||
|
||||
got := pin.EraIDs()
|
||||
want := []string{
|
||||
buildHashID(ResourceTypeAgent, "machine:machine-1"),
|
||||
buildHashID(ResourceTypeAgent, "dmi:dmi-1"),
|
||||
buildHashID(ResourceTypeAgent, "cluster:homelab:delly"),
|
||||
buildHashID(ResourceTypeAgent, "hostname:delly"),
|
||||
}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("expected %d era IDs, got %d: %v", len(want), len(got), got)
|
||||
}
|
||||
for _, id := range want {
|
||||
found := false
|
||||
for _, eraID := range got {
|
||||
if eraID == id {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("expected era set to include %q, got %v", id, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestResourceIdentityPinEraIDsSkipsWeakOnlyKeys(t *testing.T) {
|
||||
pin := ResourceIdentityPin{
|
||||
CanonicalID: "agent-custom",
|
||||
ResourceType: ResourceTypeAgent,
|
||||
Hostname: "delly",
|
||||
}
|
||||
got := pin.EraIDs()
|
||||
want := []string{"agent-custom", buildHashID(ResourceTypeAgent, "hostname:delly")}
|
||||
if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
|
||||
t.Fatalf("expected era IDs %v, got %v", want, got)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -114,6 +114,7 @@ func (a *MonitorAdapter) replaceRegistry(snapshot models.StateSnapshot, recordsB
|
|||
occurredAt = &occ
|
||||
}
|
||||
recordRegistryChanges(registry.store, before, rebuilt.List(), rebuiltAt, occurredAt, SourcePulseDiff, "")
|
||||
rebuilt.PersistIdentityPins()
|
||||
|
||||
a.mu.Lock()
|
||||
a.registry = rebuilt
|
||||
|
|
@ -294,6 +295,7 @@ func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records
|
|||
registry.IngestRecords(source, records)
|
||||
rebuiltAt := time.Now().UTC()
|
||||
recordRegistryChanges(registry.store, before, registry.List(), rebuiltAt, nil, SourcePlatformEvent, changeSourceAdapterForDataSource(source))
|
||||
registry.PersistIdentityPins()
|
||||
a.mu.Lock()
|
||||
a.lastRebuiltAt = rebuiltAt
|
||||
a.mu.Unlock()
|
||||
|
|
|
|||
|
|
@ -336,3 +336,69 @@ func TestMonitorAdapterGetRecentChangesForwardsToStore(t *testing.T) {
|
|||
t.Fatalf("Kind = %q, want %q", changes[0].Kind, ChangeCommandExecuted)
|
||||
}
|
||||
}
|
||||
|
||||
// The monitor adapter is the durable store-backed registry owner, so its
|
||||
// rebuild paths must persist canonical identity pins: a later boot window
|
||||
// (agent not yet checked in) relies on them to derive the same canonical host
|
||||
// ID it used in steady state.
|
||||
func TestMonitorAdapterRebuildPersistsIdentityPins(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
adapter := NewMonitorAdapter(NewRegistry(store))
|
||||
|
||||
const machineID = "7d465a78-test-machine-id"
|
||||
adapter.PopulateFromSnapshot(models.StateSnapshot{
|
||||
Nodes: []models.Node{{
|
||||
ID: "homelab-delly",
|
||||
Name: "delly",
|
||||
Instance: "homelab",
|
||||
ClusterName: "homelab",
|
||||
Status: "online",
|
||||
}},
|
||||
Hosts: []models.Host{{
|
||||
ID: machineID,
|
||||
MachineID: machineID,
|
||||
Hostname: "delly",
|
||||
LinkedNodeID: "homelab-delly",
|
||||
}},
|
||||
LastUpdate: time.Now().UTC(),
|
||||
})
|
||||
|
||||
pins, err := store.ListResourceIdentityPins()
|
||||
if err != nil {
|
||||
t.Fatalf("list identity pins: %v", err)
|
||||
}
|
||||
steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID)
|
||||
found := false
|
||||
for _, pin := range pins {
|
||||
if pin.CanonicalID == steadyID && pin.MachineID == machineID {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("expected snapshot rebuild to persist a machine-keyed identity pin, got %+v", pins)
|
||||
}
|
||||
|
||||
// A restart boot window rebuilds from a snapshot that does not contain
|
||||
// the agent host yet; the pinned identity must keep the canonical ID.
|
||||
adapter.PopulateFromSnapshot(models.StateSnapshot{
|
||||
Nodes: []models.Node{{
|
||||
ID: "homelab-delly",
|
||||
Name: "delly",
|
||||
Instance: "homelab",
|
||||
ClusterName: "homelab",
|
||||
Status: "online",
|
||||
}},
|
||||
LastUpdate: time.Now().UTC(),
|
||||
})
|
||||
|
||||
for _, resource := range adapter.GetAll() {
|
||||
if resource.Proxmox != nil && resource.Proxmox.NodeName == "delly" {
|
||||
if resource.ID != steadyID {
|
||||
t.Fatalf("boot-window rebuild minted %q, want pinned %q", resource.ID, steadyID)
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("expected delly node resource in boot-window rebuild")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,14 +40,15 @@ type IngestRecord struct {
|
|||
|
||||
// ResourceRegistry merges resources from multiple sources.
|
||||
type ResourceRegistry struct {
|
||||
mu sync.RWMutex
|
||||
resources map[string]*Resource
|
||||
bySource map[DataSource]map[string]string
|
||||
matcher *IdentityMatcher
|
||||
store ResourceStore
|
||||
links []ResourceLink
|
||||
exclusions map[string]struct{}
|
||||
pbsBackups []models.PBSBackup
|
||||
mu sync.RWMutex
|
||||
resources map[string]*Resource
|
||||
bySource map[DataSource]map[string]string
|
||||
matcher *IdentityMatcher
|
||||
store ResourceStore
|
||||
links []ResourceLink
|
||||
exclusions map[string]struct{}
|
||||
identityPins *identityPinIndex
|
||||
pbsBackups []models.PBSBackup
|
||||
|
||||
// Cached typed view indexes. Invalidated on ingest, rebuilt lazily on
|
||||
// first access. Protected by mu — callers hold RLock to read, and the
|
||||
|
|
@ -92,6 +93,7 @@ func NewRegistry(store ResourceStore) *ResourceRegistry {
|
|||
rr.bySource[SourceAvailability] = make(map[string]string)
|
||||
|
||||
rr.loadOverrides()
|
||||
rr.loadIdentityPins()
|
||||
return rr
|
||||
}
|
||||
|
||||
|
|
@ -2187,8 +2189,13 @@ func (rr *ResourceRegistry) ingest(source DataSource, sourceID string, resource
|
|||
rr.bySource[source] = make(map[string]string)
|
||||
}
|
||||
|
||||
resource.Identity = identity
|
||||
resource.Type = CanonicalResourceType(resource.Type)
|
||||
// Complete weak host identities from the durable pins before matching and
|
||||
// ID derivation, so canonical IDs do not depend on which sources happen to
|
||||
// be present in this rebuild (boot windows ingest the Proxmox node record
|
||||
// before the agent has checked in).
|
||||
identity = rr.completeIdentityFromPins(resource.Type, identity)
|
||||
resource.Identity = identity
|
||||
resource.Sources = []DataSource{source}
|
||||
resource.SourceStatus = map[DataSource]SourceStatus{
|
||||
source: {Status: "online", LastSeen: resource.LastSeen},
|
||||
|
|
|
|||
|
|
@ -4378,3 +4378,76 @@ func TestMergeMetric_StaleSourceDoesNotClobberLive(t *testing.T) {
|
|||
t.Fatalf("expected live proxmox CPU to survive stale agent merge, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// The merged-source host shape from the homelab canonical-ID bug: a PVE node
|
||||
// record that only knows cluster+hostname, and a pulse-agent record that
|
||||
// knows the machine ID. Whichever record mints the canonical resource decides
|
||||
// which identity key gets hashed, so without durable identity pins the
|
||||
// canonical ID flips between boot windows (node-only) and steady state
|
||||
// (agent present), fragmenting the change journal into per-boot eras.
|
||||
func mergedHostNodeResource() Resource {
|
||||
return Resource{
|
||||
Type: ResourceTypeAgent,
|
||||
Name: "delly",
|
||||
Status: StatusOnline,
|
||||
Proxmox: &ProxmoxData{SourceID: "homelab-delly", NodeName: "delly", ClusterName: "homelab"},
|
||||
}
|
||||
}
|
||||
|
||||
func mergedHostAgentResource(machineID string) Resource {
|
||||
return Resource{
|
||||
Type: ResourceTypeAgent,
|
||||
Name: "delly",
|
||||
Status: StatusOnline,
|
||||
Agent: &AgentData{AgentID: machineID, Hostname: "delly", MachineID: machineID},
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergedHostCanonicalIDStableAcrossRestartsAndIngestOrders(t *testing.T) {
|
||||
store := NewMemoryStore()
|
||||
|
||||
const machineID = "7d465a78-test-machine-id"
|
||||
nodeIdentity := ResourceIdentity{Hostnames: []string{"delly"}, ClusterName: "homelab"}
|
||||
agentIdentity := ResourceIdentity{MachineID: machineID, Hostnames: []string{"delly"}}
|
||||
// IngestSnapshot merges the linked host's identity into the node record
|
||||
// when the agent is present in the snapshot.
|
||||
steadyIdentity := mergeIdentity(nodeIdentity, agentIdentity)
|
||||
|
||||
steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID)
|
||||
bootEraID := buildHashID(ResourceTypeAgent, "cluster:homelab:delly")
|
||||
if steadyID == bootEraID {
|
||||
t.Fatalf("test setup broken: era IDs must differ")
|
||||
}
|
||||
|
||||
// Steady-state boot: node record carries the merged identity and mints
|
||||
// the machine-keyed canonical ID; the agent record merges into it.
|
||||
steady := NewRegistry(store)
|
||||
if id := steady.ingest(SourceProxmox, "homelab-delly", mergedHostNodeResource(), steadyIdentity); id != steadyID {
|
||||
t.Fatalf("steady-state node ingest minted %q, want machine-keyed %q", id, steadyID)
|
||||
}
|
||||
if id := steady.ingest(SourceAgent, machineID, mergedHostAgentResource(machineID), agentIdentity); id != steadyID {
|
||||
t.Fatalf("steady-state agent ingest resolved %q, want %q", id, steadyID)
|
||||
}
|
||||
steady.PersistIdentityPins()
|
||||
|
||||
// Restart into a boot window: the agent has not checked in yet, so the
|
||||
// node record only knows cluster+hostname. Before identity pins this
|
||||
// minted bootEraID and the change journal fragmented into a second era.
|
||||
bootWindow := NewRegistry(store)
|
||||
if id := bootWindow.ingest(SourceProxmox, "homelab-delly", mergedHostNodeResource(), nodeIdentity); id != steadyID {
|
||||
t.Fatalf("boot-window node ingest minted %q, want pinned %q", id, steadyID)
|
||||
}
|
||||
|
||||
// Restart with the opposite ingest order: agent record first into an
|
||||
// empty registry, node record after.
|
||||
reversed := NewRegistry(store)
|
||||
if id := reversed.ingest(SourceAgent, machineID, mergedHostAgentResource(machineID), agentIdentity); id != steadyID {
|
||||
t.Fatalf("agent-first ingest minted %q, want %q", id, steadyID)
|
||||
}
|
||||
if id := reversed.ingest(SourceProxmox, "homelab-delly", mergedHostNodeResource(), nodeIdentity); id != steadyID {
|
||||
t.Fatalf("node-after-agent ingest resolved %q, want %q", id, steadyID)
|
||||
}
|
||||
if got := len(reversed.List()); got != 1 {
|
||||
t.Fatalf("expected one merged resource after reversed-order ingest, got %d", got)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,6 +24,10 @@ type ResourceStore interface {
|
|||
AddExclusion(exclusion ResourceExclusion) error
|
||||
GetLinks() ([]ResourceLink, error)
|
||||
GetExclusions() ([]ResourceExclusion, error)
|
||||
// Identity pins keep canonical IDs for merged-source hosts stable across
|
||||
// restarts. See ResourceIdentityPin.
|
||||
UpsertResourceIdentityPins(pins []ResourceIdentityPin) error
|
||||
ListResourceIdentityPins() ([]ResourceIdentityPin, error)
|
||||
RecordChange(change ResourceChange) error
|
||||
GetRecentChanges(canonicalID string, since time.Time, limit int) ([]ResourceChange, error)
|
||||
GetRecentChangesFiltered(canonicalID string, since time.Time, limit int, filters ResourceChangeFilters) ([]ResourceChange, error)
|
||||
|
|
@ -95,6 +99,12 @@ type SQLiteResourceStore struct {
|
|||
resourceChangesHasSource bool
|
||||
resourceChangesObservedAtNeedsFallback bool
|
||||
mu sync.Mutex
|
||||
|
||||
// identityPinCache caches the (tiny) resource_identities table for
|
||||
// change-journal era expansion. Invalidated on pin upserts.
|
||||
identityPinMu sync.Mutex
|
||||
identityPinCache []ResourceIdentityPin
|
||||
identityPinFresh bool
|
||||
}
|
||||
|
||||
const (
|
||||
|
|
@ -432,6 +442,9 @@ func (s *SQLiteResourceStore) initSchema() error {
|
|||
if err := s.ensureResourceChangesIndexes(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.migrateResourceIdentitiesSchema(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.migrateActionAuditsSchema(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -441,6 +454,33 @@ func (s *SQLiteResourceStore) initSchema() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// migrateResourceIdentitiesSchema upgrades the resource_identities table from
|
||||
// its original schema-only shape (no code ever wrote it before identity pins)
|
||||
// to the pin layout: a cluster_name column and a unique canonical_id index so
|
||||
// pins upsert one row per canonical resource.
|
||||
func (s *SQLiteResourceStore) migrateResourceIdentitiesSchema() error {
|
||||
columns, err := s.tableColumns("resource_identities")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, ok := columns["cluster_name"]; !ok {
|
||||
if _, err := s.db.Exec("ALTER TABLE resource_identities ADD COLUMN cluster_name TEXT"); err != nil {
|
||||
return fmt.Errorf("add resource_identities.cluster_name column: %w", err)
|
||||
}
|
||||
}
|
||||
// The table shipped empty for every deployment, but stay defensive: drop
|
||||
// duplicate canonical_id rows before enforcing uniqueness.
|
||||
if _, err := s.db.Exec(`DELETE FROM resource_identities WHERE id NOT IN (
|
||||
SELECT MAX(id) FROM resource_identities GROUP BY canonical_id
|
||||
)`); err != nil {
|
||||
return fmt.Errorf("dedupe resource_identities canonical rows: %w", err)
|
||||
}
|
||||
if _, err := s.db.Exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_resource_identities_canonical_unique ON resource_identities(canonical_id)"); err != nil {
|
||||
return fmt.Errorf("ensure resource_identities canonical unique index: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// migrateActionAuditsSchema adds the verification_outcome_json column to
|
||||
// older action_audits tables so the VerificationOutcome field on
|
||||
// ActionAuditRecord persists across restarts. Records written before the
|
||||
|
|
@ -846,6 +886,151 @@ func (s *SQLiteResourceStore) Close() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *SQLiteResourceStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error {
|
||||
if len(pins) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin identity pin upsert: %w", err)
|
||||
}
|
||||
committed := false
|
||||
defer func() {
|
||||
if !committed {
|
||||
_ = tx.Rollback()
|
||||
}
|
||||
}()
|
||||
|
||||
for _, pin := range pins {
|
||||
pin = pin.normalized()
|
||||
if pin.CanonicalID == "" || !pin.hasStrongKey() {
|
||||
continue
|
||||
}
|
||||
// A strong identity key can only belong to one canonical resource.
|
||||
// When a key moves (host reinstalled, cluster slot re-occupied by a
|
||||
// different machine), the new pin claims it and stale rows go.
|
||||
if _, err := tx.Exec(`DELETE FROM resource_identities WHERE canonical_id != ? AND (
|
||||
(machine_id IS NOT NULL AND machine_id = ?)
|
||||
OR (dmi_uuid IS NOT NULL AND dmi_uuid = ?)
|
||||
OR (cluster_name IS NOT NULL AND cluster_name != '' AND cluster_name = ? AND primary_hostname = ?)
|
||||
)`,
|
||||
pin.CanonicalID,
|
||||
nullIfEmptyArg(pin.MachineID),
|
||||
nullIfEmptyArg(pin.DMIUUID),
|
||||
pin.ClusterName,
|
||||
pin.Hostname,
|
||||
); err != nil {
|
||||
return fmt.Errorf("clear conflicting identity pins for %q: %w", pin.CanonicalID, err)
|
||||
}
|
||||
if _, err := tx.Exec(`INSERT INTO resource_identities (canonical_id, resource_type, machine_id, dmi_uuid, cluster_name, primary_hostname)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(canonical_id) DO UPDATE SET
|
||||
resource_type = excluded.resource_type,
|
||||
machine_id = COALESCE(excluded.machine_id, resource_identities.machine_id),
|
||||
dmi_uuid = COALESCE(excluded.dmi_uuid, resource_identities.dmi_uuid),
|
||||
cluster_name = COALESCE(NULLIF(excluded.cluster_name, ''), resource_identities.cluster_name),
|
||||
primary_hostname = COALESCE(NULLIF(excluded.primary_hostname, ''), resource_identities.primary_hostname),
|
||||
updated_at = CURRENT_TIMESTAMP`,
|
||||
pin.CanonicalID,
|
||||
string(pin.ResourceType),
|
||||
nullIfEmptyArg(pin.MachineID),
|
||||
nullIfEmptyArg(pin.DMIUUID),
|
||||
pin.ClusterName,
|
||||
pin.Hostname,
|
||||
); err != nil {
|
||||
return fmt.Errorf("upsert identity pin for %q: %w", pin.CanonicalID, err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit identity pin upsert: %w", err)
|
||||
}
|
||||
committed = true
|
||||
|
||||
s.identityPinMu.Lock()
|
||||
s.identityPinFresh = false
|
||||
s.identityPinMu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func nullIfEmptyArg(value string) any {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return nil
|
||||
}
|
||||
return strings.TrimSpace(value)
|
||||
}
|
||||
|
||||
func (s *SQLiteResourceStore) ListResourceIdentityPins() ([]ResourceIdentityPin, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.queryResourceIdentityPins()
|
||||
}
|
||||
|
||||
func (s *SQLiteResourceStore) queryResourceIdentityPins() ([]ResourceIdentityPin, error) {
|
||||
rows, err := s.db.Query(`SELECT canonical_id, resource_type, COALESCE(machine_id, ''), COALESCE(dmi_uuid, ''), COALESCE(cluster_name, ''), COALESCE(primary_hostname, '') FROM resource_identities`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query identity pins: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var pins []ResourceIdentityPin
|
||||
for rows.Next() {
|
||||
var pin ResourceIdentityPin
|
||||
var resourceType string
|
||||
if err := rows.Scan(&pin.CanonicalID, &resourceType, &pin.MachineID, &pin.DMIUUID, &pin.ClusterName, &pin.Hostname); err != nil {
|
||||
return nil, fmt.Errorf("scan identity pin row: %w", err)
|
||||
}
|
||||
pin.ResourceType = ResourceType(resourceType)
|
||||
pins = append(pins, pin.normalized())
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("iterate identity pin rows: %w", err)
|
||||
}
|
||||
return pins, nil
|
||||
}
|
||||
|
||||
// resourceChangeIDSet expands a canonical ID to the set of IDs its journal
|
||||
// rows may have been recorded under. Canonical IDs minted before the host's
|
||||
// strongest identity key was known hash a weaker key (cluster+hostname during
|
||||
// a boot window vs machine ID in steady state); the pinned identity makes
|
||||
// every era's ID recomputable, so reads merge the eras without rewriting
|
||||
// history. Unknown IDs expand to themselves.
|
||||
func (s *SQLiteResourceStore) resourceChangeIDSet(canonicalID string) []string {
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
if canonicalID == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
s.identityPinMu.Lock()
|
||||
if !s.identityPinFresh {
|
||||
pins, err := s.queryResourceIdentityPins()
|
||||
if err == nil {
|
||||
s.identityPinCache = pins
|
||||
s.identityPinFresh = true
|
||||
}
|
||||
}
|
||||
pins := s.identityPinCache
|
||||
s.identityPinMu.Unlock()
|
||||
|
||||
return expandResourceChangeIDs(canonicalID, pins)
|
||||
}
|
||||
|
||||
func expandResourceChangeIDs(canonicalID string, pins []ResourceIdentityPin) []string {
|
||||
for _, pin := range pins {
|
||||
eraIDs := pin.EraIDs()
|
||||
for _, eraID := range eraIDs {
|
||||
if eraID == canonicalID {
|
||||
return eraIDs
|
||||
}
|
||||
}
|
||||
}
|
||||
return []string{canonicalID}
|
||||
}
|
||||
|
||||
func (s *SQLiteResourceStore) RecordChange(change ResourceChange) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
|
@ -921,7 +1106,7 @@ func (s *SQLiteResourceStore) GetRecentChangesFiltered(canonicalID string, since
|
|||
conditions := []string{}
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
if canonicalID != "" {
|
||||
conditions, args = appendRecentChangeResourceCondition(conditions, args, canonicalID, filters.IncludeRelated)
|
||||
conditions, args = appendRecentChangeResourceCondition(conditions, args, s.resourceChangeIDSet(canonicalID), filters.IncludeRelated)
|
||||
} else {
|
||||
conditions = append(conditions, observedAtExpr+" >= ?")
|
||||
args = append(args, since)
|
||||
|
|
@ -1047,7 +1232,7 @@ func (s *SQLiteResourceStore) CountRecentChanges(canonicalID string, since time.
|
|||
|
||||
func (s *SQLiteResourceStore) CountRecentChangesFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (int, error) {
|
||||
query, args := buildRecentChangeCountQuery(
|
||||
canonicalID,
|
||||
s.resourceChangeIDSet(canonicalID),
|
||||
since,
|
||||
filters,
|
||||
"SELECT COUNT(*) FROM resource_changes",
|
||||
|
|
@ -1072,7 +1257,7 @@ func (s *SQLiteResourceStore) CountRecentChangesByKind(canonicalID string, since
|
|||
|
||||
func (s *SQLiteResourceStore) CountRecentChangesByKindFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeKind]int, error) {
|
||||
query, args := buildRecentChangeCountQuery(
|
||||
canonicalID,
|
||||
s.resourceChangeIDSet(canonicalID),
|
||||
since,
|
||||
filters,
|
||||
"SELECT COALESCE(kind, ''), COUNT(*) FROM resource_changes",
|
||||
|
|
@ -1116,7 +1301,7 @@ func (s *SQLiteResourceStore) CountRecentChangesBySourceType(canonicalID string,
|
|||
func (s *SQLiteResourceStore) CountRecentChangesBySourceTypeFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeSourceType]int, error) {
|
||||
sourceTypeExpr := s.resourceChangesSourceTypeExpr()
|
||||
query, args := buildRecentChangeCountQuery(
|
||||
canonicalID,
|
||||
s.resourceChangeIDSet(canonicalID),
|
||||
since,
|
||||
filters,
|
||||
"SELECT "+sourceTypeExpr+", COUNT(*) FROM resource_changes",
|
||||
|
|
@ -1160,7 +1345,7 @@ func (s *SQLiteResourceStore) CountRecentChangesBySourceAdapter(canonicalID stri
|
|||
func (s *SQLiteResourceStore) CountRecentChangesBySourceAdapterFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeSourceAdapter]int, error) {
|
||||
sourceAdapterExpr := s.resourceChangesSourceAdapterExpr()
|
||||
query, args := buildRecentChangeCountQuery(
|
||||
canonicalID,
|
||||
s.resourceChangeIDSet(canonicalID),
|
||||
since,
|
||||
filters,
|
||||
"SELECT "+sourceAdapterExpr+", COUNT(*) FROM resource_changes",
|
||||
|
|
@ -2005,15 +2190,78 @@ type MemoryStore struct {
|
|||
exportAudits []ExportAuditRecord
|
||||
resourceOperatorState map[string]ResourceOperatorState
|
||||
loopReports map[string]LoopReport
|
||||
identityPins map[string]ResourceIdentityPin
|
||||
}
|
||||
|
||||
func NewMemoryStore() *MemoryStore {
|
||||
return &MemoryStore{
|
||||
resourceOperatorState: make(map[string]ResourceOperatorState),
|
||||
loopReports: make(map[string]LoopReport),
|
||||
identityPins: make(map[string]ResourceIdentityPin),
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MemoryStore) UpsertResourceIdentityPins(pins []ResourceIdentityPin) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
for _, pin := range pins {
|
||||
pin = pin.normalized()
|
||||
if pin.CanonicalID == "" || !pin.hasStrongKey() {
|
||||
continue
|
||||
}
|
||||
// Mirror the SQLite conflict semantics: a strong identity key belongs
|
||||
// to exactly one canonical resource.
|
||||
for canonicalID, existing := range m.identityPins {
|
||||
if canonicalID == pin.CanonicalID {
|
||||
continue
|
||||
}
|
||||
if (pin.MachineID != "" && existing.MachineID == pin.MachineID) ||
|
||||
(pin.DMIUUID != "" && existing.DMIUUID == pin.DMIUUID) ||
|
||||
(pin.ClusterName != "" && existing.ClusterName == pin.ClusterName && existing.Hostname == pin.Hostname) {
|
||||
delete(m.identityPins, canonicalID)
|
||||
}
|
||||
}
|
||||
if existing, ok := m.identityPins[pin.CanonicalID]; ok {
|
||||
if pin.MachineID == "" {
|
||||
pin.MachineID = existing.MachineID
|
||||
}
|
||||
if pin.DMIUUID == "" {
|
||||
pin.DMIUUID = existing.DMIUUID
|
||||
}
|
||||
if pin.ClusterName == "" {
|
||||
pin.ClusterName = existing.ClusterName
|
||||
}
|
||||
if pin.Hostname == "" {
|
||||
pin.Hostname = existing.Hostname
|
||||
}
|
||||
}
|
||||
m.identityPins[pin.CanonicalID] = pin
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MemoryStore) ListResourceIdentityPins() ([]ResourceIdentityPin, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
pins := make([]ResourceIdentityPin, 0, len(m.identityPins))
|
||||
for _, pin := range m.identityPins {
|
||||
pins = append(pins, pin)
|
||||
}
|
||||
return pins, nil
|
||||
}
|
||||
|
||||
func (m *MemoryStore) resourceChangeIDSetLocked(canonicalID string) []string {
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
if canonicalID == "" {
|
||||
return nil
|
||||
}
|
||||
pins := make([]ResourceIdentityPin, 0, len(m.identityPins))
|
||||
for _, pin := range m.identityPins {
|
||||
pins = append(pins, pin)
|
||||
}
|
||||
return expandResourceChangeIDs(canonicalID, pins)
|
||||
}
|
||||
|
||||
func (m *MemoryStore) AddLink(link ResourceLink) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
|
@ -2078,10 +2326,11 @@ func (m *MemoryStore) GetRecentChangesFiltered(canonicalID string, since time.Ti
|
|||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
idSet := m.resourceChangeIDSetLocked(canonicalID)
|
||||
var out []ResourceChange
|
||||
for i := len(m.changes) - 1; i >= 0; i-- {
|
||||
change := m.changes[i]
|
||||
if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) {
|
||||
if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) {
|
||||
continue
|
||||
}
|
||||
if !since.IsZero() && change.ObservedAt.Before(since) {
|
||||
|
|
@ -2106,9 +2355,10 @@ func (m *MemoryStore) CountRecentChangesFiltered(canonicalID string, since time.
|
|||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
idSet := m.resourceChangeIDSetLocked(canonicalID)
|
||||
count := 0
|
||||
for _, change := range m.changes {
|
||||
if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) {
|
||||
if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) {
|
||||
continue
|
||||
}
|
||||
if !since.IsZero() && change.ObservedAt.Before(since) {
|
||||
|
|
@ -2130,9 +2380,10 @@ func (m *MemoryStore) CountRecentChangesByKindFiltered(canonicalID string, since
|
|||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
idSet := m.resourceChangeIDSetLocked(canonicalID)
|
||||
counts := make(map[ChangeKind]int)
|
||||
for _, change := range m.changes {
|
||||
if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) {
|
||||
if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) {
|
||||
continue
|
||||
}
|
||||
if !since.IsZero() && change.ObservedAt.Before(since) {
|
||||
|
|
@ -2157,9 +2408,10 @@ func (m *MemoryStore) CountRecentChangesBySourceTypeFiltered(canonicalID string,
|
|||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
idSet := m.resourceChangeIDSetLocked(canonicalID)
|
||||
counts := make(map[ChangeSourceType]int)
|
||||
for _, change := range m.changes {
|
||||
if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) {
|
||||
if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) {
|
||||
continue
|
||||
}
|
||||
if !since.IsZero() && change.ObservedAt.Before(since) {
|
||||
|
|
@ -2184,9 +2436,10 @@ func (m *MemoryStore) CountRecentChangesBySourceAdapterFiltered(canonicalID stri
|
|||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
idSet := m.resourceChangeIDSetLocked(canonicalID)
|
||||
counts := make(map[ChangeSourceAdapter]int)
|
||||
for _, change := range m.changes {
|
||||
if canonicalID != "" && !changeMatchesResource(change, canonicalID, filters.IncludeRelated) {
|
||||
if canonicalID != "" && !changeMatchesResource(change, idSet, filters.IncludeRelated) {
|
||||
continue
|
||||
}
|
||||
if !since.IsZero() && change.ObservedAt.Before(since) {
|
||||
|
|
@ -2203,14 +2456,13 @@ func (m *MemoryStore) CountRecentChangesBySourceAdapterFiltered(canonicalID stri
|
|||
return counts, nil
|
||||
}
|
||||
|
||||
func buildRecentChangeCountQuery(canonicalID string, since time.Time, filters ResourceChangeFilters, selectClause string, observedAtExpr string, sourceTypeExpr string, sourceAdapterExpr string) (string, []any) {
|
||||
func buildRecentChangeCountQuery(canonicalIDs []string, since time.Time, filters ResourceChangeFilters, selectClause string, observedAtExpr string, sourceTypeExpr string, sourceAdapterExpr string) (string, []any) {
|
||||
query := selectClause
|
||||
args := []any{}
|
||||
conditions := []string{observedAtExpr + " >= ?"}
|
||||
args = append(args, since)
|
||||
canonicalID = CanonicalResourceID(canonicalID)
|
||||
if canonicalID != "" {
|
||||
conditions, args = appendRecentChangeResourceCondition(conditions, args, canonicalID, filters.IncludeRelated)
|
||||
if len(canonicalIDs) > 0 {
|
||||
conditions, args = appendRecentChangeResourceCondition(conditions, args, canonicalIDs, filters.IncludeRelated)
|
||||
}
|
||||
if len(filters.Kinds) > 0 {
|
||||
placeholders := make([]string, 0, len(filters.Kinds))
|
||||
|
|
@ -2240,26 +2492,44 @@ func buildRecentChangeCountQuery(canonicalID string, since time.Time, filters Re
|
|||
return query, args
|
||||
}
|
||||
|
||||
func appendRecentChangeResourceCondition(conditions []string, args []any, canonicalID string, includeRelated bool) ([]string, []any) {
|
||||
if !includeRelated {
|
||||
return append(conditions, "canonical_id = ?"), append(args, canonicalID)
|
||||
func appendRecentChangeResourceCondition(conditions []string, args []any, canonicalIDs []string, includeRelated bool) ([]string, []any) {
|
||||
placeholders := make([]string, len(canonicalIDs))
|
||||
for i, id := range canonicalIDs {
|
||||
placeholders[i] = "?"
|
||||
args = append(args, id)
|
||||
}
|
||||
return append(conditions, `(canonical_id = ? OR EXISTS (
|
||||
inClause := "(" + strings.Join(placeholders, ", ") + ")"
|
||||
if !includeRelated {
|
||||
return append(conditions, "canonical_id IN "+inClause), args
|
||||
}
|
||||
for _, id := range canonicalIDs {
|
||||
args = append(args, id)
|
||||
}
|
||||
return append(conditions, `(canonical_id IN `+inClause+` OR EXISTS (
|
||||
SELECT 1
|
||||
FROM json_each(CASE WHEN json_valid(resource_changes.related_resources) THEN resource_changes.related_resources ELSE '[]' END)
|
||||
WHERE TRIM(json_each.value) = ?
|
||||
))`), append(args, canonicalID, canonicalID)
|
||||
WHERE TRIM(json_each.value) IN `+inClause+`
|
||||
))`), args
|
||||
}
|
||||
|
||||
func changeMatchesResource(change ResourceChange, canonicalID string, includeRelated bool) bool {
|
||||
if CanonicalResourceID(change.ResourceID) == canonicalID {
|
||||
func changeMatchesResource(change ResourceChange, canonicalIDs []string, includeRelated bool) bool {
|
||||
matches := func(id string) bool {
|
||||
id = CanonicalResourceID(id)
|
||||
for _, canonicalID := range canonicalIDs {
|
||||
if id == canonicalID {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
if matches(change.ResourceID) {
|
||||
return true
|
||||
}
|
||||
if !includeRelated {
|
||||
return false
|
||||
}
|
||||
for _, relatedID := range change.RelatedResources {
|
||||
if CanonicalResourceID(relatedID) == canonicalID {
|
||||
if matches(relatedID) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2308,3 +2308,86 @@ func TestSQLiteResourceStore_ResourceOperatorState_RoundTrips(t *testing.T) {
|
|||
t.Errorf("clear must be idempotent; got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResourceChangeReadsMergeCanonicalIDEras(t *testing.T) {
|
||||
const machineID = "7d465a78-test-machine-id"
|
||||
steadyID := buildHashID(ResourceTypeAgent, "machine:"+machineID)
|
||||
bootEraID := buildHashID(ResourceTypeAgent, "cluster:homelab:delly")
|
||||
|
||||
sqliteStore, err := NewSQLiteResourceStore(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Fatalf("open sqlite store: %v", err)
|
||||
}
|
||||
defer sqliteStore.Close()
|
||||
|
||||
stores := map[string]ResourceStore{
|
||||
"memory": NewMemoryStore(),
|
||||
"sqlite": sqliteStore,
|
||||
}
|
||||
|
||||
for name, store := range stores {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
|
||||
CanonicalID: steadyID,
|
||||
ResourceType: ResourceTypeAgent,
|
||||
MachineID: machineID,
|
||||
ClusterName: "homelab",
|
||||
Hostname: "delly",
|
||||
}}); err != nil {
|
||||
t.Fatalf("upsert pin: %v", err)
|
||||
}
|
||||
|
||||
base := time.Date(2026, 6, 1, 12, 0, 0, 0, time.UTC)
|
||||
record := func(id, resourceID string, at time.Time) {
|
||||
t.Helper()
|
||||
if err := store.RecordChange(ResourceChange{
|
||||
ID: id,
|
||||
ResourceID: resourceID,
|
||||
ObservedAt: at,
|
||||
Kind: ChangeStateTransition,
|
||||
SourceType: SourcePulseDiff,
|
||||
Confidence: ConfidenceHigh,
|
||||
}); err != nil {
|
||||
t.Fatalf("record change %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
record("change-boot-era", bootEraID, base)
|
||||
record("change-steady-era", steadyID, base.Add(time.Hour))
|
||||
record("change-unrelated", "agent-unrelated", base.Add(2*time.Hour))
|
||||
|
||||
changes, err := store.GetRecentChanges(steadyID, time.Time{}, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("get changes by steady ID: %v", err)
|
||||
}
|
||||
if len(changes) != 2 {
|
||||
t.Fatalf("expected both journal eras under the steady ID, got %d changes", len(changes))
|
||||
}
|
||||
|
||||
// Reads keyed by the old era ID resolve through the pin too, so
|
||||
// callers holding a stale canonical ID see the full timeline.
|
||||
changes, err = store.GetRecentChanges(bootEraID, time.Time{}, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("get changes by boot-era ID: %v", err)
|
||||
}
|
||||
if len(changes) != 2 {
|
||||
t.Fatalf("expected both journal eras under the boot-era ID, got %d changes", len(changes))
|
||||
}
|
||||
|
||||
count, err := store.CountRecentChanges(steadyID, time.Time{})
|
||||
if err != nil {
|
||||
t.Fatalf("count changes: %v", err)
|
||||
}
|
||||
if count != 2 {
|
||||
t.Fatalf("expected era-merged count 2, got %d", count)
|
||||
}
|
||||
|
||||
unrelated, err := store.GetRecentChanges("agent-unrelated", time.Time{}, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("get unrelated changes: %v", err)
|
||||
}
|
||||
if len(unrelated) != 1 {
|
||||
t.Fatalf("expected unrelated resource to keep its own timeline, got %d changes", len(unrelated))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue