vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-09 16:00:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Pick cluster-scoped swarm dedupe winners deterministically

Browse source 
In a multi-manager swarm every manager reports the same cluster-scoped
objects (services, nodes, secrets, configs), and the registry dedupe
picked the candidate whose host had the freshest LastSeen. That ordering
flips between polls, so the winner alternated: swarm node resources
re-parented to whichever manager won (ParentID tracks the reporting
host), and services flipped name/status/updateStatus whenever the
managers' views differed slightly. Every rebuild then wrote phantom
relationship_change / state_transition / config_update rows into
resource_changes, the same class of churn as the registry-rebuild spam
fixed in 74131e56e. Mock mode had the same symptom and was fixed by
pinning a single reporting leader, but real deployments still hit this.

Select winners deterministically instead: richer candidate fields first
(unchanged intent), then managers with an available control plane, and
freshness only when the gap exceeds the docker stale threshold, i.e.
the losing manager has genuinely gone quiet; otherwise the lowest host
ID wins as a stable tiebreak. Equal candidates can no longer alternate.

Regression test rebuilds the registry across polls with two managers,
jittered LastSeen ordering, flipped snapshot host order, and slightly
divergent service views, asserting no change emission; it fails against
the old LastSeen-ordering rule with the exact swarm-node re-parenting
seen in production.
This commit is contained in:
rcourtman 2026-07-08 09:35:56 +01:00
parent c30e223704
commit ef2569b1fc
2 changed files with 265 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

114
internal/unifiedresources/registry.go
View file

