From 82fdc6cfe2e4f841f6dd78d26e8cf68caafefe67 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 10 Jun 2026 16:20:35 +0100 Subject: [PATCH] fix(monitoring): refresh discovered cluster endpoints when the cluster re-IPs Cluster endpoints were discovered once at add time and never re-read: detectClusterMembership returned immediately for instances already marked IsCluster, so when a cluster moved subnets Pulse kept dialing the dead per-node addresses for failover and kept displaying them on Settings -> Infrastructure, even though monitoring still worked through the main host fallback. The manual /refresh-cluster endpoint exists but is not wired to any UI control, so nothing ever corrected the stored addresses. The 5-minute cluster re-check now also covers cluster instances: it re-reads /cluster/status through the configured main host, and when a node's address or the node set actually changed (volatile fields like Online/LastSeen are ignored) it updates the stored endpoints, persists the config, and rebuilds the failover client so polling moves to the new addresses. User-managed fields (IP overrides, guest URLs, fingerprints) are preserved by the existing discovery helpers, and fields the cluster status API omits are inherited from the stored endpoint rather than erased. Refs #1493 --- internal/monitoring/monitor_pve_cluster.go | 150 ++++++++++++++- .../monitor_pve_cluster_refresh_test.go | 181 ++++++++++++++++++ 2 files changed, 327 insertions(+), 4 deletions(-) create mode 100644 internal/monitoring/monitor_pve_cluster_refresh_test.go diff --git a/internal/monitoring/monitor_pve_cluster.go b/internal/monitoring/monitor_pve_cluster.go index 547f5be82..72e064d98 100644 --- a/internal/monitoring/monitor_pve_cluster.go +++ b/internal/monitoring/monitor_pve_cluster.go @@ -17,11 +17,12 @@ var detectMonitorPVECluster = defaultDetectMonitorPVECluster func (m *Monitor) detectClusterMembership(ctx context.Context, instanceName string, instanceCfg *config.PVEInstance, client PVEClientInterface) { _ = client - if instanceCfg.IsCluster { - return - } - // Check every 5 minutes if this is actually a cluster + // Re-check cluster status every 5 minutes. Standalone instances may turn + // out to be cluster members (#437). Cluster instances re-discover their + // endpoints, because the addresses captured at add time go stale when the + // cluster re-IPs - without a refresh Pulse keeps dialing (and displaying) + // dead addresses forever (#1493). if time.Since(m.lastClusterCheck[instanceName]) <= 5*time.Minute { return } @@ -34,6 +35,11 @@ func (m *Monitor) detectClusterMembership(ctx context.Context, instanceName stri return } + if instanceCfg.IsCluster { + m.refreshClusterEndpoints(instanceName, instanceCfg, clusterName, clusterEndpoints) + return + } + log.Info(). Str("instance", instanceName). Str("cluster", clusterName). @@ -66,6 +72,142 @@ func (m *Monitor) detectClusterMembership(ctx context.Context, instanceName stri } } +// refreshClusterEndpoints reconciles the stored cluster endpoints with what +// the cluster reports now, keeping user-managed fields (IP overrides, guest +// URLs, fingerprints - already carried over by the discovery helpers). When +// node addresses changed, the config is persisted and the failover client is +// rebuilt so polling stops dialing the dead addresses. +func (m *Monitor) refreshClusterEndpoints(instanceName string, instanceCfg *config.PVEInstance, clusterName string, discovered []config.ClusterEndpoint) { + refreshed := mergeRefreshedClusterEndpoints(instanceCfg.ClusterEndpoints, discovered) + if !clusterEndpointIdentityChanged(instanceCfg.ClusterEndpoints, refreshed) { + return + } + + log.Info(). + Str("instance", instanceName). + Str("cluster", clusterName). + Int("endpoints", len(refreshed)). + Msg("Cluster node addresses changed since discovery - refreshing stored endpoints") + + instanceCfg.ClusterEndpoints = refreshed + if !strings.EqualFold(clusterName, "unknown cluster") { + instanceCfg.ClusterName = clusterName + } + + updated := false + for i := range m.config.PVEInstances { + if m.config.PVEInstances[i].Name == instanceName { + m.config.PVEInstances[i].ClusterName = instanceCfg.ClusterName + m.config.PVEInstances[i].ClusterEndpoints = refreshed + updated = true + break + } + } + if !updated { + return + } + + m.normalizePVEConfigState() + if m.persistence != nil { + if err := m.persistence.SaveNodesConfig(m.config.PVEInstances, m.config.PBSInstances, m.config.PMGInstances); err != nil { + log.Warn().Err(err).Msg("failed to persist refreshed cluster endpoints") + } + } + + m.rebuildPVEClusterClient(instanceName) +} + +// mergeRefreshedClusterEndpoints folds freshly discovered endpoints over the +// stored set. The cluster status API can omit per-node fields, so discovery +// never erases information Pulse already had; Pulse-side reachability +// bookkeeping is carried over because it is recomputed each poll and dropping +// it would flap the UI to "unknown" after every refresh. +func mergeRefreshedClusterEndpoints(existing, discovered []config.ClusterEndpoint) []config.ClusterEndpoint { + merged := make([]config.ClusterEndpoint, 0, len(discovered)) + for _, ep := range discovered { + for _, old := range existing { + if !strings.EqualFold(strings.TrimSpace(old.NodeName), strings.TrimSpace(ep.NodeName)) { + continue + } + if strings.TrimSpace(ep.IP) == "" { + ep.IP = old.IP + } + if strings.TrimSpace(ep.Host) == "" { + ep.Host = old.Host + } + if strings.TrimSpace(ep.NodeID) == "" { + ep.NodeID = old.NodeID + } + ep.PulseReachable = old.PulseReachable + ep.LastPulseCheck = old.LastPulseCheck + ep.PulseError = old.PulseError + break + } + merged = append(merged, ep) + } + return merged +} + +// clusterEndpointIdentityChanged reports whether the set of nodes or any +// node's address changed. Volatile fields (Online, LastSeen, Pulse +// reachability) are deliberately ignored - they change every poll and are not +// a reason to rewrite config or rebuild clients. +func clusterEndpointIdentityChanged(existing, refreshed []config.ClusterEndpoint) bool { + if len(existing) != len(refreshed) { + return true + } + byName := make(map[string]config.ClusterEndpoint, len(existing)) + for _, ep := range existing { + byName[strings.ToLower(strings.TrimSpace(ep.NodeName))] = ep + } + for _, ep := range refreshed { + old, ok := byName[strings.ToLower(strings.TrimSpace(ep.NodeName))] + if !ok { + return true + } + if strings.TrimSpace(old.IP) != strings.TrimSpace(ep.IP) || + strings.TrimSpace(old.Host) != strings.TrimSpace(ep.Host) || + strings.TrimSpace(old.NodeID) != strings.TrimSpace(ep.NodeID) { + return true + } + } + return false +} + +// rebuildPVEClusterClient swaps the failover client for a cluster instance so +// it picks up the refreshed endpoint list. The in-flight poll keeps using the +// old client; the next poll cycle gets the new one. +func (m *Monitor) rebuildPVEClusterClient(instanceName string) { + if m.config == nil || m.pveClients == nil { + return + } + + var pve *config.PVEInstance + for i := range m.config.PVEInstances { + if m.config.PVEInstances[i].Name == instanceName { + pve = &m.config.PVEInstances[i] + break + } + } + if pve == nil || !pve.IsCluster || len(pve.ClusterEndpoints) == 0 { + return + } + + endpoints, endpointFingerprints := m.buildClusterEndpointsForReconnect(*pve) + clientConfig := config.CreateProxmoxConfig(pve) + clientConfig.Timeout = m.config.ConnectionTimeout + clusterClient := proxmox.NewClusterClient(pve.Name, clientConfig, endpoints, endpointFingerprints) + + m.mu.Lock() + m.pveClients[instanceName] = clusterClient + m.mu.Unlock() + + log.Info(). + Str("instance", instanceName). + Strs("endpoints", endpoints). + Msg("Rebuilt cluster client with refreshed endpoints") +} + func (m *Monitor) updateClusterEndpointStatus(instanceName string, instanceCfg *config.PVEInstance, client PVEClientInterface, modelNodes []models.Node) { if !instanceCfg.IsCluster || len(instanceCfg.ClusterEndpoints) == 0 { return diff --git a/internal/monitoring/monitor_pve_cluster_refresh_test.go b/internal/monitoring/monitor_pve_cluster_refresh_test.go new file mode 100644 index 000000000..bd4748381 --- /dev/null +++ b/internal/monitoring/monitor_pve_cluster_refresh_test.go @@ -0,0 +1,181 @@ +package monitoring + +import ( + "context" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +// Refreshing an existing cluster's endpoints when the cluster re-IPs (#1493): +// the stored add-time addresses must be replaced with what the cluster +// reports now, and the failover client must be rebuilt so polling stops +// dialing the dead addresses. +func TestDetectClusterMembership_RefreshesStaleClusterEndpoints(t *testing.T) { + originalDetect := detectMonitorPVECluster + t.Cleanup(func() { detectMonitorPVECluster = originalDetect }) + + detectMonitorPVECluster = func(clientConfig proxmox.ClientConfig, existingEndpoints []config.ClusterEndpoint) (bool, string, []config.ClusterEndpoint) { + return true, "MyCluster", []config.ClusterEndpoint{ + // Same nodes, new subnet. 127.0.0.1:1 fails instantly (connection + // refused) so the rebuilt client's health check doesn't block. + {NodeID: "node/proxmox0", NodeName: "proxmox0", Host: "https://proxmox0:8006", IP: "127.0.0.1", Online: true, LastSeen: time.Now()}, + {NodeID: "node/proxmox1", NodeName: "proxmox1", Host: "https://proxmox1:8006", IP: "127.0.0.2", Online: true, LastSeen: time.Now()}, + } + } + + sentinel := &stubPVEClient{} + m := &Monitor{ + config: &config.Config{ + PVEInstances: []config.PVEInstance{ + { + Name: "Proxmox2", + Host: "https://127.0.0.1:1", + IsCluster: true, + ClusterName: "MyCluster", + ClusterEndpoints: []config.ClusterEndpoint{ + {NodeID: "node/proxmox0", NodeName: "proxmox0", Host: "https://proxmox0:8006", IP: "127.0.0.9"}, + {NodeID: "node/proxmox1", NodeName: "proxmox1", Host: "https://proxmox1:8006", IP: "127.0.0.10"}, + }, + }, + }, + }, + state: models.NewState(), + pveClients: map[string]PVEClientInterface{"Proxmox2": sentinel}, + lastClusterCheck: make(map[string]time.Time), + } + instanceCfg := &m.config.PVEInstances[0] + + m.detectClusterMembership(context.Background(), "Proxmox2", instanceCfg, sentinel) + + got := m.config.PVEInstances[0].ClusterEndpoints + if len(got) != 2 { + t.Fatalf("expected 2 refreshed endpoints, got %d", len(got)) + } + if got[0].IP != "127.0.0.1" || got[1].IP != "127.0.0.2" { + t.Fatalf("expected refreshed IPs, got %q and %q", got[0].IP, got[1].IP) + } + if instanceCfg.ClusterEndpoints[0].IP != "127.0.0.1" { + t.Fatalf("expected in-flight instance config to see refreshed IPs, got %q", instanceCfg.ClusterEndpoints[0].IP) + } + + replacement, ok := m.pveClients["Proxmox2"] + if !ok { + t.Fatal("expected cluster client to remain registered") + } + if replacement == PVEClientInterface(sentinel) { + t.Fatal("expected cluster client to be rebuilt after endpoint refresh") + } + if _, isCluster := replacement.(*proxmox.ClusterClient); !isCluster { + t.Fatalf("expected rebuilt client to be a ClusterClient, got %T", replacement) + } +} + +// When discovery reports the same node identities (only volatile fields like +// Online/LastSeen differ), nothing should be rewritten or rebuilt. +func TestDetectClusterMembership_NoRefreshWhenEndpointsUnchanged(t *testing.T) { + originalDetect := detectMonitorPVECluster + t.Cleanup(func() { detectMonitorPVECluster = originalDetect }) + + detectMonitorPVECluster = func(clientConfig proxmox.ClientConfig, existingEndpoints []config.ClusterEndpoint) (bool, string, []config.ClusterEndpoint) { + return true, "MyCluster", []config.ClusterEndpoint{ + {NodeID: "node/proxmox0", NodeName: "proxmox0", Host: "https://proxmox0:8006", IP: "127.0.0.9", Online: true, LastSeen: time.Now()}, + } + } + + sentinel := &stubPVEClient{} + m := &Monitor{ + config: &config.Config{ + PVEInstances: []config.PVEInstance{ + { + Name: "Proxmox2", + Host: "https://127.0.0.1:1", + IsCluster: true, + ClusterName: "MyCluster", + ClusterEndpoints: []config.ClusterEndpoint{ + {NodeID: "node/proxmox0", NodeName: "proxmox0", Host: "https://proxmox0:8006", IP: "127.0.0.9", Online: false}, + }, + }, + }, + }, + state: models.NewState(), + pveClients: map[string]PVEClientInterface{"Proxmox2": sentinel}, + lastClusterCheck: make(map[string]time.Time), + } + + m.detectClusterMembership(context.Background(), "Proxmox2", &m.config.PVEInstances[0], sentinel) + + if m.pveClients["Proxmox2"] != PVEClientInterface(sentinel) { + t.Fatal("expected cluster client to be left alone when endpoints are unchanged") + } + if m.config.PVEInstances[0].ClusterEndpoints[0].Online { + t.Fatal("expected stored endpoints to be untouched when identity is unchanged") + } +} + +func TestMergeRefreshedClusterEndpoints_InheritsMissingFields(t *testing.T) { + reachable := true + checked := time.Now().Add(-time.Minute) + existing := []config.ClusterEndpoint{ + { + NodeID: "node/proxmox0", + NodeName: "proxmox0", + Host: "https://proxmox0:8006", + IP: "10.32.21.21", + PulseReachable: &reachable, + LastPulseCheck: &checked, + PulseError: "previous error", + }, + } + discovered := []config.ClusterEndpoint{ + // Cluster status omitted the IP for this node; the stored one must + // survive the refresh. + {NodeID: "", NodeName: "proxmox0", Host: "", IP: ""}, + {NodeID: "node/proxmox1", NodeName: "proxmox1", Host: "https://proxmox1:8006", IP: "10.32.20.22"}, + } + + merged := mergeRefreshedClusterEndpoints(existing, discovered) + if len(merged) != 2 { + t.Fatalf("expected 2 merged endpoints, got %d", len(merged)) + } + if merged[0].IP != "10.32.21.21" || merged[0].Host != "https://proxmox0:8006" || merged[0].NodeID != "node/proxmox0" { + t.Fatalf("expected omitted fields to be inherited, got %+v", merged[0]) + } + if merged[0].PulseReachable != &reachable || merged[0].LastPulseCheck != &checked || merged[0].PulseError != "previous error" { + t.Fatalf("expected Pulse reachability bookkeeping to be carried over, got %+v", merged[0]) + } + if merged[1].IP != "10.32.20.22" { + t.Fatalf("expected new node to keep its discovered IP, got %q", merged[1].IP) + } +} + +func TestClusterEndpointIdentityChanged(t *testing.T) { + base := []config.ClusterEndpoint{ + {NodeID: "node/a", NodeName: "a", Host: "https://a:8006", IP: "10.0.0.1"}, + {NodeID: "node/b", NodeName: "b", Host: "https://b:8006", IP: "10.0.0.2"}, + } + + volatileOnly := []config.ClusterEndpoint{ + {NodeID: "node/a", NodeName: "a", Host: "https://a:8006", IP: "10.0.0.1", Online: true, LastSeen: time.Now()}, + {NodeID: "node/b", NodeName: "b", Host: "https://b:8006", IP: "10.0.0.2", Online: false}, + } + if clusterEndpointIdentityChanged(base, volatileOnly) { + t.Fatal("volatile-only differences must not count as identity changes") + } + + reIPed := []config.ClusterEndpoint{ + {NodeID: "node/a", NodeName: "a", Host: "https://a:8006", IP: "10.32.20.1"}, + {NodeID: "node/b", NodeName: "b", Host: "https://b:8006", IP: "10.0.0.2"}, + } + if !clusterEndpointIdentityChanged(base, reIPed) { + t.Fatal("an IP change must count as an identity change") + } + + nodeAdded := append(append([]config.ClusterEndpoint{}, base...), config.ClusterEndpoint{NodeID: "node/c", NodeName: "c", Host: "https://c:8006", IP: "10.0.0.3"}) + if !clusterEndpointIdentityChanged(base, nodeAdded) { + t.Fatal("an added node must count as an identity change") + } +}