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"
+    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` does not list the pod that is doing the cleanup
+
+    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

@@ -427,6 +427,49 @@ func FindPMMAgentsRunningOnNode(q *reform.Querier, nodeID string) ([]*Agent, err
 	return res, nil
 }
 
+// FindAgentsOnNode returns Agents attached to or running on the Node: node-level exporters, the
+// pmm-agents themselves, and external exporters in pull mode.
+func FindAgentsOnNode(q *reform.Querier, nodeID string) ([]*Agent, error) {
+	structs, err := q.SelectAllFrom(AgentTable, "WHERE runs_on_node_id = $1 OR node_id = $1 ORDER BY agent_id", nodeID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to select Agents on Node %q: %w", nodeID, err)
+	}
+
+	res := make([]*Agent, len(structs))
+	for i, str := range structs {
+		decryptedAgent := DecryptAgent(*str.(*Agent)) //nolint:forcetypeassert
+		res[i] = &decryptedAgent
+	}
+
+	return res, nil
+}
+
+// FindAgentsByPMMAgentIDs returns Agents started by any of the given pmm-agents.
+func FindAgentsByPMMAgentIDs(q *reform.Querier, pmmAgentIDs []string) ([]*Agent, error) {
+	if len(pmmAgentIDs) == 0 {
+		return []*Agent{}, nil
+	}
+
+	p := strings.Join(q.Placeholders(1, len(pmmAgentIDs)), ", ")
+	tail := fmt.Sprintf("WHERE pmm_agent_id IN (%s) ORDER BY agent_id", p)
+	args := make([]any, len(pmmAgentIDs))
+	for i, id := range pmmAgentIDs {
+		args[i] = id
+	}
+	structs, err := q.SelectAllFrom(AgentTable, tail, args...)
+	if err != nil {
+		return nil, fmt.Errorf("failed to select Agents started by pmm-agents: %w", err)
+	}
+
+	res := make([]*Agent, len(structs))
+	for i, str := range structs {
+		decryptedAgent := DecryptAgent(*str.(*Agent)) //nolint:forcetypeassert
+		res[i] = &decryptedAgent
+	}
+
+	return res, nil
+}
+
 // FindPMMAgentsForService gets pmm-agents for service.
 func FindPMMAgentsForService(q *reform.Querier, serviceID string) ([]*Agent, error) {
 	_, err := q.SelectOneFrom(ServiceTable, "WHERE service_id = $1", serviceID)

managed/models/agent_helpers_test.go

@@ -464,6 +464,55 @@ func TestAgentHelpers(t *testing.T) {
 		assert.Empty(t, agents)
 	})
 
+	t.Run("FindAgentsOnNode", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		agents, err := models.FindAgentsOnNode(q, "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
+
+		// find with non existing node.
+		agents, err = models.FindAgentsOnNode(q, "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.FindAgentsByPMMAgentIDs(q, []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.FindAgentsByPMMAgentIDs(q, []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)
+
+		agents, err = models.FindAgentsByPMMAgentIDs(q, nil)
+		require.NoError(t, err)
+		assert.Empty(t, agents)
+	})
+
 	t.Run("FindPMMAgentsForServicesOnNode", func(t *testing.T) {
 		q, teardown := setup(t)
 		defer teardown(t)

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)
+
 	return db, nil
 }
 
@@ -1517,6 +1519,44 @@ 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 this runs outside the migration transaction and only logs failures: tidying them up
+// must never keep a replica from starting.
+func removeStaleHANodes(ctx context.Context, db *reform.DB, params SetupDBParams) {
+	if params.HANodeID == "" || params.SetupFixtures == SkipFixtures {
+		return
+	}
+
+	l := logrus.WithFields(logrus.Fields{"component": "ha", "ha_node_id": params.HANodeID})
+
+	nodes, err := StaleHANodes(db.WithContext(ctx), params.HANodeID, params.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).Warn("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)
+		})
+	}
+}