keyspace: validate etcd health before adding meta-service groups

pkg/keyspace/meta_service_group.go

@@ -16,15 +16,19 @@ package keyspace
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
+	"strings"
 
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/log"
 
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/storage/kv"
+	"github.com/tikv/pd/pkg/utils/etcdutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/tikv/pd/server/config"
 )
@@ -310,6 +314,9 @@ func (m *MetaServiceGroupManager) UpdateGroupsSafely(
 	if err := config.AdjustMetaServiceGroups(metaServiceGroups); err != nil {
 		return err
 	}
+	if err := m.checkNewGroupsHealth(ctx, metaServiceGroups); err != nil {
+		return err
+	}
 	if err := m.persistGroupsLocked(ctx, metaServiceGroups, deletedGroups, persist); err != nil {
 		return err
 	}
@@ -319,6 +326,48 @@ func (m *MetaServiceGroupManager) UpdateGroupsSafely(
 	return nil
 }
 
+// checkNewGroupsHealth verifies every configured endpoint before a group is
+// added. Existing groups are intentionally skipped so an address update does
+// not change the established update semantics.
+func (m *MetaServiceGroupManager) checkNewGroupsHealth(ctx context.Context, metaServiceGroups map[string]string) error {
+	groups := make(map[string]string)
+	m.RLock()
+	for groupID, addresses := range metaServiceGroups {
+		if _, exists := m.metaServiceGroups[groupID]; !exists {
+			groups[groupID] = addresses
+		}
+	}
+	m.RUnlock()
+	for groupID, addresses := range groups {
+		for _, address := range strings.Split(addresses, ",") {
+			if err := checkEtcdServerHealth(ctx, strings.TrimSpace(address)); err != nil {
+				return fmt.Errorf("%w: group %s endpoint %s: %v", ErrMetaServiceGroupUnhealthy, groupID, address, err)
+			}
+		}
+	}
+	return nil
+}
+
+func checkEtcdServerHealth(ctx context.Context, endpoint string) error {
+	client, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{endpoint},
+		DialTimeout: etcdutil.DefaultRequestTimeout,
+	})
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err := client.Close(); err != nil {
+			log.Warn("[keyspace] failed to close meta-service group etcd client",
+				zap.String("endpoint", endpoint), zap.Error(err))
+		}
+	}()
+	if !etcdutil.IsHealthy(ctx, client) {
+		return errors.New("etcd health check failed")
+	}
+	return nil
+}
+
 // persistGroupsLocked performs the delete-guard check and persists the new
 // groups while holding the write lock, which blocks concurrent keyspace
 // assignment (AssignToGroup/PickGroup/reassign all take the read lock).

pkg/keyspace/meta_service_group_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/storage/kv"
+	"github.com/tikv/pd/pkg/utils/etcdutil"
 )
 
 type metaServiceGroupTestSuite struct {
@@ -257,6 +258,39 @@ func (suite *metaServiceGroupTestSuite) TestUpdateGroupsSafelyUsesAuthoritativeC
 	re.ErrorIs(err, ErrGroupHasAssignedKeyspaces)
 }
 