@ -238,6 +238,13 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
rr.refreshDockerNetworkAttachmentRelationships(dh)
}
// Cluster-scoped swarm objects are reported by every manager; the winner
// must be picked deterministically or the resource re-parents (and can
// swap name/status) whenever LastSeen ordering flips between polls,
// spamming phantom resource_changes rows. Freshness only breaks ties with
// a full stale threshold of hysteresis; see preferDockerSwarmHost.
dockerStaleThreshold := effectiveStaleThresholds(thresholds)[SourceDocker]
type dockerSecretCandidate struct {
host models.DockerHost
secret models.DockerSecret
@ -258,11 +265,10 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
continue
}
replace := false
if existing.secret.DriverName == "" && secret.DriverName != "" {
replace = true
}
if !replace && dh.LastSeen.After(existing.host.LastSeen) {
replace = true
if next, cur := dockerSecretRichness(secret), dockerSecretRichness(existing.secret); next != cur {
replace = next > cur
} else {
replace = preferDockerSwarmHost(dh, existing.host, dockerStaleThreshold)
}
if replace {
secretByID[sourceID] = dockerSecretCandidate{host: dh, secret: secret}
@ -293,11 +299,10 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
continue
}
replace := false
if existing.config.TemplatingDriver == "" && config.TemplatingDriver != "" {
replace = true
}
if !replace && dh.LastSeen.After(existing.host.LastSeen) {
replace = true
if next, cur := dockerConfigRichness(config), dockerConfigRichness(existing.config); next != cur {
replace = next > cur
} else {
replace = preferDockerSwarmHost(dh, existing.host, dockerStaleThreshold)
}
if replace {
configByID[sourceID] = dockerConfigCandidate{host: dh, config: config}
@ -329,16 +334,12 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
serviceByID[sourceID] = dockerServiceCandidate{host: dh, service: svc}
continue
}
// Prefer candidates with richer fields and fresher host timestamps.
// Prefer candidates with richer fields, then a deterministic host.
replace := false
if existing.service.Image == "" && svc.Image != "" {
replace = true
}
if existing.service.UpdateStatus == nil && svc.UpdateStatus != nil {
replace = true
}
if !replace && dh.LastSeen.After(existing.host.LastSeen) {
replace = true
if next, cur := dockerServiceRichness(svc), dockerServiceRichness(existing.service); next != cur {
replace = next > cur
} else {
replace = preferDockerSwarmHost(dh, existing.host, dockerStaleThreshold)
}
if replace {
serviceByID[sourceID] = dockerServiceCandidate{host: dh, service: svc}
@ -369,14 +370,10 @@ func (rr *ResourceRegistry) ingestSnapshot(snapshot models.StateSnapshot, thresh
continue
}
replace := false
if existing.node.EngineVersion == "" && node.EngineVersion != "" {
replace = true
}
if existing.node.ManagerReachability == "" && node.ManagerReachability != "" {
replace = true
}
if !replace && dh.LastSeen.After(existing.host.LastSeen) {
replace = true
if next, cur := dockerSwarmNodeRichness(node), dockerSwarmNodeRichness(existing.node); next != cur {
replace = next > cur
} else {
replace = preferDockerSwarmHost(dh, existing.host, dockerStaleThreshold)
}
if replace {
nodeByID[sourceID] = dockerNodeCandidate{host: dh, node: node}
@ -3987,6 +3984,69 @@ func dockerSwarmClusterKey(host models.DockerHost) string {
return ""
}
// preferDockerSwarmHost reports whether next should replace current as the
// reporting host for a cluster-scoped swarm object (service, node, secret,
// config). Selection must be deterministic across polls: in a multi-manager
// swarm every manager reports the same objects, and if the winner tracked raw
// LastSeen ordering it would alternate between polls, re-parenting the
// resource each rebuild and writing phantom resource_changes rows. Managers
// with an available control plane win first; freshness only counts when the
// gap exceeds the source stale threshold (i.e. the loser has genuinely gone
// quiet); otherwise the lowest host ID wins as a stable tiebreak.
func preferDockerSwarmHost(next, current models.DockerHost, staleThreshold time.Duration) bool {
nextManager := next.Swarm != nil && next.Swarm.ControlAvailable
currentManager := current.Swarm != nil && current.Swarm.ControlAvailable
if nextManager != currentManager {
return nextManager
}
if staleThreshold > 0 {
gap := next.LastSeen.Sub(current.LastSeen)
if gap > staleThreshold {
return true
}
if gap < -staleThreshold {
return false
}
}
return strings.TrimSpace(next.ID) < strings.TrimSpace(current.ID)
}
func dockerServiceRichness(service models.DockerService) int {
score := 0
if service.Image != "" {
score++
}
if service.UpdateStatus != nil {
score++
}
return score
}
func dockerSwarmNodeRichness(node models.DockerNode) int {
score := 0
if node.EngineVersion != "" {
score++
}
if node.ManagerReachability != "" {
score++
}
return score
}
func dockerSecretRichness(secret models.DockerSecret) int {
if secret.DriverName != "" {
return 1
}
return 0
}
func dockerConfigRichness(config models.DockerConfig) int {
if config.TemplatingDriver != "" {
return 1
}
return 0
}
func dockerServiceSourceID(host models.DockerHost, service models.DockerService) string {
cluster := dockerSwarmClusterKey(host)
if cluster == "" {

178
internal/unifiedresources/swarm_dedupe_test.go Normal file
View file

@ -0,0 +1,178 @@
package unifiedresources
import (
"testing"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
)
// swarmTwoManagerSnapshot builds a snapshot where two swarm managers report
// the same cluster-scoped objects (service, nodes, secret, config). The
// managers' service views differ slightly (manager-2 lags one running task),
// so an alternating dedupe winner would flip the service status between
// online and warning in addition to re-parenting the swarm node resources.
func swarmTwoManagerSnapshot(m1LastSeen, m2LastSeen time.Time, flipOrder bool) models.StateSnapshot {
swarmInfo := func(nodeID string) *models.DockerSwarmInfo {
return &models.DockerSwarmInfo{
NodeID: nodeID,
NodeRole: "manager",
LocalState: "active",
ControlAvailable: true,
ClusterID: "swarm-cluster-1",
ClusterName: "prod-swarm",
}
}
nodes := []models.DockerNode{
{ID: "node-m1", Hostname: "manager-1", Role: "manager", State: "ready", ManagerReachability: "reachable", EngineVersion: "27.0.1"},
{ID: "node-m2", Hostname: "manager-2", Role: "manager", State: "ready", ManagerReachability: "reachable", EngineVersion: "27.0.1"},
}
secrets := []models.DockerSecret{{ID: "secret-1", Name: "db-password"}}
configs := []models.DockerConfig{{ID: "config-1", Name: "app-config"}}
serviceSeenByM1 := models.DockerService{
ID: "svc-1",
Name: "web",
Image: "nginx:1.27",
Mode: "replicated",
DesiredTasks: 3,
RunningTasks: 3,
}
serviceSeenByM2 := serviceSeenByM1
serviceSeenByM2.RunningTasks = 2
m1 := models.DockerHost{
ID: "dockerhost-m1",
AgentID: "agent-m1",
Hostname: "manager-1",
Status: "online",
LastSeen: m1LastSeen,
Swarm: swarmInfo("node-m1"),
Services: []models.DockerService{serviceSeenByM1},
Nodes: nodes,
Secrets: secrets,
Configs: configs,
}
m2 := models.DockerHost{
ID: "dockerhost-m2",
AgentID: "agent-m2",
Hostname: "manager-2",
Status: "online",
LastSeen: m2LastSeen,
Swarm: swarmInfo("node-m2"),
Services: []models.DockerService{serviceSeenByM2},
Nodes: nodes,
Secrets: secrets,
Configs: configs,
}
hosts := []models.DockerHost{m1, m2}
if flipOrder {
hosts = []models.DockerHost{m2, m1}
}
return models.StateSnapshot{DockerHosts: hosts}
}
// Every state update rebuilds the registry from scratch, and in a
// multi-manager swarm the reporting hosts' LastSeen ordering flips between
// polls. The cluster-scoped dedupe winner must not track that jitter: an
// alternating winner re-parents the swarm node resources and flips the
// service status with the managers' slightly divergent views, writing
// phantom resource_changes rows on every poll.
func TestSwarmClusterScopedDedupe_NoChangeEmissionAcrossLastSeenJitter(t *testing.T) {
base := time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC)
build := func(m1LastSeen, m2LastSeen time.Time, flipOrder bool) []Resource {
rr := NewRegistry(nil)
rr.IngestSnapshot(swarmTwoManagerSnapshot(m1LastSeen, m2LastSeen, flipOrder))
return rr.List()
}
previous := build(base, base.Add(2*time.Second), false)
if len(previous) == 0 {
t.Fatal("expected resources from swarm snapshot, got none")
}
for poll := 1; poll <= 6; poll++ {
offset := time.Duration(poll) * 10 * time.Second
m1Seen := base.Add(offset)
m2Seen := base.Add(offset + 2*time.Second)
flip := poll%2 == 1
if flip {
// Alternate which manager reported most recently, well inside
// the docker stale threshold, and flip snapshot host order too.
m1Seen, m2Seen = m2Seen, m1Seen
}
current := build(m1Seen, m2Seen, flip)
beforeByID := make(map[string]Resource, len(previous))
for _, resource := range previous {
beforeByID[resource.ID] = resource
}
afterByID := make(map[string]Resource, len(current))
for _, resource := range current {
afterByID[resource.ID] = resource
}
union := make(map[string]struct{}, len(beforeByID)+len(afterByID))
for id := range beforeByID {
union[id] = struct{}{}
}
for id := range afterByID {
union[id] = struct{}{}
}
for id := range union {
before, beforeOK := beforeByID[id]
after, afterOK := afterByID[id]
change := buildResourceChange(before, beforeOK, after, afterOK, base.Add(offset), nil, SourcePulseDiff, "")
if change != nil {
t.Fatalf("poll %d: unexpected %s change for %s: from=%q to=%q reason=%q",
poll, change.Kind, id, change.From, change.To, change.Reason)
}
}
previous = current
}
}
func TestPreferDockerSwarmHost(t *testing.T) {
base := time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC)
threshold := 120 * time.Second
host := func(id string, lastSeen time.Time, controlAvailable bool) models.DockerHost {
return models.DockerHost{
ID: id,
LastSeen: lastSeen,
Swarm: &models.DockerSwarmInfo{ClusterID: "swarm-cluster-1", ControlAvailable: controlAvailable},
}
}
manager := host("host-b", base, true)
fresherWorker := host("host-a", base.Add(60*time.Second), false)
if !preferDockerSwarmHost(manager, fresherWorker, threshold) {
t.Error("control-available manager should beat a fresher worker")
}
if preferDockerSwarmHost(fresherWorker, manager, threshold) {
t.Error("fresher worker should not beat a control-available manager")
}
// Freshness jitter inside the stale threshold must not decide: the
// lowest host ID wins from either direction.
a := host("host-a", base, true)
bFresher := host("host-b", base.Add(30*time.Second), true)
if !preferDockerSwarmHost(a, bFresher, threshold) {
t.Error("lowest host ID should win when freshness jitter is inside the stale threshold")
}
if preferDockerSwarmHost(bFresher, a, threshold) {
t.Error("higher host ID should lose when freshness jitter is inside the stale threshold")
}
// Once the gap exceeds the stale threshold the fresher host wins even
// with the higher ID: the other manager has genuinely gone quiet.
bWellFresher := host("host-b", base.Add(threshold+time.Second), true)
if !preferDockerSwarmHost(bWellFresher, a, threshold) {
t.Error("host beyond the stale threshold should lose to the fresher host")
}
if preferDockerSwarmHost(a, bWellFresher, threshold) {
t.Error("stale host should not beat a host fresher by more than the stale threshold")
}
}