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") + } +}