diff --git a/server/join/client_url.go b/server/join/client_url.go new file mode 100644 index 00000000000..846b1da859a --- /dev/null +++ b/server/join/client_url.go @@ -0,0 +1,141 @@ +// Copyright 2026 TiKV Project Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package join + +import ( + "encoding/hex" + "encoding/json" + "fmt" + + "go.etcd.io/etcd/api/v3/etcdserverpb" + clientv3 "go.etcd.io/etcd/client/v3" + "go.uber.org/zap" + + "github.com/pingcap/errors" + "github.com/pingcap/log" + + "github.com/tikv/pd/pkg/slice" + "github.com/tikv/pd/pkg/storage/kv" +) + +// clientURLOwner identifies the member that has claimed a client URL. Peer URLs +// are globally unique per member (etcd enforces peer-URL uniqueness) and stable +// across restarts, so they — not the name — decide whether a registry entry +// belongs to "me". This closes the gap where two nodes sharing a name would +// otherwise both accept a claim keyed only by name. +type clientURLOwner struct { + Name string `json:"name"` + PeerURLs []string `json:"peer_urls"` +} + +// clientURLRegistryPath returns the etcd key that records which member owns a +// given advertised client URL. The URL is hex-encoded so it forms a single, +// slash-free key segment. +func clientURLRegistryPath(clusterID uint64, clientURL string) string { + return fmt.Sprintf("/pd/%d/member/client-urls/%s", clusterID, hex.EncodeToString([]byte(clientURL))) +} + +// checkClientURLConflict rejects the member if any *other* member in the current +// member list already advertises one of advertiseClientURLs. "Other" is decided +// by peer URLs (unique per member), not name, so a different member that happens +// to share our name is still checked. This catches the common case where an +// existing, live member owns the URL — e.g. issue #10999. +// +// It is read-only (no etcd write, no quorum requirement), so it runs on every +// startup — fresh join and a restart that changed its advertise-client-urls +// alike — without ever blocking a member that needs to restart to *restore* +// quorum. +func checkClientURLConflict( + advertiseClientURLs []string, + advertisePeerURLs []string, + members []*etcdserverpb.Member, +) error { + for _, url := range advertiseClientURLs { + for _, m := range members { + // Skip only our own entry, matched by peer URLs so a different + // member sharing our name is still checked. + if slice.EqualWithoutOrder(m.PeerURLs, advertisePeerURLs) { + continue + } + for _, owned := range m.ClientURLs { + if owned == url { + return errors.Errorf( + "advertise-client-urls %q is already used by member %q (id %d)", + url, m.Name, m.ID) + } + } + } + } + return nil +} + +// claimClientURLs atomically claims each advertised client URL in an etcd +// registry via a create-if-absent transaction. Because the transaction is +// serialized through raft, two concurrent joiners cannot both take the same +// client URL: the first wins and a later joiner — even one sharing the same name +// — finds a different peer URL in the owner record and is rejected. +// +// The registry entry is keyed by URL and valued by the owner's identity (name + +// peer URLs). This writes to etcd (needs quorum) and is therefore only used on +// the fresh-join path, where quorum is required anyway for MemberAdd; a restart +// relies on the read-only checkClientURLConflict instead. +// +// Known limitation: the claim is persistent, so if a member is removed and a +// *different* member later wants to reuse its client URL, the stale entry must +// be cleaned up first. Cleanup on member removal is tracked as follow-up work. +func claimClientURLs( + client *clientv3.Client, + clusterID uint64, + name string, + advertiseClientURLs []string, + advertisePeerURLs []string, +) error { + self := clientURLOwner{Name: name, PeerURLs: advertisePeerURLs} + value, err := json.Marshal(self) + if err != nil { + return errors.WithStack(err) + } + for _, url := range advertiseClientURLs { + key := clientURLRegistryPath(clusterID, url) + resp, err := kv.NewSlowLogTxn(client). + If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)). + Then(clientv3.OpPut(key, string(value))). + Else(clientv3.OpGet(key)). + Commit() + if err != nil { + return errors.WithStack(err) + } + if resp.Succeeded { + continue // freshly claimed + } + // The key already exists; make sure the owner is this member (matched by + // peer URLs), otherwise reject. + kvs := resp.Responses[0].GetResponseRange().Kvs + if len(kvs) == 0 { + return errors.Errorf("failed to claim advertise-client-urls %q, please retry", url) + } + var owner clientURLOwner + if err := json.Unmarshal(kvs[0].Value, &owner); err != nil { + return errors.Errorf("advertise-client-urls %q is claimed by an unrecognized owner", url) + } + if !slice.EqualWithoutOrder(owner.PeerURLs, advertisePeerURLs) { + return errors.Errorf( + "advertise-client-urls %q is already claimed by member %q", url, owner.Name) + } + log.Info("re-claimed an advertised client URL owned by this member", + zap.String("name", name), zap.String("client-url", url)) + } + return nil +} diff --git a/server/join/join.go b/server/join/join.go index 60be78ce3a7..5736d2548bb 100644 --- a/server/join/join.go +++ b/server/join/join.go @@ -43,6 +43,39 @@ const ( // listMemberRetryTimes is the retry times of list member. var listMemberRetryTimes = 20 +var ( + // restartListMemberRetryTimes is how many times a restarting member retries + // listing cluster members (with exponential backoff) before skipping the + // best-effort advertise-client-urls uniqueness check. + restartListMemberRetryTimes = 3 + // restartListMemberBackoff is the initial backoff between those retries; it + // doubles each time (2s, 4s, 8s), which together with the per-request + // timeouts spans roughly 20s before giving up. + restartListMemberBackoff = 2 * time.Second +) + +// retryListEtcdMembers retries ListEtcdMembers with exponential backoff. It is +// used on the restart path so a transient failure does not immediately skip the +// client-URL uniqueness check; it returns the last error if all retries fail. +func retryListEtcdMembers(client *clientv3.Client) (*clientv3.MemberListResponse, error) { + var ( + listResp *clientv3.MemberListResponse + err error + ) + backoff := restartListMemberBackoff + for i := range restartListMemberRetryTimes { + time.Sleep(backoff) + backoff *= 2 + listResp, err = etcdutil.ListEtcdMembers(client.Ctx(), client) + if err == nil { + return listResp, nil + } + log.Warn("retry listing members on restart failed", + zap.Int("retry", i+1), errs.ZapError(err)) + } + return nil, err +} + // PrepareJoinCluster sends MemberAdd command to PD cluster, // and returns the initial configuration of the PD cluster. // @@ -97,14 +130,10 @@ func PrepareJoinCluster(cfg *config.Config) error { } initialCluster := "" - // Cases with data directory. - if isDataExist(filepath.Join(cfg.DataDir, "member")) { - cfg.InitialCluster = initialCluster - cfg.InitialClusterState = embed.ClusterStateFlagExisting - return nil - } + dataExists := isDataExist(filepath.Join(cfg.DataDir, "member")) - // Below are cases without data directory. + // A client to the target cluster is needed both to verify the advertised + // client URL is unique and, for a fresh join, to run MemberAdd. tlsConfig, err := cfg.Security.ToClientTLSConfig() if err != nil { return err @@ -118,15 +147,63 @@ func PrepareJoinCluster(cfg *config.Config) error { LogConfig: &lgc, }) if err != nil { + // A restart (data already exists) must not be blocked when the cluster + // is unreachable; the member starts from its local data directory, so + // the client-URL check is best-effort in that case. + if dataExists { + log.Warn("skip advertise-client-urls uniqueness check on restart: failed to create etcd client", + errs.ZapError(err)) + cfg.InitialCluster = initialCluster + cfg.InitialClusterState = embed.ClusterStateFlagExisting + return nil + } return errors.WithStack(err) } defer client.Close() listResp, err := etcdutil.ListEtcdMembers(client.Ctx(), client) if err != nil { + if !dataExists { + return err + } + // Restart: a transient failure should not silently skip the check. + // Retry with exponential backoff, and only skip (best-effort) if the + // cluster is still unreachable afterwards — the member must still be + // able to start from its local data directory. + listResp, err = retryListEtcdMembers(client) + if err != nil { + log.Warn("skip advertise-client-urls uniqueness check on restart: failed to list members after retries", + errs.ZapError(err)) + cfg.InitialCluster = initialCluster + cfg.InitialClusterState = embed.ClusterStateFlagExisting + return nil + } + } + + // Reject a member — whether joining fresh or restarting with a modified + // advertise-client-urls — whose advertised client URL is already owned by + // another member. This is read-only (no quorum needed) and runs before the + // local etcd (re)starts and republishes its attributes, so a duplicate never + // enters the member list. See issue #10999. + if err := checkClientURLConflict( + strings.Split(cfg.AdvertiseClientUrls, ","), + strings.Split(cfg.AdvertisePeerUrls, ","), + listResp.Members, + ); err != nil { return err } + // Cases with data directory: the local etcd reads its configuration from the + // data directory, nothing more to prepare here. The registry claim is + // intentionally skipped on this path — it writes to etcd (needs quorum), and + // a restart must not depend on quorum. + if dataExists { + cfg.InitialCluster = initialCluster + cfg.InitialClusterState = embed.ClusterStateFlagExisting + return nil + } + + // Below are cases without data directory. existed := false joinedFailedToStart := false advertisePeerURLs := strings.Split(cfg.AdvertisePeerUrls, ",") @@ -153,6 +230,19 @@ func PrepareJoinCluster(cfg *config.Config) error { return errors.New("missing data or join a duplicated pd") } + // Atomically claim the advertised client URLs before MemberAdd so two + // concurrent joiners cannot both take the same client URL. Only the + // fresh-join path reaches here, where quorum is required anyway. + if err := claimClientURLs( + client, + listResp.Header.GetClusterId(), + cfg.Name, + strings.Split(cfg.AdvertiseClientUrls, ","), + strings.Split(cfg.AdvertisePeerUrls, ","), + ); err != nil { + return err + } + var addResp *clientv3.MemberAddResponse failpoint.Inject("addMemberFailed", func() { diff --git a/tests/server/join/join_test.go b/tests/server/join/join_test.go index 939873c2c8e..aeb53d7c752 100644 --- a/tests/server/join/join_test.go +++ b/tests/server/join/join_test.go @@ -23,6 +23,7 @@ import ( "github.com/tikv/pd/pkg/utils/etcdutil" "github.com/tikv/pd/server" + "github.com/tikv/pd/server/config" "github.com/tikv/pd/server/join" "github.com/tikv/pd/tests" ) @@ -161,6 +162,78 @@ func TestFailedPDJoinsPreviousCluster(t *testing.T) { re.Error(join.PrepareJoinCluster(pd2.GetConfig())) } +// Regression test for https://github.com/tikv/pd/issues/10999: a member that +// tries to join while advertising a client URL already owned by another member +// must be rejected before it can corrupt membership — even though its peer URL +// is unique and --join points to a different string than the duplicated client +// URL, so the `cfg.Join == cfg.AdvertiseClientUrls` self-join check does not +// trigger. +func TestJoinWithDuplicateClientURL(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cluster, err := tests.NewTestCluster(ctx, 1) + defer cluster.Destroy() + re.NoError(err) + + re.NoError(cluster.RunInitialServers()) + re.NotEmpty(cluster.WaitLeader()) + + pd1 := cluster.GetServer("pd1") + client := pd1.GetEtcdClient() + dupClientURL := pd1.GetConfig().AdvertiseClientUrls + + // PrepareJoinCluster (run inside cluster.Join) detects that dupClientURL is + // already owned by pd1 and refuses the join before MemberAdd. + _, err = cluster.Join(ctx, func(conf *config.Config, _ string) { + conf.AdvertiseClientUrls = dupClientURL + }) + re.Error(err) + re.Contains(err.Error(), "already used by member") + + // Membership is untouched: the orphan never entered the member list. + members, err := etcdutil.ListEtcdMembers(ctx, client) + re.NoError(err) + re.Len(members.Members, 1) +} + +// Regression test for the restart path of https://github.com/tikv/pd/issues/10999: +// a member that restarts (its data directory already exists) after its +// advertise-client-urls was changed to a value owned by another member must +// still be rejected. The uniqueness check runs on every startup, not only on a +// fresh join, and before the local etcd republishes its attributes. +func TestRestartWithModifiedDuplicateClientURL(t *testing.T) { + re := require.New(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + cluster, err := tests.NewTestCluster(ctx, 1) + defer cluster.Destroy() + re.NoError(err) + re.NoError(cluster.RunInitialServers()) + re.NotEmpty(cluster.WaitLeader()) + + // Bring up a legitimate second member with its own unique client URL so it + // has a data directory, then stop it to simulate a restart. + pd2, err := cluster.Join(ctx) + re.NoError(err) + re.NoError(pd2.Run()) + re.NotEmpty(cluster.WaitLeader()) + re.NoError(pd2.Stop()) + + dupClientURL := cluster.GetServer("pd1").GetConfig().AdvertiseClientUrls + + // Restart pd2 with its advertise-client-urls changed to pd1's. Its data + // directory still exists, so PrepareJoinCluster takes the "data exists" + // path — which must still run the uniqueness check and reject the restart. + cfg := pd2.GetConfig() + cfg.AdvertiseClientUrls = dupClientURL + err = join.PrepareJoinCluster(cfg) + re.Error(err) + re.Contains(err.Error(), "already used by member") +} + // A PD joins itself. func TestPDJoinsItself(t *testing.T) { re := require.New(t)