PMM-15227 Remove stale HA replicas from Inventory

documentation/docs/install-pmm/install-HA-clustered.md

@@ -829,6 +829,20 @@ When you scale PMM HA up or down, **all PMM pods will be recreated**. This happe
   - HAProxy continues routing to available pods during rollout
   - No data loss (distributed storage)
   - Rolling update strategy minimizes downtime
+  - The Nodes of removed replicas get removed from **Inventory > Nodes** once the remaining pods restart, unless one of the conditions in the note below applies
+
+!!! info "When PMM keeps a stale Node"
+    The cleanup runs on each replica at startup, so its log messages appear on whichever pod restarted first. PMM logs a warning (`component=ha`) and keeps the Node when:
+
+    - the Node still monitors services, for example a remote instance that was added from that replica. Re-add those services from a running replica; the next restart removes the Node
+    - `PMM_HA_PEERS` carries no readable node names, for example bare IP addresses
+    - `PMM_HA_PEERS` names no peers at all, or does not list the pod doing the cleanup. Both contradict the peer list the chart generates, so they are reported as a failure to look for stale Nodes
+
+    To see what was skipped:
+
+    ```sh
+    kubectl exec <pmm-pod> -n pmm -c pmm-ha -- grep -i "stale HA node" /srv/logs/pmm-managed.log
+    ```
 
 To scale PMM server replicas:
 
@@ -1206,4 +1220,4 @@ This Tech Preview release is designed to gather community feedback before GA. Yo
 - What works well in your environment?
 - What's challenging or confusing?
 - What features are you missing?
-- How does performance compare to single-instance deployments?
+- How does performance compare to single-instance deployments?

managed/models/agent_helpers.go

