From 8c9000ac63930f3fcca74128e213346851e3fc34 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:09:11 +0100 Subject: [PATCH] Keep node temperatures off same-named nodes at other sites Two Proxmox connections can each have a node with the same name, which is common for MSPs running one px1 per site. The node temperature lookup fell back to the first host agent whose hostname matched the node name, even when that agent was linked to the other site's node, and even when the name was ambiguous enough that the agent linker had refused to link it. Site B's px1 then showed, alerted on and recorded history for site A's sensors. The cluster sensor cache leaked the same way: an agent's SSH readings for its cluster siblings were keyed by bare node name, so a site A sibling reading fed site B's node. The poller now hands the lookup the polled node, and the lookup works on that node's slot. The state folds the views of one machine reached through two connections (a cluster added twice, or a multi-homed host added by each address) into one node held under one view's ID, and the agent link follows it. Those views used to find their agent only through the loose hostname match, so a polled view the read state does not hold also takes the one same-named node proven to be the same machine. models.NodeObservationsSameMachine and models.HostAgentBridgesNodeViews give that proof from the evidence UpdateNodesForInstance folds on, and the state's agent corroboration check now shares its rule with them. Ambiguous candidates are declined, as the state declines an ambiguous alias. An agent linked to the slot is the node's agent. The hostname fallback only takes an agent with no link when no node outside the slot and no second unlinked agent has the name. Cluster sensor cache entries record their reporting agent and serve a node only when that agent's linked node shares one of the slot's connections, or belongs to a connection whose configured view of the node has a matching TLS fingerprint. LinkedNodeID holds the Proxmox source node ID for automatic, manual and continuity-restored links, so the link match needs no normalization. The physical-disk SMART merge already matches agents by node.LinkedAgentID within one instance and is unchanged. --- .../v6/internal/subsystems/agent-lifecycle.md | 13 + .../v6/internal/subsystems/monitoring.md | 40 +++ internal/models/models.go | 16 +- internal/models/node_machine_identity.go | 82 +++++ internal/models/node_machine_identity_test.go | 124 ++++++++ internal/monitoring/host_agent_temps.go | 291 ++++++++++++++---- internal/monitoring/host_agent_temps_test.go | 234 +++++++------- internal/monitoring/monitor.go | 6 +- .../monitoring/monitor_additional_test.go | 210 +++++++++++++ internal/monitoring/monitor_agents.go | 15 +- .../monitoring/monitor_host_agents_test.go | 33 ++ .../monitor_polling_node_helpers.go | 8 +- internal/monitoring/monitor_pve.go | 2 +- 13 files changed, 881 insertions(+), 193 deletions(-) create mode 100644 internal/models/node_machine_identity.go create mode 100644 internal/models/node_machine_identity_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 6813b04986..3d384e6479 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -288,6 +288,19 @@ not disable the local host, SMART, Ceph, or Proxmox reporting modules, alter agent enrollment or identity, or add remote command authority. The disabled path is pinned by `TestCollectClusterSensors_Disabled`; configuration parsing is pinned by `TestLoadConfigDisableClusterPeerSensorsFlag`. +Pulse attributes peer readings to the agent that reported them. Peers are +named by bare Proxmox node name, which a node of another connection can share, +so monitoring keys its cluster sensor cache by reporting agent and node name. +A reading serves a node only when the agent's currently linked node belongs to +that node's connection, or to another connection with a fingerprint-proven view +of the node (a cluster added twice); attribution follows the agent's link like +its own sensors. An unlinked reporter's peer readings serve no node, nor do +those of a reporter whose manual link names a node ID the read state does not +hold. This reads +the existing agent link and adds no identity, enrollment, link or command +authority; the ingest path is pinned by +`TestApplyHostReportScopesClusterSensorsToReportersConnection` in +`internal/monitoring/monitor_host_agents_test.go`. On supported Linux systemd hosts, the opt-in safe runtime is a root-owned, unprivileged monitoring collector plus the no-network typed helper, with remediation installed only as the separate root-owned `pulse-agent-runner`. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 576a27650b..0e2bda7c3d 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -3789,6 +3789,46 @@ exists or the agent payload has no usable positive reading. Identity-only or zero-temperature SMART rows do not count as usable by themselves, but the runtime must not keep probing legacy SSH solely to augment an otherwise healthy agent temperature payload with SMART data. +A host agent's sensors feed the node the agent is linked to. Several Proxmox +connections can each have a node of the same name (one `px1` per site), and +`Host.LinkedNodeID` always holds the Proxmox source node ID, whether the link is +automatic, manual or restored from host continuity. The poller therefore hands +`getHostAgentTemperatureForNode` in `internal/monitoring/host_agent_temps.go` the +polled node itself, and the lookup works on that node's slot. A node the read +state holds under its own source ID is its own slot. The state folds the views +of one machine reached through two connections (a cluster added twice, or a +multi-homed host added by each address) into one node kept under one view's ID, +chosen by its merge preference, and an automatic agent link follows that node. +So a polled view the read state does not hold also takes the one same-named +node proven to be the same machine, by `models.NodeObservationsSameMachine` or, +for the agent linked to that node, `models.HostAgentBridgesNodeViews` (the agent +reports both endpoints and the views' TLS fingerprints do not contradict), both +in `internal/models/node_machine_identity.go`. These are pairwise forms of +evidence `UpdateNodesForInstance` folds on, and it shares the agent +corroboration rule with them; +`TestNodeObservationsSameMachineMatchesStateFold` and +`TestHostAgentBridgesNodeViewsMatchesStateFold` pin them against the state's +fold for the covered cases. Two or more such candidates are ambiguous and none +is taken, as the state declines an ambiguous alias. An agent linked to a slot ID +is the node's agent. A manual link stays pinned to the source ID the operator +chose, because the state never transfers operator intent to a replacement +provider identity, so while the state keeps that machine under the other view's +ID the node has no linked agent and no agent reading. Otherwise the hostname +fallback takes only an agent with no node, VM or container link whose hostname +is exactly the node name, and only when no Proxmox node outside the slot and no +second unlinked agent has that name. Cluster-sibling readings an agent collects over SSH arrive keyed by bare +node name, so the cluster sensor cache records the reporting agent. A reading +serves the node only when the node that agent is currently linked to belongs to +one of the slot's connections, or to a connection whose configured view of the +node carries a matching TLS fingerprint (the other connection of a cluster added +twice). The reading follows the agent's link the way the agent's own sensors do, +and a reading from an unlinked agent serves no node. Without these bounds, site +B's `px1` showed, alerted on and recorded history for site A's agent or +cluster-sibling sensors. `TestNodeTemperatureStaysWithinSameNamedNodesSite` in +`internal/monitoring/monitor_additional_test.go` pins the poller lookup, +including the folded-view cases, and +`TestApplyHostReportScopesClusterSensorsToReportersConnection` in +`internal/monitoring/monitor_host_agents_test.go` pins the agent ingest side. A node temperature presented as available must describe the node now, because node alert evaluation, node history writes, reporting, and the UI all read it as a live measurement. A host agent linked to a Proxmox node merges into one host diff --git a/internal/models/models.go b/internal/models/models.go index 30aaa665e6..0656616c9a 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -4051,7 +4051,7 @@ func (s *State) UpdateNodesForInstance(instanceName string, nodes []Node) { hostAgentByHostname := make(map[string]map[string]struct{}) // lowercase hostname -> hostAgentIDs hostAgentByIP := make(map[string]map[string]struct{}) // normalized ip -> hostAgentIDs validHostAgentIDs := make(map[string]bool) // set of existing host agent IDs - hostHostnameByID := make(map[string]string) // hostAgentID -> normalized full hostname + hostByID := make(map[string]Host) // hostAgentID -> host addHostAlias := func(name, hostID string) { name = strings.TrimSpace(strings.ToLower(name)) if name == "" || hostID == "" { @@ -4086,7 +4086,7 @@ func (s *State) UpdateNodesForInstance(instanceName string, nodes []Node) { for _, host := range s.Hosts { if host.ID != "" { validHostAgentIDs[host.ID] = true - hostHostnameByID[host.ID] = strings.TrimSpace(strings.ToLower(host.Hostname)) + hostByID[host.ID] = host addHostAlias(host.Hostname, host.ID) // Also index by short hostname if idx := strings.Index(host.Hostname, "."); idx > 0 { @@ -4109,16 +4109,8 @@ func (s *State) UpdateNodesForInstance(instanceName string, nodes []Node) { // Preserve legitimate split views when the same agent strongly bridges // both endpoints through an exact IP or a full (dotted) hostname. hostStronglyCorroboratesNode := func(hostID string, node Node) bool { - endpoint := extractHostEndpoint(node.Host) - if endpoint == "" { - return false - } - if ip := normalizeIPAddress(endpoint); ip != "" { - _, ok := hostIPsByID[hostID][ip] - return ok - } - hostname := hostHostnameByID[hostID] - return strings.Contains(endpoint, ".") && endpoint == hostname + host, ok := hostByID[hostID] + return ok && hostReportsNodeEndpoint(host, node) } sharedAgentProvesNodePair := func(hostID string, existing, candidate Node) bool { if nodeCrossViewMergeProven(existing, candidate) { diff --git a/internal/models/node_machine_identity.go b/internal/models/node_machine_identity.go new file mode 100644 index 0000000000..98ad82a30a --- /dev/null +++ b/internal/models/node_machine_identity.go @@ -0,0 +1,82 @@ +package models + +import "strings" + +// NodeObservationsSameMachine reports whether two observations of a Proxmox +// node are one machine by the evidence UpdateNodesForInstance folds node views +// into one slot on: the same logical key, or the same node name with no +// provider identity conflict and either positive cross-view proof (the same +// connection, the same cluster with matching TLS fingerprints, matching +// fingerprints, or for unclustered views the same endpoint) or a shared +// endpoint address. A cluster added through two connections is such a pair. +// Callers that see a polled node before it is merged, such as the node's host +// agent lookup, use this so they agree with the state on which views are the +// same node. +func NodeObservationsSameMachine(a, b Node) bool { + if key := nodeLogicalKey(a); key != "" && key == nodeLogicalKey(b) { + return true + } + name := normalizeNodeIdentityPart(a.Name) + if name == "" || name != normalizeNodeIdentityPart(b.Name) || nodeProviderIdentityConflicts(a, b) { + return false + } + if nodeCrossViewMergeProven(a, b) { + return true + } + // An address alias folds on the conflict check alone (resolveNodeMergeKey). + otherAliases := nodeEndpointMergeAliases(b) + for _, alias := range nodeEndpointMergeAliases(a) { + if !strings.HasPrefix(alias, "endpoint-ip:") { + continue + } + for _, other := range otherAliases { + if alias == other { + return true + } + } + } + return false +} + +// HostAgentBridgesNodeViews reports whether a host agent linked to view a proves +// view b is the same machine, the way UpdateNodesForInstance lets that agent +// fold b into a's slot: neither view names a cluster, their TLS fingerprints do +// not contradict each other (the state rejects an agent for a node whose known +// fingerprint differs from its linked node's), and the agent's own report names +// both endpoints, by exact address or full hostname. A host with several +// addresses added through two standalone connections is such a pair. +func HostAgentBridgesNodeViews(host Host, a, b Node) bool { + if normalizeNodeIdentityPart(a.ClusterName) != "" || normalizeNodeIdentityPart(b.ClusterName) != "" { + return false + } + fpA := normalizeNodeTLSFingerprint(a.TLSFingerprint) + fpB := normalizeNodeTLSFingerprint(b.TLSFingerprint) + if fpA != "" && fpB != "" && fpA != fpB { + return false + } + return hostReportsNodeEndpoint(host, a) && hostReportsNodeEndpoint(host, b) +} + +// hostReportsNodeEndpoint reports whether a host agent's own report names the +// node's endpoint: the endpoint address among the addresses it reports, or the +// endpoint's full (dotted) hostname as its hostname. +func hostReportsNodeEndpoint(host Host, node Node) bool { + endpoint := extractHostEndpoint(node.Host) + if endpoint == "" { + return false + } + if ip := normalizeIPAddress(endpoint); ip != "" { + if normalizeIPAddress(host.ReportIP) == ip { + return true + } + for _, iface := range host.NetworkInterfaces { + for _, address := range iface.Addresses { + if normalizeIPAddress(address) == ip { + return true + } + } + } + return false + } + return strings.Contains(endpoint, ".") && endpoint == strings.TrimSpace(strings.ToLower(host.Hostname)) +} diff --git a/internal/models/node_machine_identity_test.go b/internal/models/node_machine_identity_test.go new file mode 100644 index 0000000000..65c06cba98 --- /dev/null +++ b/internal/models/node_machine_identity_test.go @@ -0,0 +1,124 @@ +package models + +import "testing" + +// NodeObservationsSameMachine must agree with the fold UpdateNodesForInstance +// applies to two views of a node from different connections, so callers that +// see a polled node before it is merged treat the same views as one node. +func TestNodeObservationsSameMachineMatchesStateFold(t *testing.T) { + clusterNode := func(instance, cluster, host, fingerprint string) Node { + return Node{ + ID: instance + "-pve01", NodeIdentity: instance + "-pve01", Name: "pve01", Instance: instance, + ClusterName: cluster, IsClusterMember: cluster != "", Host: host, TLSFingerprint: fingerprint, Status: "online", + } + } + for _, tc := range []struct { + name string + existing, candidate Node + wantSameMachineFolds bool + }{ + { + name: "same cluster added twice with matching fingerprints", + existing: clusterNode("enacon-a", "enacon", "https://192.168.1.11:8006", "AA:AA:AA:AA"), + candidate: clusterNode("enacon-b", "enacon", "https://192.168.1.11:8006", "aaaaaaaa"), + wantSameMachineFolds: true, + }, + { + name: "same-named clusters without fingerprints", + existing: clusterNode("site-a", "cluster", "https://192.168.1.11:8006", ""), + candidate: clusterNode("site-b", "cluster", "https://192.168.1.11:8006", ""), + }, + { + name: "same-named clusters with different fingerprints", + existing: clusterNode("site-a", "cluster", "https://192.168.1.11:8006", "AA:AA"), + candidate: clusterNode("site-b", "cluster", "https://192.168.1.11:8006", "BB:BB"), + }, + { + name: "different clusters reusing a node name and address", + existing: clusterNode("site-a", "a", "https://192.168.1.11:8006", "AA:AA"), + candidate: clusterNode("site-b", "b", "https://192.168.1.11:8006", "AA:AA"), + }, + { + name: "standalone nodes of one name at different addresses", + existing: clusterNode("site-a", "", "https://10.0.0.5:8006", ""), + candidate: clusterNode("site-b", "", "https://10.0.1.5:8006", ""), + }, + { + name: "standalone connections to one address", + existing: clusterNode("by-ip", "", "https://10.0.0.5:8006", ""), + candidate: clusterNode("by-ip-again", "", "https://10.0.0.5:8006", ""), + wantSameMachineFolds: true, + }, + { + name: "standalone view of a cluster member's address", + existing: clusterNode("cluster-view", "enacon", "https://192.168.1.11:8006", ""), + candidate: clusterNode("standalone-view", "", "https://192.168.1.11:8006", ""), + wantSameMachineFolds: true, + }, + { + name: "same cluster by hostname with matching fingerprints", + existing: clusterNode("enacon-a", "enacon", "https://pve01.lan:8006", "AA:AA"), + candidate: clusterNode("enacon-b", "enacon", "https://pve01.lan:8006", "AA:AA"), + wantSameMachineFolds: true, + }, + { + name: "different clusters by hostname", + existing: clusterNode("site-a", "a", "https://pve01.lan:8006", ""), + candidate: clusterNode("site-b", "b", "https://pve01.lan:8006", ""), + }, + } { + t.Run(tc.name, func(t *testing.T) { + state := &State{Nodes: []Node{tc.existing}} + state.UpdateNodesForInstance(tc.candidate.Instance, []Node{tc.candidate}) + if folded := len(state.Nodes) == 1; folded != tc.wantSameMachineFolds { + t.Fatalf("state fold = %v (nodes %#v), want %v", folded, state.Nodes, tc.wantSameMachineFolds) + } + if got := NodeObservationsSameMachine(tc.existing, tc.candidate); got != tc.wantSameMachineFolds { + t.Fatalf("NodeObservationsSameMachine = %v, want %v (the state's fold)", got, tc.wantSameMachineFolds) + } + if got := NodeObservationsSameMachine(tc.candidate, tc.existing); got != tc.wantSameMachineFolds { + t.Fatalf("NodeObservationsSameMachine reversed = %v, want %v", got, tc.wantSameMachineFolds) + } + }) + } +} + +// HostAgentBridgesNodeViews must agree with the fold UpdateNodesForInstance lets +// a linked agent make between two standalone views of one machine. +func TestHostAgentBridgesNodeViewsMatchesStateFold(t *testing.T) { + standalone := func(instance, host string) Node { + return Node{ID: instance + "-minipc", NodeIdentity: instance + "-minipc", Name: "minipc", Instance: instance, Host: host, Status: "online"} + } + for _, tc := range []struct { + name string + addresses []string + existingFP, candidateFP string + wantBridge bool + }{ + {name: "agent reports both connection addresses", addresses: []string{"10.0.0.5/24", "10.0.1.5/24"}, wantBridge: true}, + {name: "agent reports only the linked view's address", addresses: []string{"10.0.0.5/24"}}, + {name: "matching fingerprints", addresses: []string{"10.0.0.5/24", "10.0.1.5/24"}, existingFP: "AA:AA", candidateFP: "aaaa", wantBridge: true}, + {name: "contradicting fingerprints", addresses: []string{"10.0.0.5/24", "10.0.1.5/24"}, existingFP: "AA:AA", candidateFP: "BB:BB"}, + } { + t.Run(tc.name, func(t *testing.T) { + existing := standalone("lan-a", "https://10.0.0.5:8006") + existing.LinkedAgentID = "agent-minipc" + existing.TLSFingerprint = tc.existingFP + host := Host{ + ID: "agent-minipc", Hostname: "minipc", LinkedNodeID: existing.ID, ReportIP: "10.0.0.5", + NetworkInterfaces: []HostNetworkInterface{{Name: "eth0", Addresses: tc.addresses}}, + } + candidate := standalone("lan-b", "https://10.0.1.5:8006") + candidate.TLSFingerprint = tc.candidateFP + + state := &State{Nodes: []Node{existing}, Hosts: []Host{host}} + state.UpdateNodesForInstance(candidate.Instance, []Node{candidate}) + if folded := len(state.Nodes) == 1; folded != tc.wantBridge { + t.Fatalf("state fold = %v (nodes %#v), want %v", folded, state.Nodes, tc.wantBridge) + } + if got := HostAgentBridgesNodeViews(host, existing, candidate); got != tc.wantBridge { + t.Fatalf("HostAgentBridgesNodeViews = %v, want %v (the state's fold)", got, tc.wantBridge) + } + }) + } +} diff --git a/internal/monitoring/host_agent_temps.go b/internal/monitoring/host_agent_temps.go index 8b3fc4d3e0..83ae970aac 100644 --- a/internal/monitoring/host_agent_temps.go +++ b/internal/monitoring/host_agent_temps.go @@ -3,6 +3,7 @@ package monitoring import ( "math" "regexp" + "slices" "sort" "strconv" "strings" @@ -19,7 +20,7 @@ import ( // duplicate hostname scenarios), then falls back to hostname matching. // Returns nil if no matching host agent is found or if no temperature data is available. func (m *Monitor) getHostAgentTemperature(nodeName string) *models.Temperature { - return m.getHostAgentTemperatureByID("", nodeName) + return m.getHostAgentTemperatureForNode(models.Node{Name: nodeName}) } func shouldSkipTemperatureSSHCollection(hostAgentTemp *models.Temperature) bool { @@ -77,19 +78,23 @@ func hasUsableSMARTTemperature(temp *models.Temperature) bool { return false } -// getHostAgentTemperatureByID looks for a matching host agent by node ID first, -// then falls back to hostname matching. This correctly handles clusters where -// multiple nodes may have the same hostname (e.g., "px1" on different IPs). -func (m *Monitor) getHostAgentTemperatureByID(nodeID, nodeName string) *models.Temperature { +// getHostAgentTemperatureForNode returns a polled Proxmox node's reading from +// its host agent (see hostAgentForNode), else from a sibling agent's cluster +// sensor cache. node is the poller's model of the node: its source ID, native +// name, connection and identity stamps. +func (m *Monitor) getHostAgentTemperatureForNode(node models.Node) *models.Temperature { readState := m.GetUnifiedReadStateOrSnapshot() if readState == nil { return nil } + hosts := readState.Hosts() + nodes := readState.Nodes() + slot := m.polledNodeSlot(hosts, nodes, node) - matchedHost := hostAgentForNode(readState.Hosts(), nodeID, nodeName) + matchedHost := hostAgentForNode(hosts, nodes, slot, node.Name) if matchedHost == nil { // No directly-linked host agent found — check cluster sensor cache - return m.getClusterSensorTemperature(nodeName) + return m.getClusterSensorTemperature(hosts, nodes, node, slot) } // Check if the host agent has temperature data. SMART-only reports are @@ -98,7 +103,7 @@ func (m *Monitor) getHostAgentTemperatureByID(nodeID, nodeName string) *models.T sensors := matchedHost.Sensors() if sensors == nil || (len(sensors.TemperatureCelsius) == 0 && len(sensors.SMART) == 0) { // Host agent exists but has no temperature data — try cluster cache - return m.getClusterSensorTemperature(nodeName) + return m.getClusterSensorTemperature(hosts, nodes, node, slot) } // An agent that stopped reporting keeps its last sensors in state. Those @@ -108,7 +113,7 @@ func (m *Monitor) getHostAgentTemperatureByID(nodeID, nodeName string) *models.T // every PVE poll, so read the agent source's sighting. agentStatus, ok := matchedHost.SourceStatus(unifiedresources.SourceAgent) if !ok || !hostAgentReportCurrent(agentStatus.LastSeen, matchedHost.IntervalSeconds(), time.Now()) { - return m.getClusterSensorTemperature(nodeName) + return m.getClusterSensorTemperature(hosts, nodes, node, slot) } // Convert host agent sensor data to Temperature model, stamped with the @@ -125,7 +130,7 @@ func (m *Monitor) getHostAgentTemperatureByID(nodeID, nodeName string) *models.T // reading of that age is dropped too; it is already older than the lease, and // the bound errs toward a gap rather than a stale value. A reading stamped // after that report keeps the ordinary carry window. -func (m *Monitor) carriedTemperatureOutlivesAgentLease(nodeID, nodeName string, temp *models.Temperature, now time.Time) bool { +func (m *Monitor) carriedTemperatureOutlivesAgentLease(node models.Node, temp *models.Temperature, now time.Time) bool { if temp == nil { return false } @@ -133,7 +138,9 @@ func (m *Monitor) carriedTemperatureOutlivesAgentLease(nodeID, nodeName string, if readState == nil { return false } - matchedHost := hostAgentForNode(readState.Hosts(), nodeID, nodeName) + hosts := readState.Hosts() + nodes := readState.Nodes() + matchedHost := hostAgentForNode(hosts, nodes, m.polledNodeSlot(hosts, nodes, node), node.Name) if matchedHost == nil { return false } @@ -145,37 +152,186 @@ func (m *Monitor) carriedTemperatureOutlivesAgentLease(nodeID, nodeName string, return !temp.LastUpdate.After(agentStatus.LastSeen) } -// hostAgentForNode returns the host agent for a Proxmox node: the agent linked -// to the node ID, else one whose hostname matches. This correctly handles -// clusters where multiple nodes may have the same hostname (e.g., "px1" on -// different IPs). -func hostAgentForNode(hosts []*unifiedresources.HostView, nodeID, nodeName string) *unifiedresources.HostView { - // First, try to find a host agent that is explicitly linked to this node - // via LinkedNodeID. This is the most reliable method and handles duplicate - // hostnames correctly. - if nodeID != "" { - for i := range hosts { - if hosts[i].LinkedNodeID() == nodeID { - log.Debug(). - Str("nodeID", nodeID). - Str("hostAgentID", hosts[i].ID()). - Str("hostname", hosts[i].Hostname()). - Msg("Matched host agent to node via LinkedNodeID") - return hosts[i] - } +// nodeSlot is the read-state identity of one polled Proxmox node: the source +// node IDs and connections that denote it. +type nodeSlot struct { + ids []string + instances []string +} + +func (s *nodeSlot) add(id, instance string) { + if id = strings.TrimSpace(id); id != "" && !s.hasID(id) { + s.ids = append(s.ids, id) + } + if instance = strings.TrimSpace(instance); instance != "" && !s.hasInstance(instance) { + s.instances = append(s.instances, instance) + } +} + +func (s nodeSlot) hasID(id string) bool { + return id != "" && slices.Contains(s.ids, id) +} + +func (s nodeSlot) hasInstance(instance string) bool { + return instance != "" && slices.Contains(s.instances, instance) +} + +// polledNodeSlot returns the read-state identity of a polled Proxmox node. A +// node the read state holds under its own source ID is that node alone. The +// state folds the views of one machine reached through two connections (a +// cluster added twice, or a multi-homed host added by each address) into one +// slot, kept under whichever view its merge preference keeps, so a polled view +// the read state does not hold also counts as the one same-named node proven to +// be that machine: by models.NodeObservationsSameMachine, or by +// models.HostAgentBridgesNodeViews for the agent linked to that node. Two or +// more such candidates are ambiguous and none is taken, as the state's alias +// resolution declines them. Agent links and cluster-sensor scope are matched +// against the slot, never against a same-named node of another machine. +func (m *Monitor) polledNodeSlot(hosts []*unifiedresources.HostView, nodes []*unifiedresources.NodeView, node models.Node) nodeSlot { + var slot nodeSlot + if strings.TrimSpace(node.ID) == "" { + return slot + } + for _, view := range nodes { + if view != nil && view.SourceID() == node.ID { + slot.add(node.ID, firstNonEmptyString(node.Instance, view.Instance())) + return slot + } + } + slot.add(node.ID, node.Instance) + + var partner *unifiedresources.NodeView + for _, view := range nodes { + if view == nil || !strings.EqualFold(view.NodeName(), node.Name) { + continue + } + if !m.nodeViewIsSameMachine(hosts, view, node) { + continue + } + if partner != nil { + return slot + } + partner = view + } + if partner != nil { + slot.add(partner.SourceID(), partner.Instance()) + } + return slot +} + +// nodeViewIsSameMachine reports whether a read-state Proxmox node and a polled +// node are one machine by the evidence the state folds node views on. +func (m *Monitor) nodeViewIsSameMachine(hosts []*unifiedresources.HostView, view *unifiedresources.NodeView, node models.Node) bool { + stateNode := m.stateNodeFromView(view) + if models.NodeObservationsSameMachine(stateNode, node) { + return true + } + for _, host := range hosts { + if host.LinkedNodeID() == stateNode.ID { + return models.HostAgentBridgesNodeViews(hostFromReadStateView(host), stateNode, node) + } + } + return false +} + +// nodeHasProvenViewInConnection reports whether a polled node has a view under +// another Proxmox connection that is the same machine: a cluster added through +// two connections lists the same members, each with its TLS fingerprint, so +// that connection's view of the node's name is the polled node. +func (m *Monitor) nodeHasProvenViewInConnection(node models.Node, instance string) bool { + instanceCfg := m.getInstanceConfig(instance) + if instanceCfg == nil || !instanceCfg.IsCluster { + return false + } + fingerprint := pveNodeTLSFingerprint(instanceCfg, node.Name) + if fingerprint == "" { + return false + } + return models.NodeObservationsSameMachine(models.Node{ + Name: node.Name, + Instance: instanceCfg.Name, + ClusterName: instanceCfg.ClusterName, + IsClusterMember: true, + TLSFingerprint: fingerprint, + }, node) +} + +// stateNodeFromView restores the identity a read-state Proxmox node carries in +// state, including the config-derived TLS fingerprint the registry does not +// keep, the way previousNodesForInstance does. +func (m *Monitor) stateNodeFromView(view *unifiedresources.NodeView) models.Node { + node := previousNodeFromView(view) + node.TLSFingerprint = pveNodeTLSFingerprint(m.getInstanceConfig(node.Instance), node.Name) + return node +} + +// hostAgentForNode returns the host agent for a polled Proxmox node: the agent +// linked to the node's slot, else the one unlinked agent whose hostname is the +// node's name. Several connections can each have a node with the same name +// (one "px1" per site), so the hostname fallback never takes an agent linked to +// another node or to a guest, and finds nothing when a Proxmox node outside the +// slot or a second unlinked agent has the name. Every link source (automatic, +// manual, restored from continuity) stores the Proxmox source node ID, so slot +// IDs compare with LinkedNodeID directly. +func hostAgentForNode(hosts []*unifiedresources.HostView, nodes []*unifiedresources.NodeView, slot nodeSlot, nodeName string) *unifiedresources.HostView { + for _, host := range hosts { + if slot.hasID(host.LinkedNodeID()) { + log.Debug(). + Str("nodeID", host.LinkedNodeID()). + Str("hostAgentID", host.ID()). + Str("hostname", host.Hostname()). + Msg("Matched host agent to node via LinkedNodeID") + return host } } - // Fallback: match by hostname if no linked host was found - // This maintains backwards compatibility for setups where linking hasn't occurred yet - nodeLower := strings.ToLower(strings.TrimSpace(nodeName)) - for i := range hosts { - hostnameLower := strings.ToLower(strings.TrimSpace(hosts[i].Hostname())) - if hostnameLower == nodeLower { - return hosts[i] + // Fallback for an agent that is not linked yet. + name := strings.ToLower(strings.TrimSpace(nodeName)) + if name == "" || otherProxmoxNodeHasName(nodes, slot, name) { + return nil + } + var match *unifiedresources.HostView + for _, host := range hosts { + if host.LinkedNodeID() != "" || host.LinkedVMID() != "" || host.LinkedContainerID() != "" { + continue + } + if strings.ToLower(strings.TrimSpace(host.Hostname())) != name { + continue + } + if match != nil { + return nil + } + match = host + } + return match +} + +// otherProxmoxNodeHasName reports whether a Proxmox node outside slot is named +// name (lowercase). With an empty slot every node of that name counts. +func otherProxmoxNodeHasName(nodes []*unifiedresources.NodeView, slot nodeSlot, name string) bool { + for _, node := range nodes { + if node == nil || slot.hasID(node.SourceID()) { + continue + } + if strings.ToLower(node.NodeName()) == name { + return true } } - return nil + return false +} + +// proxmoxNodeInstance returns the Proxmox connection a node belongs to, or "" +// when the node is not in the read state. +func proxmoxNodeInstance(nodes []*unifiedresources.NodeView, nodeID string) string { + if nodeID == "" { + return "" + } + for _, node := range nodes { + if node != nil && node.SourceID() == nodeID { + return node.Instance() + } + } + return "" } func convertUnifiedHostSensorsToTemperature(sensors *unifiedresources.HostSensorMeta, lastSeen time.Time) *models.Temperature { @@ -243,30 +399,57 @@ func cloneSMARTAttributesModel(src *models.SMARTAttributes) *models.SMARTAttribu return &dest } -// getClusterSensorTemperature looks up cached temperature data that was collected -// by a sibling agent in the same Proxmox cluster via SSH. Returns nil if no -// recent data is available. -func (m *Monitor) getClusterSensorTemperature(nodeName string) *models.Temperature { - if nodeName == "" { - return nil - } - - key := strings.ToLower(strings.TrimSpace(nodeName)) - - m.clusterSensorsMu.RLock() - entry, ok := m.clusterSensorsCache[key] - m.clusterSensorsMu.RUnlock() - - if !ok { +// getClusterSensorTemperature returns the reading a host agent on a sibling +// cluster node collected for this node over SSH. Siblings are reported by bare +// node name, which nodes of different connections can share, so a reading only +// serves the node when the node its reporting agent is linked to belongs to one +// of the slot's connections, or to a connection with a proven view of the node +// (a cluster added twice). Readings from an unlinked agent serve no node. +func (m *Monitor) getClusterSensorTemperature(hosts []*unifiedresources.HostView, nodes []*unifiedresources.NodeView, node models.Node, slot nodeSlot) *models.Temperature { + name := strings.ToLower(strings.TrimSpace(node.Name)) + if name == "" { return nil } // Reuse the same staleness threshold as direct host agents (2 minutes) - if !isHostAgentTemperatureRecent(entry.updatedAt) { + var candidates []clusterSensorsCacheEntry + m.clusterSensorsMu.RLock() + for _, entry := range m.clusterSensorsCache { + if entry.nodeName == name && isHostAgentTemperatureRecent(entry.updatedAt) { + candidates = append(candidates, entry) + } + } + m.clusterSensorsMu.RUnlock() + + var newest *clusterSensorsCacheEntry + for i := range candidates { + reporterInstance := proxmoxNodeInstance(nodes, linkedNodeIDForAgent(hosts, candidates[i].reporterID)) + if reporterInstance == "" || + (!slot.hasInstance(reporterInstance) && !m.nodeHasProvenViewInConnection(node, reporterInstance)) { + continue + } + if newest == nil || candidates[i].updatedAt.After(newest.updatedAt) { + newest = &candidates[i] + } + } + if newest == nil { return nil } + return convertHostSensorsToTemperature(newest.sensors, newest.updatedAt) +} - return convertHostSensorsToTemperature(entry.sensors, entry.updatedAt) +// linkedNodeIDForAgent returns the Proxmox source node ID the agent with the +// given agent ID (models.Host.ID) is linked to, or "". +func linkedNodeIDForAgent(hosts []*unifiedresources.HostView, agentID string) string { + if agentID == "" { + return "" + } + for _, host := range hosts { + if host.AgentID() == agentID { + return host.LinkedNodeID() + } + } + return "" } // convertHostSensorsToTemperature converts HostSensorSummary to the Temperature model. diff --git a/internal/monitoring/host_agent_temps_test.go b/internal/monitoring/host_agent_temps_test.go index 5885b47ea8..7d330a21e3 100644 --- a/internal/monitoring/host_agent_temps_test.go +++ b/internal/monitoring/host_agent_temps_test.go @@ -8,6 +8,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" unifiedresources "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" "github.com/stretchr/testify/assert" ) @@ -466,7 +467,7 @@ func TestGetHostAgentTemperature(t *testing.T) { } m.state.UpsertHost(host) - result := m.getHostAgentTemperatureByID("node-123", "different-name") + result := m.getHostAgentTemperatureForNode(models.Node{ID: "node-123", Name: "different-name"}) assert.NotNil(t, result) assert.Equal(t, 60.0, result.CPUPackage) }) @@ -484,7 +485,7 @@ func TestGetHostAgentTemperature(t *testing.T) { } m.state.UpsertHost(host) - result := m.getHostAgentTemperatureByID("node-smart-only", "different-name") + result := m.getHostAgentTemperatureForNode(models.Node{ID: "node-smart-only", Name: "different-name"}) assert.NotNil(t, result) assert.True(t, result.Available) assert.True(t, result.HasSMART) @@ -573,115 +574,142 @@ func TestConvertHostSensorsToTemperature_ExtraBranches(t *testing.T) { }) } -func TestGetClusterSensorTemperature(t *testing.T) { - m := &Monitor{ +// readStateMonitor returns a monitor whose unified read state is the registry +// projection of nodes and hosts, as the poller sees it. Nodes are linked back to +// the hosts linked to them, as the agent linker leaves them. +func readStateMonitor(nodes []models.Node, hosts []models.Host) *Monitor { + for i := range nodes { + for _, host := range hosts { + if host.LinkedNodeID == nodes[i].ID { + nodes[i].LinkedAgentID = host.ID + } + } + } + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + adapter.PopulateFromSnapshot(models.StateSnapshot{Nodes: nodes, Hosts: hosts}) + return &Monitor{ state: models.NewState(), + resourceStore: adapter, clusterSensorsCache: make(map[string]clusterSensorsCacheEntry), } +} + +func readStateTestNode(instance, name string) models.Node { + return models.Node{ID: instance + "-" + name, Instance: instance, Name: name, Status: "online", LastSeen: time.Now()} +} + +func reportingTestHost(id, hostname, linkedNodeID string, cpuPackage float64) models.Host { + host := models.Host{ID: id, Hostname: hostname, Status: "online", LinkedNodeID: linkedNodeID, IntervalSeconds: 30, LastSeen: time.Now()} + if cpuPackage > 0 { + host.Sensors = models.HostSensorSummary{TemperatureCelsius: map[string]float64{"cpu_package": cpuPackage}} + } + return host +} + +func clusterSensorReport(nodeName string, cpuPackage float64) []agentshost.ClusterNodeSensors { + return []agentshost.ClusterNodeSensors{{ + NodeName: nodeName, + Sensors: agentshost.Sensors{TemperatureCelsius: map[string]float64{"cpu_package": cpuPackage, "cpu_core_0": cpuPackage - 3}}, + }} +} + +func TestGetClusterSensorTemperature(t *testing.T) { + m := readStateMonitor( + []models.Node{readStateTestNode("pve1", "node1"), readStateTestNode("pve1", "node2"), readStateTestNode("pve1", "MyNode"), readStateTestNode("pve1", "stale-node")}, + []models.Host{reportingTestHost("agent-node1", "node1", "pve1-node1", 0), reportingTestHost("agent-unlinked", "elsewhere", "", 0)}, + ) + lookup := func(nodeID, nodeName string) *models.Temperature { + readState := m.GetUnifiedReadStateOrSnapshot() + polled := models.Node{ID: nodeID, Name: nodeName} + slot := m.polledNodeSlot(readState.Hosts(), readState.Nodes(), polled) + return m.getClusterSensorTemperature(readState.Hosts(), readState.Nodes(), polled, slot) + } t.Run("empty cache returns nil", func(t *testing.T) { - result := m.getClusterSensorTemperature("node1") - assert.Nil(t, result) + assert.Nil(t, lookup("pve1-node2", "node2")) }) t.Run("cached data returned", func(t *testing.T) { - m.clusterSensorsCache["node2"] = clusterSensorsCacheEntry{ - sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{ - "cpu_package": 58.0, - "cpu_core_0": 55.0, - }, - }, - updatedAt: time.Now(), - } + m.applyClusterSensors("agent-node1", clusterSensorReport("node2", 58), time.Now()) - result := m.getClusterSensorTemperature("node2") - assert.NotNil(t, result) - assert.Equal(t, 58.0, result.CPUPackage) - assert.True(t, result.HasCPU) + result := lookup("pve1-node2", "node2") + if assert.NotNil(t, result) { + assert.Equal(t, 58.0, result.CPUPackage) + assert.True(t, result.HasCPU) + } }) t.Run("stale data returns nil", func(t *testing.T) { - m.clusterSensorsCache["stale-node"] = clusterSensorsCacheEntry{ - sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{"cpu_package": 50.0}, - }, - updatedAt: time.Now().Add(-5 * time.Minute), // older than 2min threshold - } + m.applyClusterSensors("agent-node1", clusterSensorReport("stale-node", 50), time.Now().Add(-3*time.Minute)) // older than 2min threshold - result := m.getClusterSensorTemperature("stale-node") - assert.Nil(t, result) + assert.Nil(t, lookup("pve1-stale-node", "stale-node")) }) t.Run("case insensitive lookup", func(t *testing.T) { - m.clusterSensorsCache["mynode"] = clusterSensorsCacheEntry{ - sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{"cpu_package": 60.0}, - }, - updatedAt: time.Now(), - } + m.applyClusterSensors("agent-node1", clusterSensorReport("mynode", 60), time.Now()) - result := m.getClusterSensorTemperature("MyNode") - assert.NotNil(t, result) - assert.Equal(t, 60.0, result.CPUPackage) + result := lookup("pve1-MyNode", "MyNode") + if assert.NotNil(t, result) { + assert.Equal(t, 60.0, result.CPUPackage) + } + }) + + t.Run("reading from an unlinked agent serves no node", func(t *testing.T) { + m.applyClusterSensors("agent-unlinked", clusterSensorReport("node1", 40), time.Now()) + + assert.Nil(t, lookup("pve1-node1", "node1")) + }) + + t.Run("node of another connection returns nil", func(t *testing.T) { + assert.Nil(t, lookup("", "node2")) + assert.Nil(t, lookup("pve9-node2", "node2")) + }) + + t.Run("node not yet in the read state is scoped by its own connection", func(t *testing.T) { + m.applyClusterSensors("agent-node1", clusterSensorReport("node9", 47), time.Now()) + readState := m.GetUnifiedReadStateOrSnapshot() + polled := readStateTestNode("pve1", "node9") + result := m.getClusterSensorTemperature(readState.Hosts(), readState.Nodes(), polled, m.polledNodeSlot(readState.Hosts(), readState.Nodes(), polled)) + if assert.NotNil(t, result) { + assert.Equal(t, 47.0, result.CPUPackage) + } }) t.Run("empty node name returns nil", func(t *testing.T) { - result := m.getClusterSensorTemperature("") - assert.Nil(t, result) + assert.Nil(t, lookup("pve1-node2", "")) }) } func TestGetHostAgentTemperatureByID_ClusterFallback(t *testing.T) { - m := &Monitor{ - state: models.NewState(), - clusterSensorsCache: make(map[string]clusterSensorsCacheEntry), - } + m := readStateMonitor( + []models.Node{readStateTestNode("pve1", "sibling"), readStateTestNode("pve1", "orphan-node")}, + []models.Host{reportingTestHost("agent-sibling", "sibling", "pve1-sibling", 0)}, + ) - // No host agent, but cluster cache has data - m.clusterSensorsCache["orphan-node"] = clusterSensorsCacheEntry{ - sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{ - "cpu_package": 62.0, - }, - }, - updatedAt: time.Now(), - } + // No host agent on the node, but a sibling's agent collected its sensors + m.applyClusterSensors("agent-sibling", clusterSensorReport("orphan-node", 62), time.Now()) - result := m.getHostAgentTemperatureByID("", "orphan-node") - assert.NotNil(t, result, "should fall back to cluster sensor cache") - assert.Equal(t, 62.0, result.CPUPackage) + result := m.getHostAgentTemperatureForNode(models.Node{ID: "pve1-orphan-node", Name: "orphan-node"}) + if assert.NotNil(t, result, "should fall back to cluster sensor cache") { + assert.Equal(t, 62.0, result.CPUPackage) + } } func TestGetHostAgentTemperatureByID_LocalAgentTakesPriority(t *testing.T) { - m := &Monitor{ - state: models.NewState(), - clusterSensorsCache: make(map[string]clusterSensorsCacheEntry), - } - - // Both local agent and cluster cache have data for the same node - m.state.UpsertHost(models.Host{ - ID: "host-local", - Hostname: "shared-node", - Sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{ - "cpu_package": 70.0, // local agent reports 70 - }, + m := readStateMonitor( + []models.Node{readStateTestNode("pve1", "sibling"), readStateTestNode("pve1", "shared-node")}, + []models.Host{ + reportingTestHost("agent-sibling", "sibling", "pve1-sibling", 0), + reportingTestHost("host-local", "shared-node", "pve1-shared-node", 70), // local agent reports 70 }, - LastSeen: time.Now(), - }) - m.clusterSensorsCache["shared-node"] = clusterSensorsCacheEntry{ - sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{ - "cpu_package": 55.0, // cluster cache says 55 - }, - }, - updatedAt: time.Now(), - } + ) + // Both the local agent and the cluster cache have data for the same node + m.applyClusterSensors("agent-sibling", clusterSensorReport("shared-node", 55), time.Now()) // cluster cache says 55 - result := m.getHostAgentTemperatureByID("", "shared-node") - assert.NotNil(t, result) - assert.Equal(t, 70.0, result.CPUPackage, "local agent data should take priority over cluster cache") + result := m.getHostAgentTemperatureForNode(models.Node{ID: "pve1-shared-node", Name: "shared-node"}) + if assert.NotNil(t, result) { + assert.Equal(t, 70.0, result.CPUPackage, "local agent data should take priority over cluster cache") + } } // An agent that stops reporting keeps its last sensors in state. Past the @@ -689,45 +717,23 @@ func TestGetHostAgentTemperatureByID_LocalAgentTakesPriority(t *testing.T) { // linked node as a live reading; the cluster cache (with its own recency) is // still consulted. func TestGetHostAgentTemperatureByID_IgnoresAgentPastReportingLease(t *testing.T) { - m := &Monitor{ - state: models.NewState(), - clusterSensorsCache: make(map[string]clusterSensorsCacheEntry), - } - m.state.UpsertHost(models.Host{ - ID: "host-silent", - Hostname: "silent-node", - LinkedNodeID: "node-silent", - IntervalSeconds: 30, - Sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{"cpu_package": 95.0}, - }, - LastSeen: time.Now().Add(-hostAgentHealthWindow(30) - time.Minute), - }) + nodes := []models.Node{readStateTestNode("pve1", "silent-node"), readStateTestNode("pve1", "sibling")} + sibling := reportingTestHost("agent-sibling", "sibling", "pve1-sibling", 0) + silent := reportingTestHost("host-silent", "silent-node", "pve1-silent-node", 95) + silent.LastSeen = time.Now().Add(-hostAgentHealthWindow(30) - time.Minute) + m := readStateMonitor(nodes, []models.Host{silent, sibling}) - assert.Nil(t, m.getHostAgentTemperatureByID("node-silent", "silent-node"), + assert.Nil(t, m.getHostAgentTemperatureForNode(models.Node{ID: "pve1-silent-node", Name: "silent-node"}), "a silent agent's retained sensors must not be presented as a current reading") - m.clusterSensorsCache["silent-node"] = clusterSensorsCacheEntry{ - sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{"cpu_package": 58.0}, - }, - updatedAt: time.Now(), - } - if result := m.getHostAgentTemperatureByID("node-silent", "silent-node"); assert.NotNil(t, result) { + m.applyClusterSensors("agent-sibling", clusterSensorReport("silent-node", 58), time.Now()) + if result := m.getHostAgentTemperatureForNode(models.Node{ID: "pve1-silent-node", Name: "silent-node"}); assert.NotNil(t, result) { assert.Equal(t, 58.0, result.CPUPackage, "a recent cluster-cache reading still serves the node") } - m.state.UpsertHost(models.Host{ - ID: "host-silent", - Hostname: "silent-node", - LinkedNodeID: "node-silent", - IntervalSeconds: 30, - Sensors: models.HostSensorSummary{ - TemperatureCelsius: map[string]float64{"cpu_package": 71.0}, - }, - LastSeen: time.Now(), - }) - if result := m.getHostAgentTemperatureByID("node-silent", "silent-node"); assert.NotNil(t, result) { + reporting := reportingTestHost("host-silent", "silent-node", "pve1-silent-node", 71) + m.resourceStore.(*unifiedresources.MonitorAdapter).PopulateFromSnapshot(models.StateSnapshot{Nodes: nodes, Hosts: []models.Host{reporting, sibling}}) + if result := m.getHostAgentTemperatureForNode(models.Node{ID: "pve1-silent-node", Name: "silent-node"}); assert.NotNil(t, result) { assert.Equal(t, 71.0, result.CPUPackage, "a reporting agent's reading is used again") } } @@ -804,7 +810,7 @@ func TestGetHostAgentTemperatureByID_UsesUnifiedReadState(t *testing.T) { resourceStore: unifiedresources.NewMonitorAdapter(registry), } - result := m.getHostAgentTemperatureByID("node-readstate", "ignored") + result := m.getHostAgentTemperatureForNode(models.Node{ID: "node-readstate", Name: "ignored"}) assert.NotNil(t, result) assert.Equal(t, 71.0, result.CPUPackage) if assert.Len(t, result.SMART, 1) { diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 65df2b3a2e..167d0ce22b 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1373,8 +1373,10 @@ func (m *Monitor) getRuntimeContext() context.Context { // clusterSensorsCacheEntry stores temperature data collected by a sibling agent via SSH. type clusterSensorsCacheEntry struct { - sensors models.HostSensorSummary - updatedAt time.Time + reporterID string // host agent that collected the reading over SSH + nodeName string // lowercase Proxmox node name the reading describes + sensors models.HostSensorSummary + updatedAt time.Time } type rrdMemCacheEntry struct { diff --git a/internal/monitoring/monitor_additional_test.go b/internal/monitoring/monitor_additional_test.go index 6a9b566dae..93a3e66fbe 100644 --- a/internal/monitoring/monitor_additional_test.go +++ b/internal/monitoring/monitor_additional_test.go @@ -1585,3 +1585,213 @@ func TestCarriedNodeTemperatureLeavesDiskReadingsOut(t *testing.T) { }) } } + +// Two Proxmox connections can each have a node with the same name (one "px1" +// per site). The poller's temperature lookup must never hand one site's node the +// host agent, or the agent's cluster-sibling SSH readings, of the other. +func TestNodeTemperatureStaysWithinSameNamedNodesSite(t *testing.T) { + instanceCfg := config.PVEInstance{Name: "siteB"} + poll := func(m *Monitor, instance, nodeName string) *models.Temperature { + m.config = &config.Config{TemperatureMonitoringEnabled: true} + polled := readStateTestNode(instance, nodeName) + m.collectNodeTemperatureData(context.Background(), instance, &instanceCfg, proxmox.Node{Node: nodeName}, &polled, nil, "online") + return polled.Temperature + } + sites := func() []models.Node { + return []models.Node{readStateTestNode("siteA", "px1"), readStateTestNode("siteA", "px2"), readStateTestNode("siteB", "px1")} + } + + t.Run("agent linked to the other site's node", func(t *testing.T) { + m := readStateMonitor(sites(), []models.Host{reportingTestHost("agent-a", "px1", "siteA-px1", 77)}) + if temp := poll(m, "siteA", "px1"); temp == nil || temp.CPUPackage != 77 { + t.Fatalf("siteA px1 temperature = %#v, want its linked agent's 77", temp) + } + if temp := poll(m, "siteB", "px1"); temp != nil { + t.Fatalf("siteB px1 took site A's agent reading: %#v", temp) + } + }) + + t.Run("unlinked agent whose hostname several nodes share", func(t *testing.T) { + m := readStateMonitor(sites(), []models.Host{reportingTestHost("agent-x", "px1", "", 66)}) + for _, instance := range []string{"siteA", "siteB"} { + if temp := poll(m, instance, "px1"); temp != nil { + t.Fatalf("%s px1 took an agent the linker could not place: %#v", instance, temp) + } + } + }) + + t.Run("unlinked agent whose hostname one node has", func(t *testing.T) { + m := readStateMonitor(sites(), []models.Host{reportingTestHost("agent-y", "px2", "", 61)}) + if temp := poll(m, "siteA", "px2"); temp == nil || temp.CPUPackage != 61 { + t.Fatalf("siteA px2 temperature = %#v, want the unambiguous agent's 61", temp) + } + }) + + t.Run("two unlinked agents with the node's hostname", func(t *testing.T) { + m := readStateMonitor(sites(), []models.Host{reportingTestHost("agent-y1", "px2", "", 61), reportingTestHost("agent-y2", "px2", "", 62)}) + if temp := poll(m, "siteA", "px2"); temp != nil { + t.Fatalf("siteA px2 took one of two candidate agents: %#v", temp) + } + }) + + t.Run("one cluster added through two connections", func(t *testing.T) { + // The state folds both views of pve01 into one slot, held under whichever + // view merged last, and the agent link follows the slot. Each view's poll + // must still take the agent's reading. + var instances []config.PVEInstance + for _, name := range []string{"enacon-a", "enacon-b"} { + instances = append(instances, config.PVEInstance{ + Name: name, IsCluster: true, ClusterName: "enacon", + ClusterEndpoints: []config.ClusterEndpoint{{NodeName: "pve01", Host: "https://192.168.1.11:8006", Fingerprint: "AA:AA"}}, + }) + } + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + m := &Monitor{config: &config.Config{TemperatureMonitoringEnabled: true, PVEInstances: instances}, state: models.NewState(), resourceStore: adapter} + view := func(instanceCfg *config.PVEInstance) models.Node { + return models.Node{ + ID: instanceCfg.Name + "-pve01", NodeIdentity: instanceCfg.Name + "-pve01", Name: "pve01", Instance: instanceCfg.Name, + ClusterName: "enacon", IsClusterMember: true, Host: "https://192.168.1.11:8006", + TLSFingerprint: pveNodeTLSFingerprint(instanceCfg, "pve01"), Status: "online", LastSeen: time.Now(), + } + } + m.state.UpdateNodesForInstance("enacon-a", []models.Node{view(&instances[0])}) + m.state.UpsertHost(reportingTestHost("agent-pve01", "pve01", "", 64)) + if err := m.state.LinkHostAgentToNode("agent-pve01", "enacon-a-pve01"); err != nil { + t.Fatalf("LinkHostAgentToNode: %v", err) + } + adapter.PopulateFromSnapshot(m.state.GetSnapshot()) + + for round := 0; round < 3; round++ { + for i := range instances { + polled := view(&instances[i]) + m.collectNodeTemperatureData(context.Background(), instances[i].Name, &instances[i], proxmox.Node{Node: "pve01"}, &polled, nil, "online") + if polled.Temperature == nil || polled.Temperature.CPUPackage != 64 { + t.Fatalf("round %d %s temperature = %#v, want the folded node's agent reading 64", round, instances[i].Name, polled.Temperature) + } + m.state.UpdateNodesForInstance(instances[i].Name, []models.Node{polled}) + adapter.PopulateFromSnapshot(m.state.GetSnapshot()) + } + } + if nodes := m.state.GetSnapshot().Nodes; len(nodes) != 1 { + t.Fatalf("state holds %d nodes, want both views folded into one", len(nodes)) + } + }) + + t.Run("unclustered node whose address two sites' namesakes share", func(t *testing.T) { + // Both site nodes would fold with the polled view by address alone; the + // state declines an ambiguous alias, so the lookup must take neither. + nodes := sites() + for i := range nodes { + nodes[i].ClusterName = map[string]string{"siteA": "a", "siteB": "b"}[nodes[i].Instance] + nodes[i].Host = "https://10.0.0.5:8006" + } + m := readStateMonitor(nodes, []models.Host{reportingTestHost("agent-a", "px1", "siteA-px1", 77)}) + polled := readStateTestNode("siteC", "px1") + polled.Host = "https://10.0.0.5:8006" + m.config = &config.Config{TemperatureMonitoringEnabled: true} + m.collectNodeTemperatureData(context.Background(), "siteC", &config.PVEInstance{Name: "siteC"}, proxmox.Node{Node: "px1"}, &polled, nil, "online") + if polled.Temperature != nil { + t.Fatalf("siteC px1 took a namesake's agent reading: %#v", polled.Temperature) + } + }) + + t.Run("multi-homed host added through two standalone connections", func(t *testing.T) { + // The agent reports both connection addresses, so the state folds the two + // views into one slot through it. Each view's poll must take its reading. + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + m := &Monitor{config: &config.Config{TemperatureMonitoringEnabled: true}, state: models.NewState(), resourceStore: adapter} + views := map[string]string{"lan-a": "https://10.0.0.5:8006", "lan-b": "https://10.0.1.5:8006"} + view := func(instance string) models.Node { + polled := readStateTestNode(instance, "minipc") + polled.Host = views[instance] + return polled + } + m.state.UpdateNodesForInstance("lan-a", []models.Node{view("lan-a")}) + agent := reportingTestHost("agent-minipc", "minipc", "", 64) + agent.ReportIP = "10.0.0.5" + agent.NetworkInterfaces = []models.HostNetworkInterface{{Name: "eth0", Addresses: []string{"10.0.0.5/24"}}, {Name: "eth1", Addresses: []string{"10.0.1.5/24"}}} + m.state.UpsertHost(agent) + if err := m.state.LinkHostAgentToNode("agent-minipc", "lan-a-minipc"); err != nil { + t.Fatalf("LinkHostAgentToNode: %v", err) + } + adapter.PopulateFromSnapshot(m.state.GetSnapshot()) + + for round := 0; round < 3; round++ { + for _, instance := range []string{"lan-a", "lan-b"} { + polled := view(instance) + m.collectNodeTemperatureData(context.Background(), instance, &config.PVEInstance{Name: instance}, proxmox.Node{Node: "minipc"}, &polled, nil, "online") + if polled.Temperature == nil || polled.Temperature.CPUPackage != 64 { + t.Fatalf("round %d %s temperature = %#v, want the folded host's agent reading 64", round, instance, polled.Temperature) + } + m.state.UpdateNodesForInstance(instance, []models.Node{polled}) + adapter.PopulateFromSnapshot(m.state.GetSnapshot()) + } + } + if nodes := m.state.GetSnapshot().Nodes; len(nodes) != 1 { + t.Fatalf("state holds %d nodes, want both views folded into one", len(nodes)) + } + }) + + t.Run("namesake's agent reporting the polled node's address", func(t *testing.T) { + // Site A's px1 agent also reports 10.0.1.5, the address site B's px1 + // uses. The fingerprints contradict, so site B's node is another machine. + siteA := readStateTestNode("siteA", "px1") + siteA.Host = "https://10.0.0.5:8006" + agent := reportingTestHost("agent-a", "px1", "siteA-px1", 77) + agent.ReportIP = "10.0.0.5" + agent.NetworkInterfaces = []models.HostNetworkInterface{{Name: "eth0", Addresses: []string{"10.0.0.5/24", "10.0.1.5/24"}}} + m := readStateMonitor([]models.Node{siteA}, []models.Host{agent}) + m.config = &config.Config{TemperatureMonitoringEnabled: true, PVEInstances: []config.PVEInstance{ + {Name: "siteA", Host: "https://10.0.0.5:8006", Fingerprint: "AA:AA"}, + {Name: "siteB", Host: "https://10.0.1.5:8006", Fingerprint: "BB:BB"}, + }} + polled := readStateTestNode("siteB", "px1") + polled.Host = "https://10.0.1.5:8006" + polled.TLSFingerprint = pveNodeTLSFingerprint(&m.config.PVEInstances[1], "px1") + m.collectNodeTemperatureData(context.Background(), "siteB", &m.config.PVEInstances[1], proxmox.Node{Node: "px1"}, &polled, nil, "online") + if polled.Temperature != nil { + t.Fatalf("siteB px1 took site A's agent reading: %#v", polled.Temperature) + } + }) + + t.Run("sibling reading across the two connections of one cluster", func(t *testing.T) { + // A cluster added twice can keep px1 under one connection's view and the + // reporting px2 under the other's; both connections list the same + // members with the same fingerprints, so the reading is px1's. + var instances []config.PVEInstance + for _, name := range []string{"enacon-a", "enacon-b"} { + instances = append(instances, config.PVEInstance{ + Name: name, IsCluster: true, ClusterName: "enacon", + ClusterEndpoints: []config.ClusterEndpoint{{NodeName: "px1", Fingerprint: "11:11"}, {NodeName: "px2", Fingerprint: "22:22"}}, + }) + } + member := func(instanceCfg *config.PVEInstance, name string) models.Node { + node := readStateTestNode(instanceCfg.Name, name) + node.ClusterName, node.IsClusterMember = "enacon", true + node.TLSFingerprint = pveNodeTLSFingerprint(instanceCfg, name) + return node + } + m := readStateMonitor( + []models.Node{member(&instances[0], "px1"), member(&instances[1], "px2")}, + []models.Host{reportingTestHost("agent-px2", "px2", "enacon-b-px2", 50)}, + ) + m.applyClusterSensors("agent-px2", clusterSensorReport("px1", 88), time.Now()) + m.config = &config.Config{TemperatureMonitoringEnabled: true, PVEInstances: instances} + polled := member(&instances[0], "px1") + m.collectNodeTemperatureData(context.Background(), "enacon-a", &instances[0], proxmox.Node{Node: "px1"}, &polled, nil, "online") + if polled.Temperature == nil || polled.Temperature.CPUPackage != 88 { + t.Fatalf("px1 temperature = %#v, want the sibling reading 88", polled.Temperature) + } + }) + + t.Run("cluster sibling reading from the other site", func(t *testing.T) { + m := readStateMonitor(sites(), []models.Host{reportingTestHost("agent-a2", "px2", "siteA-px2", 50)}) + m.applyClusterSensors("agent-a2", clusterSensorReport("px1", 88), time.Now()) + if temp := poll(m, "siteA", "px1"); temp == nil || temp.CPUPackage != 88 { + t.Fatalf("siteA px1 temperature = %#v, want its cluster sibling's 88", temp) + } + if temp := poll(m, "siteB", "px1"); temp != nil { + t.Fatalf("siteB px1 took site A's cluster sibling reading: %#v", temp) + } + }) +} diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 5fcf15d549..376fe3c101 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -3781,7 +3781,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. } // Store cluster peer sensor data if present and evict stale entries - m.applyClusterSensors(report.ClusterSensors, observedAt) + m.applyClusterSensors(host.ID, report.ClusterSensors, observedAt) // Availability results are ingested only once the host identity is // committed, because ownership is checked against that host ID. m.ApplyProbeAvailabilityResults(host.ID, probeAvailabilityResultsFromReport(report.AvailabilityResults)) @@ -4351,10 +4351,11 @@ func (m *Monitor) proxmoxPhysicalDiskMatchesForLinkedNode(linkedNodeID string) [ return matches } -// applyClusterSensors stores temperature data collected from Proxmox cluster -// siblings via SSH. Each entry is keyed by lowercase node name so that -// getHostAgentTemperatureByID can use it as a fallback. -func (m *Monitor) applyClusterSensors(entries []agentshost.ClusterNodeSensors, reportTime time.Time) { +// applyClusterSensors stores temperature data a host agent collected from its +// Proxmox cluster siblings via SSH, so that getHostAgentTemperatureForNode can use +// it as a fallback. Entries are keyed by reporting agent and lowercase node +// name: agents in different clusters can each report a sibling of the same name. +func (m *Monitor) applyClusterSensors(reporterID string, entries []agentshost.ClusterNodeSensors, reportTime time.Time) { // Fast path: nothing to add and cache is empty — skip lock if len(entries) == 0 { m.clusterSensorsMu.RLock() @@ -4377,7 +4378,9 @@ func (m *Monitor) applyClusterSensors(entries []agentshost.ClusterNodeSensors, r continue } - m.clusterSensorsCache[nodeName] = clusterSensorsCacheEntry{ + m.clusterSensorsCache[reporterID+"\x00"+nodeName] = clusterSensorsCacheEntry{ + reporterID: reporterID, + nodeName: nodeName, sensors: models.HostSensorSummary{ TemperatureCelsius: cloneStringFloatMap(entry.Sensors.TemperatureCelsius), FanRPM: cloneStringFloatMap(entry.Sensors.FanRPM), diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 3ef3414d7d..8fc230e0b6 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -972,6 +972,39 @@ func TestApplyHostReportRefreshesUnifiedReadStateWithoutBroadcast(t *testing.T) t.Fatal("accepted host report did not refresh the canonical headless read state") } +// An agent reports its Proxmox cluster siblings' sensors by bare node name, +// which another connection's node can share. Ingest records the reporting +// agent, so the readings serve only nodes of that agent's own connection. +func TestApplyHostReportScopesClusterSensorsToReportersConnection(t *testing.T) { + monitor := newTestMonitor(t) + monitor.clusterSensorsCache = make(map[string]clusterSensorsCacheEntry) + adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) + monitor.SetResourceStore(adapter) + monitor.state.UpdateNodesForInstance("siteA", []models.Node{readStateTestNode("siteA", "px1"), readStateTestNode("siteA", "px2")}) + monitor.state.UpdateNodesForInstance("siteB", []models.Node{readStateTestNode("siteB", "px1")}) + adapter.PopulateFromSnapshot(monitor.state.GetSnapshot()) + + host, err := monitor.ApplyHostReport(agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "agent-px2", IntervalSeconds: 30}, + Host: agentshost.HostInfo{ID: "machine-px2", Hostname: "px2", Platform: "linux"}, + ClusterSensors: clusterSensorReport("px1", 88), + Timestamp: time.Now().UTC(), + }, nil) + if err != nil { + t.Fatalf("ApplyHostReport: %v", err) + } + if host.LinkedNodeID != "siteA-px2" { + t.Fatalf("reporting agent linked to %q, want siteA-px2", host.LinkedNodeID) + } + + if temp := monitor.getHostAgentTemperatureForNode(readStateTestNode("siteA", "px1")); temp == nil || temp.CPUPackage != 88 { + t.Fatalf("siteA px1 temperature = %#v, want the sibling reading 88", temp) + } + if temp := monitor.getHostAgentTemperatureForNode(readStateTestNode("siteB", "px1")); temp != nil { + t.Fatalf("siteB px1 took site A's cluster sibling reading: %#v", temp) + } +} + func TestApplyHostReportProjectsAndPreservesLibvirtInventory(t *testing.T) { monitor := newTestMonitor(t) adapter := unifiedresources.NewMonitorAdapter(unifiedresources.NewRegistry(nil)) diff --git a/internal/monitoring/monitor_polling_node_helpers.go b/internal/monitoring/monitor_polling_node_helpers.go index e8ef1e849a..dd5e9abba9 100644 --- a/internal/monitoring/monitor_polling_node_helpers.go +++ b/internal/monitoring/monitor_polling_node_helpers.go @@ -164,9 +164,9 @@ func (m *Monitor) collectNodeTemperatureData( if effectiveStatus == "online" && tempMonitoringEnabled { // First, check if there's a matching host agent with temperature data. // Host agent temperatures are preferred because they don't require SSH access. - // Use getHostAgentTemperatureByID with the unique node ID to correctly handle - // duplicate hostname scenarios (e.g., two "px1" nodes on different IPs). - hostAgentTemp := m.getHostAgentTemperatureByID(modelNode.ID, node.Node) + // Match by the polled node's identity, not its name, so two connections' + // same-named nodes (e.g., "px1" at two sites) never share an agent. + hostAgentTemp := m.getHostAgentTemperatureForNode(*modelNode) if hostAgentTemp != nil { log.Debug(). Str("node", node.Node). @@ -331,7 +331,7 @@ func (m *Monitor) collectNodeTemperatureData( } } - if m.carriedTemperatureOutlivesAgentLease(modelNode.ID, node.Node, prevTemp, time.Now()) { + if m.carriedTemperatureOutlivesAgentLease(*modelNode, prevTemp, time.Now()) { // The reading came from a host agent that has stopped reporting. // Its reporting lease already rode out transient gaps, so // carrying it would outlive that lease. diff --git a/internal/monitoring/monitor_pve.go b/internal/monitoring/monitor_pve.go index 7f5405f896..8b03fac7e0 100644 --- a/internal/monitoring/monitor_pve.go +++ b/internal/monitoring/monitor_pve.go @@ -785,7 +785,7 @@ func (m *Monitor) boundCarriedNodeTemperature(node *models.Node, now time.Time) return } if now.Sub(node.Temperature.LastUpdate) > m.nodeTemperatureCarryWindow() || - m.carriedTemperatureOutlivesAgentLease(node.ID, node.Name, node.Temperature, now) { + m.carriedTemperatureOutlivesAgentLease(*node, node.Temperature, now) { node.Temperature = nil return }