vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-07-10 00:14:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

fix(monitoring): refresh discovered cluster endpoints when the cluster re-IPs

Browse source 
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
This commit is contained in:
rcourtman 2026-06-10 16:20:35 +01:00
parent 6eaea2fc3a
commit 82fdc6cfe2
2 changed files with 327 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

150
internal/monitoring/monitor_pve_cluster.go
View file

@ -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

181
internal/monitoring/monitor_pve_cluster_refresh_test.go Normal file
View file

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