Derive cluster-scoped canonical host IDs from full hostnames

Sibling of #1559: canonicalIDFromIdentity hashed the short hostname in
its cluster arm, so two machine-keyless Docker Swarm members with FQDN
hostnames sharing a first label (cloud.a, cloud.b) derived the same
cluster:<swarm>:cloud canonical ID and fully merged in the registry.
PVE is unaffected (single-label node names hash identically).

Both hostname-derived arms now hash NormalizeFullHostname. Derivation
stays a pure function of identity so store-less registries keep deriving
the same IDs as the durable one. Compatibility for hosts minted under
the short era: journal reads already merge both eras via
ResourceIdentityPin.EraIDs, and new canonical-ID succession re-keys
resource_operator_state and action_audits rows to the successor ID at
pin persist (never across a contradicting machine key, never while the
old ID is still live, never rewriting journal rows) so operator intent
like never-auto-remediate survives the era change. Already-merged
keyless pairs cannot be retroactively split; the merged rows succeed to
whichever member persists first.
This commit is contained in:
rcourtman
2026-07-11 19:05:14 +01:00
parent a23a8fad9a
commit e5dbbb3145
6 changed files with 549 additions and 5 deletions
@@ -186,11 +186,81 @@ func (rr *ResourceRegistry) completeIdentityFromPins(resourceType ResourceType,
return identity
}
// successionsFor reports the pinned canonical IDs the given (new or changed)
// pin supersedes: pins for the same physical host whose canonical ID was
// minted in an earlier era of the chooseNewID ladder. Same-machine proof is a
// matching strong key; a machine-keyless pin in the same cluster whose
// hostname is the incoming pin's hostname or its collapsed short form is the
// same host re-derived after the short→full hostname derivation fix. A
// machine-keyless old pin claimed by a machine-keyed incoming pin is the
// common "agent gained /etc/machine-id" upgrade. Rows with a contradicting
// machine key (host reinstalled, cluster slot re-occupied by a different
// machine) never succeed: the new machine must mint fresh, not absorb the
// old host's operator state.
func (index *identityPinIndex) successionsFor(pin ResourceIdentityPin) []CanonicalIDSuccession {
if index == nil {
return nil
}
pin = pin.normalized()
if pin.CanonicalID == "" {
return nil
}
var successions []CanonicalIDSuccession
seen := make(map[string]struct{})
consider := func(old ResourceIdentityPin) {
if old.CanonicalID == "" || old.CanonicalID == pin.CanonicalID {
return
}
if _, dup := seen[old.CanonicalID]; dup {
return
}
seen[old.CanonicalID] = struct{}{}
successions = append(successions, CanonicalIDSuccession{
OldCanonicalID: old.CanonicalID,
NewCanonicalID: pin.CanonicalID,
})
}
if pin.MachineID != "" {
if old, ok := index.byMachineID[pin.MachineID]; ok {
consider(old)
}
}
if pin.DMIUUID != "" {
if old, ok := index.byDMIUUID[pin.DMIUUID]; ok {
consider(old)
}
}
if pin.ClusterName != "" && pin.Hostname != "" {
shortHostname := NormalizeHostname(pin.Hostname)
for _, old := range index.byClusterShortHost[clusterHostPinKey(pin.ClusterName, shortHostname)] {
if old.MachineID != "" || old.DMIUUID != "" {
// Machine-keyed rows only succeed through their own keys.
continue
}
if old.Hostname != pin.Hostname && old.Hostname != shortHostname {
// A different FQDN sharing the short name is a different host.
continue
}
consider(old)
}
}
return successions
}
// 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.
//
// When a new pin supersedes an earlier era's pin for the same physical host
// (see successionsFor), the store re-keys the host's operator-owned rows to
// the new canonical ID before the pin write, so operator intent
// (never-auto-remediate, maintenance windows) and action-audit history
// survive the era change. Successions are skipped while the old canonical ID
// still belongs to a live resource: a genuinely short-named host must not
// have its rows stolen by an FQDN sibling. Change-journal rows are never
// rewritten; EraIDs merges those at read time.
func (rr *ResourceRegistry) PersistIdentityPins() {
if rr.store == nil {
return
@@ -198,6 +268,7 @@ func (rr *ResourceRegistry) PersistIdentityPins() {
rr.mu.RLock()
var pins []ResourceIdentityPin
var successions []CanonicalIDSuccession
for _, resource := range rr.resources {
pin, ok := identityPinForResource(resource)
if !ok {
@@ -207,12 +278,25 @@ func (rr *ResourceRegistry) PersistIdentityPins() {
continue
}
pins = append(pins, pin)
for _, succession := range rr.identityPins.successionsFor(pin) {
if _, live := rr.resources[succession.OldCanonicalID]; live {
continue
}
successions = append(successions, succession)
}
}
rr.mu.RUnlock()
if len(pins) == 0 {
return
}
if len(successions) > 0 {
if successor, ok := rr.store.(canonicalIDSuccessor); ok {
if err := successor.ApplyCanonicalIDSuccessions(successions); err != nil {
log.Printf("unifiedresources: failed to apply canonical ID successions: %v", err)
}
}
}
if err := rr.store.UpsertResourceIdentityPins(pins); err != nil {
log.Printf("unifiedresources: failed to persist identity pins: %v", err)
return
@@ -0,0 +1,110 @@
package unifiedresources
import "fmt"
// CanonicalIDSuccession records that the physical host durably pinned under
// OldCanonicalID now mints NewCanonicalID. Canonical IDs hash the strongest
// identity key known at mint time, so a host's ID moves eras when a stronger
// key appears (a containerized agent gains /etc/machine-id) or when the
// derivation itself is corrected (cluster-scoped IDs switched from short to
// full hostnames so FQDN Swarm members stop collapsing). Succession re-keys
// the operator-owned rows, meaning operator state (never-auto-remediate,
// maintenance windows, criticality, notes) and action-audit history, onto
// the new canonical ID and drops the superseded identity pin row.
// Change-journal rows are intentionally left untouched: history is never
// rewritten, and ResourceIdentityPin.EraIDs merges old-era journal rows into
// reads keyed by the new ID.
type CanonicalIDSuccession struct {
OldCanonicalID string
NewCanonicalID string
}
// canonicalIDSuccessor is an optional store capability, mirroring the
// maintenance-lifecycle pattern in resource_operator_state.go. The durable
// registry applies successions during PersistIdentityPins when it detects
// that a new pin supersedes an earlier era's pin for the same physical host.
type canonicalIDSuccessor interface {
ApplyCanonicalIDSuccessions(successions []CanonicalIDSuccession) error
}
// ApplyCanonicalIDSuccessions re-keys operator-owned rows from each
// superseded canonical ID to its successor and deletes the superseded
// identity pin row, in one transaction. Re-keys never clobber rows already
// present under the new ID (those are fresher); a shadowed old row is left
// behind, which matches the pre-succession orphaning behavior.
func (s *SQLiteResourceStore) ApplyCanonicalIDSuccessions(successions []CanonicalIDSuccession) error {
if len(successions) == 0 {
return nil
}
s.mu.Lock()
defer s.mu.Unlock()
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("begin canonical ID succession: %w", err)
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback()
}
}()
for _, succession := range successions {
oldID := CanonicalResourceID(succession.OldCanonicalID)
newID := CanonicalResourceID(succession.NewCanonicalID)
if oldID == "" || newID == "" || oldID == newID {
continue
}
for _, stmt := range []string{
`UPDATE OR IGNORE resource_operator_state SET canonical_id = ? WHERE canonical_id = ?`,
`UPDATE action_audits SET canonical_id = ? WHERE canonical_id = ?`,
} {
if _, err := tx.Exec(stmt, newID, oldID); err != nil {
return fmt.Errorf("re-key canonical rows %q -> %q: %w", oldID, newID, err)
}
}
// The successor pin's EraIDs keep the old era's journal rows
// readable; the superseded pin row itself is stale bucket noise.
if _, err := tx.Exec(`DELETE FROM resource_identities WHERE canonical_id = ?`, oldID); err != nil {
return fmt.Errorf("delete superseded identity pin %q: %w", oldID, err)
}
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("commit canonical ID succession: %w", err)
}
committed = true
s.identityPinMu.Lock()
s.identityPinFresh = false
s.identityPinMu.Unlock()
return nil
}
// ApplyCanonicalIDSuccessions mirrors the SQLite semantics for the in-memory
// test store. Action audits are not re-keyed here: MemoryStore filters them
// by the audit record's embedded request resource ID rather than a separate
// index column, and audit artifacts are never mutated.
func (m *MemoryStore) ApplyCanonicalIDSuccessions(successions []CanonicalIDSuccession) error {
m.mu.Lock()
defer m.mu.Unlock()
for _, succession := range successions {
oldID := CanonicalResourceID(succession.OldCanonicalID)
newID := CanonicalResourceID(succession.NewCanonicalID)
if oldID == "" || newID == "" || oldID == newID {
continue
}
if state, ok := m.resourceOperatorState[oldID]; ok {
if _, taken := m.resourceOperatorState[newID]; !taken {
state.CanonicalID = newID
m.resourceOperatorState[newID] = state
delete(m.resourceOperatorState, oldID)
}
}
delete(m.identityPins, oldID)
}
return nil
}
@@ -0,0 +1,134 @@
package unifiedresources
import (
"testing"
"time"
)
// ApplyCanonicalIDSuccessions must re-key operator state and action-audit
// history rows to the successor canonical ID, drop the superseded identity
// pin row, and never clobber rows already present under the successor.
func TestSQLiteApplyCanonicalIDSuccessions(t *testing.T) {
store, err := NewSQLiteResourceStore(t.TempDir(), "default")
if err != nil {
t.Fatalf("NewSQLiteResourceStore: %v", err)
}
defer store.Close()
oldID := buildHashID(ResourceTypeAgent, "cluster:prod-swarm:cloud")
newID := buildHashID(ResourceTypeAgent, "cluster:prod-swarm:cloud.a")
if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
CanonicalID: oldID,
ResourceType: ResourceTypeAgent,
ClusterName: "prod-swarm",
Hostname: "cloud",
}}); err != nil {
t.Fatalf("seed identity pin: %v", err)
}
if err := store.SetResourceOperatorState(ResourceOperatorState{
CanonicalID: oldID,
NeverAutoRemediate: true,
Note: "flaky PSU, hands off",
}); err != nil {
t.Fatalf("seed operator state: %v", err)
}
// Seed the audit history row directly: only the canonical_id index
// column participates in succession, the audit artifact itself is
// opaque here.
now := time.Now().UTC()
if _, err := store.db.Exec(`INSERT INTO action_audits
(id, action_id, canonical_id, request_id, created_at, updated_at, state, request_json, plan_json)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
"audit-1", "audit-1", oldID, "req-1", now.Add(-time.Hour), now.Add(-time.Hour),
string(ActionStatePending), `{"resourceId":"`+oldID+`"}`, `{}`,
); err != nil {
t.Fatalf("seed action audit: %v", err)
}
if err := store.ApplyCanonicalIDSuccessions([]CanonicalIDSuccession{{
OldCanonicalID: oldID,
NewCanonicalID: newID,
}}); err != nil {
t.Fatalf("ApplyCanonicalIDSuccessions: %v", err)
}
if _, found, err := store.GetResourceOperatorState(oldID); err != nil || found {
t.Fatalf("operator state still keyed by superseded ID (found=%v, err=%v)", found, err)
}
state, found, err := store.GetResourceOperatorState(newID)
if err != nil || !found {
t.Fatalf("operator state missing under successor ID (found=%v, err=%v)", found, err)
}
if !state.NeverAutoRemediate || state.Note != "flaky PSU, hands off" {
t.Fatalf("operator state mutated across succession: %+v", state)
}
oldAudits, err := store.GetActionAudits(oldID, time.Time{}, 10)
if err != nil {
t.Fatalf("GetActionAudits(old): %v", err)
}
if len(oldAudits) != 0 {
t.Fatalf("action audits still keyed by superseded ID: %d rows", len(oldAudits))
}
newAudits, err := store.GetActionAudits(newID, time.Time{}, 10)
if err != nil {
t.Fatalf("GetActionAudits(new): %v", err)
}
if len(newAudits) != 1 || newAudits[0].ID != "audit-1" {
t.Fatalf("action audit history missing under successor ID: %+v", newAudits)
}
pins, err := store.ListResourceIdentityPins()
if err != nil {
t.Fatalf("ListResourceIdentityPins: %v", err)
}
for _, pin := range pins {
if pin.CanonicalID == oldID {
t.Fatalf("superseded identity pin row still present")
}
}
}
// When the successor ID already has an operator-state row, the succession
// must keep it (it is fresher) rather than overwrite it with the superseded
// row.
func TestSQLiteCanonicalIDSuccessionKeepsExistingTargetRow(t *testing.T) {
store, err := NewSQLiteResourceStore(t.TempDir(), "default")
if err != nil {
t.Fatalf("NewSQLiteResourceStore: %v", err)
}
defer store.Close()
oldID := buildHashID(ResourceTypeAgent, "cluster:prod-swarm:cloud")
newID := buildHashID(ResourceTypeAgent, "cluster:prod-swarm:cloud.a")
if err := store.SetResourceOperatorState(ResourceOperatorState{
CanonicalID: oldID,
Note: "stale note from the collapsed era",
}); err != nil {
t.Fatalf("seed old operator state: %v", err)
}
if err := store.SetResourceOperatorState(ResourceOperatorState{
CanonicalID: newID,
NeverAutoRemediate: true,
Note: "fresh note under the successor",
}); err != nil {
t.Fatalf("seed new operator state: %v", err)
}
if err := store.ApplyCanonicalIDSuccessions([]CanonicalIDSuccession{{
OldCanonicalID: oldID,
NewCanonicalID: newID,
}}); err != nil {
t.Fatalf("ApplyCanonicalIDSuccessions: %v", err)
}
state, found, err := store.GetResourceOperatorState(newID)
if err != nil || !found {
t.Fatalf("successor operator state missing (found=%v, err=%v)", found, err)
}
if !state.NeverAutoRemediate || state.Note != "fresh note under the successor" {
t.Fatalf("succession overwrote the fresher successor row: %+v", state)
}
}
+8 -2
View File
@@ -3657,10 +3657,16 @@ func (rr *ResourceRegistry) canonicalIDFromIdentity(resourceType ResourceType, i
stable = "machine:" + strings.TrimSpace(identity.MachineID)
case identity.DMIUUID != "":
stable = "dmi:" + strings.TrimSpace(identity.DMIUUID)
// Hostname-derived keys hash the full dotted name: distinct machines that
// share a short name (cloud.rnd-lax1 vs cloud.gce-or1, or two FQDN Swarm
// members without machine IDs) must mint distinct canonical IDs. IDs
// minted under the historical short-hostname derivation stay readable
// through ResourceIdentityPin.EraIDs (journal reads) and succeed their
// operator-owned rows via canonical-ID succession at pin persist.
case identity.ClusterName != "" && len(identity.Hostnames) > 0:
stable = fmt.Sprintf("cluster:%s:%s", identity.ClusterName, NormalizeHostname(identity.Hostnames[0]))
stable = fmt.Sprintf("cluster:%s:%s", identity.ClusterName, NormalizeFullHostname(identity.Hostnames[0]))
case len(identity.Hostnames) > 0:
stable = "hostname:" + NormalizeHostname(identity.Hostnames[0])
stable = "hostname:" + NormalizeFullHostname(identity.Hostnames[0])
case len(identity.IPAddresses) > 0:
stable = "ip:" + NormalizeIP(identity.IPAddresses[0])
default:
+180
View File
@@ -4921,3 +4921,183 @@ func TestResourceRegistryUsesConfiguredProxmoxStaleThresholds(t *testing.T) {
t.Fatalf("cloned SourceStatus[proxmox] = %+v, want online", status)
}
}
// Two Docker Swarm members with FQDN hostnames and no machine keys
// (containerized agents without /etc/machine-id) must mint distinct
// cluster-scoped canonical IDs. The historical derivation hashed the short
// hostname, so cloud.a and cloud.b both derived cluster:<swarm>:cloud and
// fully merged in the registry (sibling of #1559).
func TestSwarmFQDNHostsWithoutMachineIDsStayDistinct(t *testing.T) {
store := NewMemoryStore()
registry := NewRegistry(store)
swarmHostIdentity := func(hostname string) ResourceIdentity {
return ResourceIdentity{
ClusterName: "prod-swarm",
Hostnames: []string{hostname, "prod-swarm:" + hostname},
}
}
swarmHostResource := func(hostname string) Resource {
return Resource{
Type: ResourceTypeAgent,
Name: hostname,
Status: StatusOnline,
Docker: &DockerData{HostSourceID: "docker:" + hostname, Hostname: hostname},
}
}
idA := registry.ingest(SourceDocker, "docker:cloud.a", swarmHostResource("cloud.a"), swarmHostIdentity("cloud.a"))
idB := registry.ingest(SourceDocker, "docker:cloud.b", swarmHostResource("cloud.b"), swarmHostIdentity("cloud.b"))
if idA == "" || idB == "" {
t.Fatalf("swarm host ingest returned empty IDs: %q, %q", idA, idB)
}
if idA == idB {
t.Fatalf("swarm members cloud.a and cloud.b collapsed to canonical ID %q", idA)
}
if got := len(registry.List()); got != 2 {
t.Fatalf("expected 2 distinct swarm host resources, got %d", got)
}
// Journal rows recorded under the collapsed short-hostname era must stay
// part of the timeline read through the new full-hostname ID.
shortEraID := buildHashID(ResourceTypeAgent, "cluster:prod-swarm:cloud")
if shortEraID == idA || shortEraID == idB {
t.Fatalf("full-hostname derivation still matches the short era ID %q", shortEraID)
}
if err := store.RecordChange(ResourceChange{
ID: "change-short-era",
ResourceID: shortEraID,
ObservedAt: time.Now().Add(-time.Hour),
Kind: ChangeStateTransition,
}); err != nil {
t.Fatalf("RecordChange: %v", err)
}
registry.PersistIdentityPins()
changes, err := store.GetRecentChanges(idA, time.Time{}, 10)
if err != nil {
t.Fatalf("GetRecentChanges: %v", err)
}
found := false
for _, change := range changes {
if change.ID == "change-short-era" {
found = true
}
}
if !found {
t.Fatalf("short-era journal row missing from timeline of %q; got %d changes", idA, len(changes))
}
}
// A machine-keyless swarm host whose canonical ID was minted under the
// collapsed short-hostname derivation must carry its operator-owned rows to
// the full-hostname ID: the superseded pin row goes, operator state
// (never-auto-remediate) is re-keyed, and the new pin preserves the full
// dotted hostname.
func TestCanonicalIDSuccessionMovesOperatorState(t *testing.T) {
store := NewMemoryStore()
shortEraID := buildHashID(ResourceTypeAgent, "cluster:prod-swarm:cloud")
if err := store.UpsertResourceIdentityPins([]ResourceIdentityPin{{
CanonicalID: shortEraID,
ResourceType: ResourceTypeAgent,
ClusterName: "prod-swarm",
Hostname: "cloud",
}}); err != nil {
t.Fatalf("seed identity pin: %v", err)
}
if err := store.SetResourceOperatorState(ResourceOperatorState{
CanonicalID: shortEraID,
NeverAutoRemediate: true,
}); err != nil {
t.Fatalf("seed operator state: %v", err)
}
registry := NewRegistry(store)
identity := ResourceIdentity{
ClusterName: "prod-swarm",
Hostnames: []string{"cloud.a", "prod-swarm:cloud.a"},
}
resource := Resource{
Type: ResourceTypeAgent,
Name: "cloud.a",
Status: StatusOnline,
Docker: &DockerData{HostSourceID: "docker:cloud.a", Hostname: "cloud.a"},
}
newID := registry.ingest(SourceDocker, "docker:cloud.a", resource, identity)
if newID == "" || newID == shortEraID {
t.Fatalf("expected a fresh full-hostname canonical ID, got %q (short era %q)", newID, shortEraID)
}
registry.PersistIdentityPins()
if _, found, err := store.GetResourceOperatorState(shortEraID); err != nil || found {
t.Fatalf("operator state still keyed by superseded ID (found=%v, err=%v)", found, err)
}
state, found, err := store.GetResourceOperatorState(newID)
if err != nil || !found {
t.Fatalf("operator state missing under successor ID (found=%v, err=%v)", found, err)
}
if !state.NeverAutoRemediate {
t.Fatalf("never-auto-remediate flag lost across canonical ID succession")
}
pins, err := store.ListResourceIdentityPins()
if err != nil {
t.Fatalf("ListResourceIdentityPins: %v", err)
}
byID := make(map[string]ResourceIdentityPin, len(pins))
for _, pin := range pins {
byID[pin.CanonicalID] = pin
}
if _, stale := byID[shortEraID]; stale {
t.Fatalf("superseded short-era pin row still present")
}
successor, ok := byID[newID]
if !ok {
t.Fatalf("successor pin missing; pins: %+v", pins)
}
if successor.Hostname != "cloud.a" {
t.Fatalf("successor pin hostname = %q, want full dotted cloud.a", successor.Hostname)
}
}
// A genuinely short-named host that is alive in the registry owns its
// canonical ID; an FQDN sibling sharing the short name must not steal its
// operator rows through succession.
func TestCanonicalIDSuccessionSkipsLiveShortNamedHost(t *testing.T) {
store := NewMemoryStore()
registry := NewRegistry(store)
ingestSwarmHost := func(hostname string) string {
return registry.ingest(SourceDocker, "docker:"+hostname, Resource{
Type: ResourceTypeAgent,
Name: hostname,
Status: StatusOnline,
Docker: &DockerData{HostSourceID: "docker:" + hostname, Hostname: hostname},
}, ResourceIdentity{
ClusterName: "prod-swarm",
Hostnames: []string{hostname, "prod-swarm:" + hostname},
})
}
shortID := ingestSwarmHost("cloud")
registry.PersistIdentityPins()
if err := store.SetResourceOperatorState(ResourceOperatorState{
CanonicalID: shortID,
NeverAutoRemediate: true,
}); err != nil {
t.Fatalf("seed operator state: %v", err)
}
fqdnID := ingestSwarmHost("cloud.a")
if fqdnID == shortID {
t.Fatalf("cloud and cloud.a collapsed to canonical ID %q", fqdnID)
}
registry.PersistIdentityPins()
state, found, err := store.GetResourceOperatorState(shortID)
if err != nil || !found || !state.NeverAutoRemediate {
t.Fatalf("live short-named host lost its operator state (found=%v, err=%v)", found, err)
}
if _, stolen, _ := store.GetResourceOperatorState(fqdnID); stolen {
t.Fatalf("FQDN sibling stole the live short-named host's operator state")
}
}