@@ -218,8 +218,13 @@ func checkUniqueAgentID(q *reform.Querier, id string) error {
 type AgentFilters struct {
 	// Return only Agents started by this pmm-agent.
 	PMMAgentID string
+	// Return only Agents started by any of these pmm-agents. An empty slice is not a filter.
+	PMMAgentIDs []string
 	// Return only Agents that provide insights for that Node.
 	NodeID string
+	// Return only Agents attached to or running on that Node: node-level exporters, the pmm-agents
+	// themselves, and external exporters in pull mode.
+	OnNodeID string
 	// Return only Agents that provide insights for that Service.
 	ServiceID string
 	// Return Agents with provided type.
@@ -246,6 +251,14 @@ func FindAgents(q *reform.Querier, filters AgentFilters) ([]*Agent, error) {
 		args = append(args, filters.PMMAgentID)
 		idx++
 	}
+	if len(filters.PMMAgentIDs) != 0 {
+		p := strings.Join(q.Placeholders(idx, len(filters.PMMAgentIDs)), ", ")
+		conditions = append(conditions, "pmm_agent_id IN ("+p+")")
+		for _, id := range filters.PMMAgentIDs {
+			args = append(args, id)
+		}
+		idx += len(filters.PMMAgentIDs)
+	}
 	if filters.NodeID != "" {
 		_, err := FindNodeByID(q, filters.NodeID)
 		if err != nil {
@@ -255,6 +268,12 @@ func FindAgents(q *reform.Querier, filters AgentFilters) ([]*Agent, error) {
 		args = append(args, filters.NodeID)
 		idx++
 	}
+	if filters.OnNodeID != "" {
+		// No existence check: the callers tolerate a Node that another actor has just removed.
+		conditions = append(conditions, fmt.Sprintf("(runs_on_node_id = %s OR node_id = %s)", q.Placeholder(idx), q.Placeholder(idx+1)))
+		args = append(args, filters.OnNodeID, filters.OnNodeID)
+		idx += 2
+	}
 	if filters.ServiceID != "" {
 		_, err := FindServiceByID(q, filters.ServiceID)
 		if err != nil {
@@ -294,6 +313,12 @@ func FindAgents(q *reform.Querier, filters AgentFilters) ([]*Agent, error) {
 		return nil, err
 	}
 
+	// Decryption is not free; skip it when the caller has already gone away.
+	err = q.Context().Err()
+	if err != nil {
+		return nil, err
+	}
+
 	agents := make([]*Agent, len(structs))
 	for i, s := range structs {
 		decryptedAgent := DecryptAgent(*s.(*Agent)) //nolint:forcetypeassert

managed/models/agent_helpers_test.go

@@ -464,6 +464,67 @@ func TestAgentHelpers(t *testing.T) {
 		assert.Empty(t, agents)
 	})
 
+	t.Run("FindAgentsByOnNodeID", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		agents, err := models.FindAgents(q, models.AgentFilters{OnNodeID: "N1"})
+		require.NoError(t, err)
+		agentIDs := make([]string, len(agents))
+		for i, agent := range agents {
+			agentIDs[i] = agent.AgentID
+			assert.True(t,
+				pointer.GetString(agent.RunsOnNodeID) == "N1" || pointer.GetString(agent.NodeID) == "N1",
+				"%s is on neither runs_on_node_id nor node_id of N1", agent.AgentID)
+		}
+		assert.Contains(t, agentIDs, "A1")    // a pmm-agent running on the Node
+		assert.Contains(t, agentIDs, "A3")    // an exporter attached to the Node
+		assert.Contains(t, agentIDs, "A7")    // attached to the Node, but started by a pmm-agent on N2
+		assert.NotContains(t, agentIDs, "A2") // bound to a Service, not to the Node
+
+		// NodeID alone can't answer this: it never matches the pmm-agent running on the Node.
+		agents, err = models.FindAgents(q, models.AgentFilters{NodeID: "N1"})
+		require.NoError(t, err)
+		for _, agent := range agents {
+			assert.NotEqual(t, "A1", agent.AgentID)
+		}
+
+		// Unlike NodeID, OnNodeID does not probe for the Node, so an unknown one is not an error.
+		agents, err = models.FindAgents(q, models.AgentFilters{OnNodeID: "X1"})
+		require.NoError(t, err)
+		assert.Empty(t, agents)
+	})
+
+	t.Run("FindAgentsByPMMAgentIDs", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		agents, err := models.FindAgents(q, models.AgentFilters{PMMAgentIDs: []string{"A1"}})
+		require.NoError(t, err)
+		agentIDs := make([]string, len(agents))
+		for i, agent := range agents {
+			agentIDs[i] = agent.AgentID
+		}
+		assert.Equal(t, []string{"A2", "A3"}, agentIDs)
+
+		agents, err = models.FindAgents(q, models.AgentFilters{PMMAgentIDs: []string{"A1", "A4"}})
+		require.NoError(t, err)
+		agentIDs = make([]string, len(agents))
+		for i, agent := range agents {
+			agentIDs[i] = agent.AgentID
+		}
+		assert.Equal(t, []string{"A2", "A3", "A5", "A6", "A7"}, agentIDs)
+
+		// An empty slice is not a filter, so it matches everything - callers that build the list
+		// dynamically have to check for that themselves.
+		filtered, err := models.FindAgents(q, models.AgentFilters{PMMAgentIDs: nil})
+		require.NoError(t, err)
+		all, err := models.FindAgents(q, models.AgentFilters{})
+		require.NoError(t, err)
+		assert.Equal(t, all, filtered)
+		assert.NotEmpty(t, filtered)
+	})
+
 	t.Run("FindPMMAgentsForServicesOnNode", func(t *testing.T) {
 		q, teardown := setup(t)
 		defer teardown(t)

managed/models/agent_model.go

@@ -95,8 +95,8 @@ func GetRTAAgentTypes() []AgentType {
 var v2_42 = version.MustParse("2.42.0-0")
 
 // PMMServerAgentID is a special Agent ID representing pmm-agent on PMM Server.
-// It takes the value of "pmm-server" in regular non-HA setups, while in Active/Active HA setups
-// it is set to the actual pmm-agent's Agent ID, which is a UUID.
+// It takes the value of "pmm-server" in non-HA setups, while in HA setups it is set to the actual
+// pmm-agent's Agent ID, which is a UUID.
 var PMMServerAgentID = string("pmm-server")
 
 // AgentConfigFilePath is the default path to pmm-agent config file; it changes to /srv in HA setups.

managed/models/database.go

@@ -1297,6 +1297,8 @@ func SetupDB(ctx context.Context, sqlDB *sql.DB, params SetupDBParams) (*reform.
 		return nil, err
 	}
 
+	removeStaleHANodes(ctx, db, params.HANodeID, params.HAPeers)
+
 	return db, nil
 }
 
@@ -1517,6 +1519,58 @@ func migrateDB(db *reform.DB, params SetupDBParams) error {
 	})
 }
 
+// removeStaleHANodes drops the Inventory Nodes of HA replicas that were scaled away. Those rows are
+// cosmetic, so failures are only logged and never returned: tidying them up must not take a replica
+// with it.
+//
+// It runs at startup on every replica, deliberately not on the elected leader. PMM_HA_PEERS is
+// fixed into a process's environment when the pod starts and cannot be re-read, so only at startup
+// is it guaranteed to describe the current cluster - the pod was just created from the current
+// StatefulSet template. Leadership can be acquired much later: a rolling update recreates the
+// leader, forcing an election that a not-yet-updated pod can win, and that pod would then sweep
+// with a peer list older than the cluster. On a scale-up that deletes the new replica's Node and
+// Agents, which was reproduced on a 2 -> 3 scale-up. It also has to work without a leader at all,
+// since a replica scaled down to one cannot reach quorum.
+//
+// Replicas racing each other is harmless: each Node is removed in a transaction of its own, and one
+// that another replica already took is tolerated.
+//
+// The localHANodeID argument is the calling replica's own PMM_HA_NODE_ID; see FindStaleHANodes.
+func removeStaleHANodes(ctx context.Context, db *reform.DB, localHANodeID string, haPeers []string) {
+	if localHANodeID == "" {
+		return
+	}
+
+	l := logrus.WithFields(logrus.Fields{"component": "ha", "ha_node_id": localHANodeID})
+
+	nodes, err := FindStaleHANodes(db.WithContext(ctx), localHANodeID, haPeers)
+	if err != nil {
+		l.WithError(err).Warn("Failed to look for stale HA nodes.")
+		return
+	}
+
+	for _, node := range nodes {
+		nodeL := l.WithFields(logrus.Fields{"node_id": node.NodeID, "node_name": node.NodeName})
+
+		// A transaction per Node: a failure rolls that Node back whole instead of leaving it
+		// half-removed, and leaves the Nodes this sweep hasn't reached yet alone.
+		err := db.InTransactionContext(ctx, nil, func(tx *reform.TX) error {
+			return RemoveStaleHANode(tx.Querier, node.NodeID)
+		})
+		switch {
+		case err == nil:
+			nodeL.Info("Removed stale HA node, it is not a part of the cluster anymore.")
+		case errors.Is(err, reform.ErrNoRows), status.Code(err) == codes.NotFound:
+			nodeL.WithError(err).Info("Stale HA node was already removed by another replica.")
+		case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
+			nodeL.WithError(err).Info("Startup was cancelled, stopping the removal of stale HA nodes.")
+			return
+		default:
+			nodeL.WithError(err).Warn("Failed to remove a stale HA node, keeping it.")
+		}
+	}
+}
+
 type agentConfig struct {
 	ID string `yaml:"id"`
 }

managed/models/models_test.go

@@ -69,3 +69,36 @@ func TestLabels(t *testing.T) {
 		tests.AssertGRPCError(t, status.New(codes.InvalidArgument, `Invalid label name "__1".`), err)
 	})
 }
+
+// The two outcomes are not symmetric: an entry that yields no name stops the whole sweep, while one
+// that yields a name is trusted as naming a live replica. Reading a name out of an entry that carries
+// none would turn "keep every Node" into "remove every Node this entry didn't name".
+func TestHAPeerNodeName(t *testing.T) {
+	for _, tc := range []struct {
+		peer string
+		name string
+		ok   bool
+	}{
+		{peer: "pmm-ha-0.monitoring-service.pmm.svc.cluster.local", name: "pmm-ha-0", ok: true}, // what the chart renders
+		{peer: "pmm-ha-0.pmm-ha:9761", name: "pmm-ha-0", ok: true},
+		{peer: " pmm-ha-1.pmm-ha.pmm.svc.cluster.local ", name: "pmm-ha-1", ok: true}, // trimmed
+		{peer: "pmm-ha-2:9761", name: "pmm-ha-2", ok: true},                           // a dotless host with a port
+		{peer: "pmm-ha-2", name: "pmm-ha-2", ok: true},
+		{peer: "10.244.1.7"}, // bare IPv4, with and without a port
+		{peer: "10.244.1.7:9761"},
+		{peer: "2001:db8::7"}, // IPv6, unbracketed and bracketed
+		{peer: "[2001:db8::7]:9761"},
+		{peer: "[2001:db8::7]"},
+		{peer: "pmm-ha-2/10.0.0.2"}, // memberlist's "name/address" form
+		{peer: "pmm-ha-2/[2001:db8::7]:9761"},
+		{peer: ":9761"},
+		{peer: ""},
+		{peer: "   "},
+	} {
+		t.Run(tc.peer, func(t *testing.T) {
+			name, ok := haPeerNodeName(tc.peer)
+			assert.Equal(t, tc.ok, ok)
+			assert.Equal(t, tc.name, name)
+		})
+	}
+}