+func (suite *metaServiceGroupTestSuite) TestUpdateGroupsSafelyChecksNewGroupHealth() {
+	re := suite.Require()
+	servers, _, cleanup := etcdutil.NewTestEtcdCluster(suite.T(), 1, nil)
+	defer cleanup()
+	endpoint := servers[0].Config().ListenClientUrls[0].String()
+
+	groups := mockMetaServiceGroups()
+	groups["healthy"] = endpoint
+	persisted := false
+	err := suite.manager.UpdateGroupsSafely(suite.ctx, groups, nil, func() error {
+		persisted = true
+		return nil
+	}, nil)
+	re.NoError(err)
+	re.True(persisted)
+	re.Equal(endpoint, suite.manager.GetGroups()["healthy"])
+
+	updatedGroups := make(map[string]string, len(groups)+1)
+	for groupID, addresses := range groups {
+		updatedGroups[groupID] = addresses
+	}
+	updatedGroups["unhealthy"] = endpoint + ",http://127.0.0.1:1"
+	persisted = false
+	err = suite.manager.UpdateGroupsSafely(suite.ctx, updatedGroups, nil, func() error {
+		persisted = true
+		return nil
+	}, nil)
+	re.ErrorIs(err, ErrMetaServiceGroupUnhealthy)
+	re.False(persisted)
+	_, exists := suite.manager.GetGroups()["unhealthy"]
+	re.False(exists)
+}
+
 func (suite *metaServiceGroupTestSuite) TestAssignToGroupRejectsNegativeCount() {
 	re := suite.Require()
 	_, err := suite.manager.AssignToGroup(suite.ctx, -1)

pkg/keyspace/util.go

@@ -55,6 +55,9 @@ var (
 	// that still has keyspaces assigned to it. It is exported so HTTP handlers can
 	// map it to a 400 Bad Request via errors.Is.
 	ErrGroupHasAssignedKeyspaces = errors.New("cannot delete meta-service group with assigned keyspaces")
+	// ErrMetaServiceGroupUnhealthy is returned when a new meta-service group's
+	// etcd server cannot pass the health check.
+	ErrMetaServiceGroupUnhealthy = errors.New("meta-service group etcd server is unhealthy")
 
 	// stateTransitionTable lists all allowed next state for the given current state.
 	// Note that transit from any state to itself is allowed for idempotence.

server/apiv2/handlers/meta_service_group.go

@@ -139,7 +139,7 @@ func PatchMetaServiceGroups(c *gin.Context) {
 	}, func() {
 		svr.UpdateKeyspaceConfig(newCfg)
 	}); err != nil {
-		if errors.Is(err, keyspace.ErrGroupHasAssignedKeyspaces) {
+		if errors.Is(err, keyspace.ErrGroupHasAssignedKeyspaces) || errors.Is(err, keyspace.ErrMetaServiceGroupUnhealthy) {
 			c.AbortWithStatusJSON(http.StatusBadRequest, err.Error())
 			return
 		}

tests/server/apiv2/handlers/meta_service_group_test.go

@@ -112,7 +112,7 @@ func (suite *metaServiceGroupTestSuite) TestUpdateMetaServiceGroupsViaConfigAPI(
 	re := suite.Require()
 	// Adding a new group through /config should succeed and be visible via v2 API.
 	added := mockMetaServiceGroups()
-	added["etcd-group-x"] = "etcd-group-x.example.local"
+	added["etcd-group-x"] = suite.cluster.GetEtcdClient().Endpoints()[0]
 	code, body := suite.setMetaServiceGroupsViaConfig(re, added)
 	re.Equal(http.StatusOK, code, body)
 	groups := mustLoadMetaServiceGroups(re, suite.server)
@@ -124,7 +124,7 @@ func (suite *metaServiceGroupTestSuite) TestUpdateMetaServiceGroupsViaConfigAPI(
 		}
 	}
 	re.NotNil(x, "etcd-group-x should be added via /config")
-	re.Equal("etcd-group-x.example.local", x.Addresses)
+	re.Equal(suite.cluster.GetEtcdClient().Endpoints()[0], x.Addresses)
 
 	// Updating an existing group's address through /config should also work.
 	added["etcd-group-x"] = "etcd-group-x-modified.example.local"
@@ -138,6 +138,18 @@ func (suite *metaServiceGroupTestSuite) TestUpdateMetaServiceGroupsViaConfigAPI(
 	}
 }
 
+func (suite *metaServiceGroupTestSuite) TestUpdateMetaServiceGroupsViaConfigAPIRejectsUnhealthyGroup() {
+	re := suite.Require()
+	groups := mockMetaServiceGroups()
+	groups["unhealthy"] = "http://127.0.0.1:1"
+	code, body := suite.setMetaServiceGroupsViaConfig(re, groups)
+	re.Equal(http.StatusBadRequest, code, body)
+	re.Contains(body, "meta-service group etcd server is unhealthy")
+	for _, group := range mustLoadMetaServiceGroups(re, suite.server) {
+		re.NotEqual("unhealthy", group.ID)
+	}
+}
+
 func (suite *metaServiceGroupTestSuite) setMetaServiceGroupsViaConfig(re *require.Assertions, groups map[string]string) (int, string) {
 	payload, err := json.Marshal(map[string]any{"keyspace.meta-service-groups": groups})
 	re.NoError(err)
@@ -177,8 +189,8 @@ func (suite *metaServiceGroupTestSuite) TestMetaServiceGroupOperations() {
 		re.InDelta(collectedStatus.Status.AssignmentCount, len(keyspaces)/len(groups), 1)
 	}
 	// Add two more meta-service groups.
-	addr4 := "etcd-group-4.tidb-serverless.cluster.svc.local"
-	addr5 := "etcd-group-5.tidb-serverless.cluster.svc.local"
+	addr4 := suite.cluster.GetEtcdClient().Endpoints()[0]
+	addr5 := suite.cluster.GetEtcdClient().Endpoints()[0]
 	patch := map[string]*string{
 		"etcd-group-4": &addr4,
 		"etcd-group-5": &addr5,
@@ -253,7 +265,7 @@ func (suite *metaServiceGroupTestSuite) TestMetaServiceGroupOperations() {
 	mustPatchMetaServiceGroupsFail(re, suite.server, normalizedDuplicatePatch)
 
 	// Delete a newly-added group with no assigned keyspaces.
-	unusedAddr := "etcd-group-unused.tidb-serverless.cluster.svc.local"
+	unusedAddr := suite.cluster.GetEtcdClient().Endpoints()[0]
 	groups = mustPatchMetaServiceGroups(re, suite.server, map[string]*string{
 		"etcd-group-unused": &unusedAddr,
 	})

tools/pd-ctl/tests/meta_service_group/meta_service_group_test.go

@@ -93,10 +93,11 @@ func (suite *metaServiceGroupCLITestSuite) TestListMetaServiceGroups() {
 func (suite *metaServiceGroupCLITestSuite) TestUpsertMetaServiceGroup() {
 	re := suite.Require()
 	cmd := ctl.GetRootCmd()
+	endpoint := suite.cluster.GetEtcdClient().Endpoints()[0]
 
 	// Add a new group.
 	output, err := tests.ExecuteCommand(cmd, "-u", suite.pdAddr, "meta-service-group", "upsert",
-		"--group", "group-2=addr2.example.com")
+		"--group", "group-2="+endpoint)
 	re.NoError(err)
 	var groups []*handlers.MetaServiceGroupStatus
 	re.NoError(json.Unmarshal(output, &groups))
@@ -105,7 +106,7 @@ func (suite *metaServiceGroupCLITestSuite) TestUpsertMetaServiceGroup() {
 	for _, g := range groups {
 		if g.ID == "group-2" {
 			found = true
-			re.Equal("addr2.example.com", g.Addresses)
+			re.Equal(endpoint, g.Addresses)
 		}
 	}
 	re.True(found)
@@ -123,8 +124,8 @@ func (suite *metaServiceGroupCLITestSuite) TestUpsertMetaServiceGroup() {
 
 	// Upsert multiple groups at once.
 	output, err = tests.ExecuteCommand(cmd, "-u", suite.pdAddr, "meta-service-group", "upsert",
-		"--group", "group-3=addr3.example.com",
-		"--group", "group-4=addr4.example.com")
+		"--group", "group-3="+endpoint,
+		"--group", "group-4="+endpoint)
 	re.NoError(err)
 	re.NoError(json.Unmarshal(output, &groups))
 	re.Len(groups, 5)
@@ -249,7 +250,7 @@ func (suite *metaServiceGroupCLITestSuite) TestSetStatusEscapesSpecialCharID() {
 	re := suite.Require()
 	specialID := "group a?b#c%d"
 	_, err := tests.ExecuteCommand(ctl.GetRootCmd(), "-u", suite.pdAddr, "meta-service-group", "upsert",
-		"--group", specialID+"=addr-special.example.com")
+		"--group", specialID+"="+suite.cluster.GetEtcdClient().Endpoints()[0])
 	re.NoError(err)
 
 	output, err := tests.ExecuteCommand(ctl.GetRootCmd(), "-u", suite.pdAddr, "meta-service-group", "set-enabled", specialID, "true")