PMM-15227 Remove stale HA replicas from Inventory

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

@@ -829,6 +829,7 @@ 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 disappear from **Inventory > Nodes** once the remaining pods restart
 
 To scale PMM server replicas:
 

managed/models/database.go

@@ -1528,6 +1528,12 @@ func setupPMMServerHAAgents(q *reform.Querier, params SetupDBParams) error {
 	// create PMM Server Node and associated Agents in HA mode
 	logrus.Infof("Setting up PMM Server agents in HA mode, Node ID: %s", params.HANodeID)
 
+	// Before the "agent already exists" early return, so restarted replicas still clean up.
+	err := RemoveStaleHANodes(q, params.HANodeID, params.HAPeers)
+	if err != nil {
+		return err
+	}
+
 	file, err := os.Open(AgentConfigFilePath)
 	if err != nil {
 		return err

managed/models/node_helpers.go

@@ -18,10 +18,12 @@ package models
 import (
 	"errors"
 	"fmt"
+	"net"
 	"strings"
 
 	"github.com/AlekSi/pointer"
 	"github.com/google/uuid"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"gopkg.in/reform.v1"
@@ -252,13 +254,19 @@ func CreateNode(q *reform.Querier, nodeType NodeType, params *CreateNodeParams)
 }
 
 // RemoveNode removes single Node.
-func RemoveNode(q *reform.Querier, id string, mode RemoveMode) error { //nolint:gocognit
+func RemoveNode(q *reform.Querier, id string, mode RemoveMode) error {
+	return removeNode(q, id, mode, false)
+}
+
+// removeNode removes a single Node. The allowPMMServerNode flag lifts the ban on Nodes flagged as PMM
+// Server Nodes; only the HA cleanup sets it, to reap replicas that are no longer part of the cluster.
+func removeNode(q *reform.Querier, id string, mode RemoveMode, allowPMMServerNode bool) error { //nolint:gocognit
 	n, err := FindNodeByID(q, id)
 	if err != nil {
 		return err
 	}
 
-	if n.IsPMMServerNode || id == PMMServerNodeID {
+	if id == PMMServerNodeID || (!allowPMMServerNode && n.IsPMMServerNode) {
 		return status.Error(codes.PermissionDenied, "PMM Server node can't be removed.")
 	}
 
@@ -334,3 +342,105 @@ func RemoveNode(q *reform.Querier, id string, mode RemoveMode) error { //nolint:
 	}
 	return nil
 }
+
+// RemoveStaleHANodes removes the PMM Server Nodes of HA replicas that are no longer configured peers,
+// e.g. after a scale-down. Peers are the source of truth because they are regenerated from the replica
+// count and restart every replica, while a missing memberlist member may just be restarting.
+func RemoveStaleHANodes(q *reform.Querier, haNodeID string, haPeers []string) error {
+	if len(haPeers) == 0 {
+		return nil
+	}
+
+	expected := make(map[string]struct{}, len(haPeers))
+	for _, peer := range haPeers {
+		name, ok := haPeerNodeName(peer)
+		if !ok {
+			// Trusting the rest would treat a partial list as the whole cluster and remove live replicas.
+			logrus.Warnf("Can't read a node name from PMM_HA_PEERS entry %q, skipping the removal of stale HA nodes.", peer)
+			return nil
+		}
+		expected[name] = struct{}{}
+	}
+
+	if _, ok := expected[haNodeID]; !ok {
+		logrus.Warnf("PMM_HA_PEERS %v doesn't list this node (PMM_HA_NODE_ID %q), skipping the removal of stale HA nodes.", haPeers, haNodeID)
+		return nil
+	}
+
+	nodes, err := FindNodes(q, NodeFilters{})
+	if err != nil {
+		return fmt.Errorf("failed to list Nodes for stale HA node cleanup: %w", err)
+	}
+
+	for _, node := range nodes {
+		// Only HA replicas set this flag; every other Node is one the user monitors.
+		if !node.IsPMMServerNode {
+			continue
+		}
+		if _, ok := expected[node.NodeName]; ok {
+			continue
+		}
+
+		monitored, err := haNodeMonitoredServices(q, node.NodeID)
+		if err != nil {
+			return err
+		}
+		if len(monitored) != 0 {
+			logrus.Warnf("Keeping stale HA node %q (%s): it still monitors services %v, which would be removed with it. "+
+				"Re-add them from a running replica and remove the node from Inventory.", node.NodeName, node.NodeID, monitored)
+			continue
+		}
+
+		err = removeNode(q, node.NodeID, RemoveCascade, true)
+		switch {
+		case err == nil:
+			logrus.Infof("Removed stale HA node %q (%s), it is not a part of the cluster anymore.", node.NodeName, node.NodeID)
+		case errors.Is(err, reform.ErrNoRows), status.Code(err) == codes.NotFound:
+			logrus.Infof("Stale HA node %q (%s) was already removed by another replica.", node.NodeName, node.NodeID)
+		default:
+			return fmt.Errorf("failed to remove stale HA node %q: %w", node.NodeName, err)
+		}
+	}
+
+	return nil
+}
+
+// haPeerNodeName maps a PMM_HA_PEERS entry ("pmm-ha-0.pmm-ha.pmm.svc.cluster.local:9761") to a Node
+// name: the first label is the pod's PMM_HA_NODE_ID. Reports false for entries with no name, like bare IPs.
+func haPeerNodeName(peer string) (string, bool) {
+	host, _, _ := strings.Cut(strings.TrimSpace(peer), ":")
+	if net.ParseIP(host) != nil {
+		return "", false
+	}
+	// "/" is memberlist's "name/address" form, "[" an IPv6 literal; neither starts with a node name.
+	label, _, _ := strings.Cut(host, ".")
+	if label == "" || strings.ContainsAny(label, "/[") {
+		return "", false
+	}
+	return label, true
+}
+
+// haNodeMonitoredServices returns the IDs of Services whose exporters run under a replica's pmm-agent.
+// Remote instances bind theirs to the replica that added them (see management.RDSService), so removing
+// that replica's Node takes them with it.
+func haNodeMonitoredServices(q *reform.Querier, nodeID string) ([]string, error) {
+	pmmAgents, err := FindPMMAgentsRunningOnNode(q, nodeID)
+	if err != nil {
+		return nil, err
+	}
+
+	var serviceIDs []string
+	for _, pmmAgent := range pmmAgents {
+		agents, err := FindAgents(q, AgentFilters{PMMAgentID: pmmAgent.AgentID})
+		if err != nil {
+			return nil, err
+		}
+		for _, agent := range agents {
+			if agent.ServiceID != nil {
+				serviceIDs = append(serviceIDs, *agent.ServiceID)
+			}
+		}
+	}
+
+	return serviceIDs, nil
+}

managed/models/node_helpers_test.go

@@ -267,3 +267,158 @@ func TestNodeHelpers(t *testing.T) {
 		require.Len(t, nodes, 2) // PMM Server + HA PMM Server node
 	})
 }
+
+func TestRemoveStaleHANodes(t *testing.T) {
+	sqlDB := testdb.Open(t, models.SetupFixtures, nil)
+	t.Cleanup(func() {
+		require.NoError(t, sqlDB.Close())
+	})
+
+	// Two HA replica Nodes, one with a node_exporter, plus an unrelated monitored Node.
+	setup := func(t *testing.T) (*reform.Querier, func(t *testing.T)) {
+		t.Helper()
+		db := reform.NewDB(sqlDB, postgresql.Dialect, reform.NewPrintfLogger(t.Logf))
+		tx, err := db.Begin()
+		require.NoError(t, err)
+		q := tx.Querier
+
+		for _, str := range []reform.Struct{
+			&models.Node{
+				NodeID:          "ha-node-1",
+				NodeType:        models.GenericNodeType,
+				NodeName:        "pmm-ha-1",
+				Address:         models.LocalhostAddr,
+				IsPMMServerNode: true,
+			},
+			&models.Agent{
+				AgentID:      "ha-agent-1",
+				AgentType:    models.PMMAgentType,
+				RunsOnNodeID: new("ha-node-1"),
+			},
+			&models.Node{
+				NodeID:          "ha-node-2",
+				NodeType:        models.GenericNodeType,
+				NodeName:        "pmm-ha-2",
+				Address:         models.LocalhostAddr,
+				IsPMMServerNode: true,
+			},
+			&models.Agent{
+				AgentID:      "ha-agent-2",
+				AgentType:    models.PMMAgentType,
+				RunsOnNodeID: new("ha-node-2"),
+			},
+			&models.Agent{
+				AgentID:    "ha-node-exporter-2",
+				AgentType:  models.NodeExporterType,
+				PMMAgentID: new("ha-agent-2"),
+				NodeID:     new("ha-node-2"),
+			},
+			&models.Node{
+				NodeID:   "monitored-node",
+				NodeType: models.GenericNodeType,
+				NodeName: "Monitored Node",
+			},
+		} {
+			require.NoError(t, q.Insert(str), "failed to INSERT %+v", str)
+		}
+
+		teardown := func(t *testing.T) {
+			t.Helper()
+			require.NoError(t, tx.Rollback())
+		}
+		return q, teardown
+	}
+
+	assertNodeExists := func(t *testing.T, q *reform.Querier, nodeID string) {
+		t.Helper()
+		_, err := models.FindNodeByID(q, nodeID)
+		assert.NoError(t, err)
+	}
+
+	t.Run("RemovesScaledDownReplicaWithItsAgents", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		peers := []string{"pmm-ha-0.pmm-ha.pmm.svc.cluster.local:9761", " pmm-ha-1.pmm-ha.pmm.svc.cluster.local "}
+		require.NoError(t, models.RemoveStaleHANodes(q, "pmm-ha-1", peers))
+
+		assertNodeExists(t, q, "ha-node-1")
+		_, err := models.FindAgentByID(q, "ha-agent-1")
+		require.NoError(t, err)
+
+		_, err = models.FindNodeByID(q, "ha-node-2")
+		tests.AssertGRPCErrorCode(t, codes.NotFound, err)
+
+		// the removal cascades to the agents of the stale node
+		for _, agentID := range []string{"ha-agent-2", "ha-node-exporter-2"} {
+			_, err := models.FindAgentByID(q, agentID)
+			tests.AssertGRPCErrorCode(t, codes.NotFound, err)
+		}
+
+		// neither monitored nodes nor the pre-HA pmm-server Node are touched
+		assertNodeExists(t, q, "monitored-node")
+		assertNodeExists(t, q, models.PMMServerNodeID)
+	})
+
+	t.Run("KeepsAllReplicasWhenNothingWasScaledDown", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		// a dotless host with a port is what a hand-written PMM_HA_PEERS looks like
+		peers := []string{"pmm-ha-1.pmm-ha:9761", "pmm-ha-2:9761"}
+		require.NoError(t, models.RemoveStaleHANodes(q, "pmm-ha-1", peers))
+
+		assertNodeExists(t, q, "ha-node-1")
+		assertNodeExists(t, q, "ha-node-2")
+	})
+
+	t.Run("KeepsScaledDownReplicaThatStillMonitorsServices", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		// an exporter for a remote instance, bound to the scaled-down replica's pmm-agent
+		for _, str := range []reform.Struct{
+			&models.Service{
+				ServiceID:   "rds-service",
+				ServiceType: models.MySQLServiceType,
+				ServiceName: "RDS instance",
+				NodeID:      "monitored-node",
+				Address:     new("rds.example.com"),
+				Port:        new(uint16(3306)),
+			},
+			&models.Agent{
+				AgentID:    "rds-exporter",
+				AgentType:  models.MySQLdExporterType,
+				PMMAgentID: new("ha-agent-2"),
+				ServiceID:  new("rds-service"),
+			},
+		} {
+			require.NoError(t, q.Insert(str), "failed to INSERT %+v", str)
+		}
+
+		peers := []string{"pmm-ha-0.pmm-ha:9761", "pmm-ha-1.pmm-ha:9761"}
+		require.NoError(t, models.RemoveStaleHANodes(q, "pmm-ha-1", peers))
+
+		assertNodeExists(t, q, "ha-node-2")
+		_, err := models.FindAgentByID(q, "rds-exporter")
+		require.NoError(t, err)
+	})
+
+	t.Run("DoesNothingWhenPeersCantBeTrusted", func(t *testing.T) {
+		q, teardown := setup(t)
+		defer teardown(t)
+
+		for _, peers := range [][]string{
+			{"pmm-ha-2.pmm-ha:9761"},                      // lists only the other replica
+			{"10.244.1.7:9761", "10.244.2.8:9761"},        // no node names to read
+			{"pmm-ha-1.pmm-ha:9761", "10.244.2.8:9761"},   // mixed: one entry hides a live replica
+			{"pmm-ha-1.pmm-ha:9761", "pmm-ha-2/10.0.0.2"}, // memberlist "name/address" form
+			nil,
+		} {
+			require.NoError(t, models.RemoveStaleHANodes(q, "pmm-ha-1", peers))
+
+			assertNodeExists(t, q, "ha-node-1")
+			assertNodeExists(t, q, "ha-node-2")
+		}
+	})
+}