Repository: tikv/pd (go)

Issue:
pd can not select new leader which lead qps drop to zero when inject pdleader io delay 500ms last for 5mins
## Bug Report

<!-- Please answer these questions before submitting your issue. Thanks! -->

### What did you do?
1、run workload
2、inject pdleader io delay 500ms last for 5mins
apiVersion: chaos-mesh.org/v1alpha1
kind: IOChaos
metadata:
  name: kv-timeout-data
  namespace: testbed-xxx
spec:
  action: latency
  mode: one
  selector:
    namespaces:
      - testbed-xxx
    labelSelectors:
      statefulset.kubernetes.io/pod-name: tc-pd-0
  volumePath: /var/lib/pd
  path: "/var/lib/pd/data/**/*"
  delay: "500ms"
  percent: 100
  duration: "300s"


### What did you expect to see?
1、pd can  select new leader when inject pdleader io delay 500ms last for 5mins
2、qps can recover within 2mins when inject pdleader io delay 500ms

### What did you see instead?
pd can not select new leader when inject pdleader io delay 500ms last for 5mins
write qps drop to zero
![3bf7daa4-ac99-4ae7-8651-52b59d4ea66e](https://github.com/tikv/pd/assets/84712107/5c53b413-8925-403c-8571-1a16dec97b8f)
![46775ab0-f274-434b-a4b8-b61f1b71e352](https://github.com/tikv/pd/assets/84712107/85b900dd-1b7b-45e2-9df0-84e9d0e34e03)

### What version of PD are you using (`pd-server -V`)?
./pd-server -V
 Release Version: v6.5.0-nightly
Edition: Community
Git Commit Hash: 77d6f5b6785eb4e153495bf1c32b165019ae119e
Git Branch: heads/refs/tags/v6.5.0-nightly
UTC Build Time:  2023-10-07 10:21:09
2023-10-19T09:33:53.650+0800




Pull request description:
*: cherry-pick the etcd client health checker improvements
### What problem does this PR solve?

<!--

Please create an issue first to describe the problem.
There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".
For more info, check https://github.com/tikv/pd/blob/master/CONTRIBUTING.md#linking-issues.

-->
Issue Number: close #7251. Cherry-pick part of code from #7725. #7727, #7743,  #7737 and #7779.

### What is changed and how does it work?

<!--

You could use the "commit message" code block to add more description to the final commit message.
For more info, check https://github.com/tikv/pd/blob/master/CONTRIBUTING.md#format-of-the-commit-message.

-->

```commit-message
Cherry-pick the etcd client health checker improvements from #7725. #7727, #7743,  #7737 and #7779.
```

### Check List

<!-- Remove the items that are not applicable. -->

Tests

<!-- At least one of these tests must be included. -->

- Unit test
- Integration test
- Manual test

![image](https://github.com/tikv/pd/assets/1446531/c21e2e48-6774-4619-92b9-781ad8d5411a)

![image](https://github.com/tikv/pd/assets/1446531/c520966e-44f4-4037-9d7f-5b64b4ec7cb9)

### Release note

<!--

A bugfix or a new feature needs a release note. If there is no need to give a release note, just leave it with the `None`.

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

-->

```release-note
Enhance PD availability in the event of IO hang causing node unavailability.
```


Patch:
diff --git a/pkg/mcs/utils/util.go b/pkg/mcs/utils/util.go
index 682e73f20..fa4e40664 100644
--- a/pkg/mcs/utils/util.go
+++ b/pkg/mcs/utils/util.go
@@ -161,12 +161,12 @@ func InitClient(s server) error {
 	if err != nil {
 		return err
 	}
-	etcdClient, httpClient, err := etcdutil.CreateClients(tlsConfig, backendUrls)
+	etcdClient, err := etcdutil.CreateEtcdClient(tlsConfig, backendUrls, "mcs-etcd-client")
 	if err != nil {
 		return err
 	}
 	s.SetETCDClient(etcdClient)
-	s.SetHTTPClient(httpClient)
+	s.SetHTTPClient(etcdutil.CreateHTTPClient(tlsConfig))
 	return nil
 }
 
diff --git a/pkg/member/member.go b/pkg/member/member.go
index fe2188b8e..0daec8e8d 100644
--- a/pkg/member/member.go
+++ b/pkg/member/member.go
@@ -355,7 +355,7 @@ func (m *EmbeddedEtcdMember) ResignEtcdLeader(ctx context.Context, from string,
 	log.Info("try to resign etcd leader to next pd-server", zap.String("from", from), zap.String("to", nextEtcdLeader))
 	// Determine next etcd leader candidates.
 	var etcdLeaderIDs []uint64
-	res, err := etcdutil.ListEtcdMembers(m.client)
+	res, err := etcdutil.ListEtcdMembers(ctx, m.client)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/utils/etcdutil/etcdutil.go b/pkg/utils/etcdutil/etcdutil.go
index e004247c6..44248b9a6 100644
--- a/pkg/utils/etcdutil/etcdutil.go
+++ b/pkg/utils/etcdutil/etcdutil.go
@@ -17,7 +17,6 @@ package etcdutil
 import (
 	"context"
 	"crypto/tls"
-	"fmt"
 	"math/rand"
 	"net/http"
 	"net/url"
@@ -107,13 +106,13 @@ func AddEtcdMember(client *clientv3.Client, urls []string) (*clientv3.MemberAddR
 }
 
 // ListEtcdMembers returns a list of internal etcd members.
-func ListEtcdMembers(client *clientv3.Client) (*clientv3.MemberListResponse, error) {
+func ListEtcdMembers(ctx context.Context, client *clientv3.Client) (*clientv3.MemberListResponse, error) {
 	failpoint.Inject("SlowEtcdMemberList", func(val failpoint.Value) {
 		d := val.(int)
 		time.Sleep(time.Duration(d) * time.Second)
 	})
-	ctx, cancel := context.WithTimeout(client.Ctx(), DefaultRequestTimeout)
-	listResp, err := client.MemberList(ctx)
+	newCtx, cancel := context.WithTimeout(ctx, DefaultRequestTimeout)
+	listResp, err := client.MemberList(newCtx)
 	cancel()
 	if err != nil {
 		return listResp, errs.ErrEtcdMemberList.Wrap(err).GenWithStackByCause()
@@ -247,7 +246,7 @@ func newClient(tlsConfig *tls.Config, endpoints ...string) (*clientv3.Client, er
 }
 
 // CreateEtcdClient creates etcd v3 client with detecting endpoints.
-func CreateEtcdClient(tlsConfig *tls.Config, acURLs []url.URL) (*clientv3.Client, error) {
+func CreateEtcdClient(tlsConfig *tls.Config, acURLs []url.URL, sourceOpt ...string) (*clientv3.Client, error) {
 	urls := make([]string, 0, len(acURLs))
 	for _, u := range acURLs {
 		urls = append(urls, u.String())
@@ -264,185 +263,16 @@ func CreateEtcdClient(tlsConfig *tls.Config, acURLs []url.URL) (*clientv3.Client
 	failpoint.Inject("closeTick", func() {
 		failpoint.Return(client, err)
 	})
-
-	checker := &healthyChecker{
-		tlsConfig: tlsConfig,
+	source := "default-etcd-client"
+	if len(sourceOpt) > 0 {
+		source = sourceOpt[0]
 	}
-	eps := syncUrls(client)
-	checker.update(eps)
-
-	// Create a goroutine to check the health of etcd endpoints periodically.
-	go func(client *clientv3.Client) {
-		defer logutil.LogPanic()
-		ticker := time.NewTicker(tickerInterval)
-		defer ticker.Stop()
-		lastAvailable := time.Now()
-		for {
-			select {
-			case <-client.Ctx().Done():
-				log.Info("etcd client is closed, exit health check goroutine")
-				checker.Range(func(key, value interface{}) bool {
-					client := value.(*healthyClient)
-					client.Close()
-					return true
-				})
-				return
-			case <-ticker.C:
-				usedEps := client.Endpoints()
-				healthyEps := checker.patrol(client.Ctx())
-				if len(healthyEps) == 0 {
-					// when all endpoints are unhealthy, try to reset endpoints to update connect
-					// rather than delete them to avoid there is no any endpoint in client.
-					// Note: reset endpoints will trigger subconn closed, and then trigger reconnect.
-					// otherwise, the subconn will be retrying in grpc layer and use exponential backoff,
-					// and it cannot recover as soon as possible.
-					if time.Since(lastAvailable) > etcdServerDisconnectedTimeout {
-						log.Info("no available endpoint, try to reset endpoints", zap.Strings("last-endpoints", usedEps))
-						client.SetEndpoints([]string{}...)
-						client.SetEndpoints(usedEps...)
-					}
-				} else {
-					if !typeutil.AreStringSlicesEquivalent(healthyEps, usedEps) {
-						client.SetEndpoints(healthyEps...)
-						change := fmt.Sprintf("%d->%d", len(usedEps), len(healthyEps))
-						etcdStateGauge.WithLabelValues("endpoints").Set(float64(len(healthyEps)))
-						log.Info("update endpoints", zap.String("num-change", change),
-							zap.Strings("last-endpoints", usedEps), zap.Strings("endpoints", client.Endpoints()))
-					}
-					lastAvailable = time.Now()
-				}
-			}
-		}
-	}(client)
-
-	// Notes: use another goroutine to update endpoints to avoid blocking health check in the first goroutine.
-	go func(client *clientv3.Client) {
-		defer logutil.LogPanic()
-		ticker := time.NewTicker(tickerInterval)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-client.Ctx().Done():
-				log.Info("etcd client is closed, exit update endpoint goroutine")
-				return
-			case <-ticker.C:
-				eps := syncUrls(client)
-				checker.update(eps)
-			}
-		}
-	}(client)
-
+	initHealthChecker(tickerInterval, tlsConfig, client, source)
 	return client, err
 }
 
-type healthyClient struct {
-	*clientv3.Client
-	lastHealth time.Time
-}
-
-type healthyChecker struct {
-	sync.Map  // map[string]*healthyClient
-	tlsConfig *tls.Config
-}
-
-func (checker *healthyChecker) patrol(ctx context.Context) []string {
-	// See https://github.com/etcd-io/etcd/blob/85b640cee793e25f3837c47200089d14a8392dc7/etcdctl/ctlv3/command/ep_command.go#L105-L145
-	var wg sync.WaitGroup
-	count := 0
-	checker.Range(func(key, value interface{}) bool {
-		count++
-		return true
-	})
-	hch := make(chan string, count)
-	healthyList := make([]string, 0, count)
-	checker.Range(func(key, value interface{}) bool {
-		wg.Add(1)
-		go func(key, value interface{}) {
-			defer wg.Done()
-			defer logutil.LogPanic()
-			ep := key.(string)
-			client := value.(*healthyClient)
-			if IsHealthy(ctx, client.Client) {
-				hch <- ep
-				checker.Store(ep, &healthyClient{
-					Client:     client.Client,
-					lastHealth: time.Now(),
-				})
-				return
-			}
-		}(key, value)
-		return true
-	})
-	wg.Wait()
-	close(hch)
-	for h := range hch {
-		healthyList = append(healthyList, h)
-	}
-	return healthyList
-}
-
-func (checker *healthyChecker) update(eps []string) {
-	for _, ep := range eps {
-		// check if client exists, if not, create one, if exists, check if it's offline or disconnected.
-		if client, ok := checker.Load(ep); ok {
-			lastHealthy := client.(*healthyClient).lastHealth
-			if time.Since(lastHealthy) > etcdServerOfflineTimeout {
-				log.Info("some etcd server maybe offline", zap.String("endpoint", ep))
-				checker.Delete(ep)
-			}
-			if time.Since(lastHealthy) > etcdServerDisconnectedTimeout {
-				// try to reset client endpoint to trigger reconnect
-				client.(*healthyClient).Client.SetEndpoints([]string{}...)
-				client.(*healthyClient).Client.SetEndpoints(ep)
-			}
-			continue
-		}
-		checker.addClient(ep, time.Now())
-	}
-}
-
-func (checker *healthyChecker) addClient(ep string, lastHealth time.Time) {
-	client, err := newClient(checker.tlsConfig, ep)
-	if err != nil {
-		log.Error("failed to create etcd healthy client", zap.Error(err))
-		return
-	}
-	checker.Store(ep, &healthyClient{
-		Client:     client,
-		lastHealth: lastHealth,
-	})
-}
-
-func syncUrls(client *clientv3.Client) []string {
-	// See https://github.com/etcd-io/etcd/blob/85b640cee793e25f3837c47200089d14a8392dc7/clientv3/client.go#L170-L183
-	ctx, cancel := context.WithTimeout(clientv3.WithRequireLeader(client.Ctx()), DefaultRequestTimeout)
-	defer cancel()
-	mresp, err := client.MemberList(ctx)
-	if err != nil {
-		log.Error("failed to list members", errs.ZapError(err))
-		return []string{}
-	}
-	var eps []string
-	for _, m := range mresp.Members {
-		if len(m.Name) != 0 && !m.IsLearner {
-			eps = append(eps, m.ClientURLs...)
-		}
-	}
-	return eps
-}
-
-// CreateClients creates etcd v3 client and http client.
-func CreateClients(tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client, *http.Client, error) {
-	client, err := CreateEtcdClient(tlsConfig, acUrls)
-	if err != nil {
-		return nil, nil, errs.ErrNewEtcdClient.Wrap(err).GenWithStackByCause()
-	}
-	httpClient := createHTTPClient(tlsConfig)
-	return client, httpClient, nil
-}
-
-// createHTTPClient creates a http client with the given tls config.
-func createHTTPClient(tlsConfig *tls.Config) *http.Client {
+// CreateHTTPClient creates a http client with the given tls config.
+func CreateHTTPClient(tlsConfig *tls.Config) *http.Client {
 	// FIXME: Currently, there is no timeout set for certain requests, such as GetRegions,
 	// which may take a significant amount of time. However, it might be necessary to
 	// define an appropriate timeout in the future.
diff --git a/pkg/utils/etcdutil/health_checker.go b/pkg/utils/etcdutil/health_checker.go
new file mode 100644
index 000000000..d1a57e97c
--- /dev/null
+++ b/pkg/utils/etcdutil/health_checker.go
@@ -0,0 +1,462 @@
+// Copyright 2024 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 etcdutil
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/pingcap/log"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/utils/logutil"
+	"github.com/tikv/pd/pkg/utils/typeutil"
+	"go.etcd.io/etcd/clientv3"
+	"go.uber.org/zap"
+)
+
+const pickedCountThreshold = 3
+
+// healthyClient will wrap an etcd client and record its last health time.
+// The etcd client inside will only maintain one connection to the etcd server
+// to make sure each healthyClient could be used to check the health of a certain
+// etcd endpoint without involving the load balancer of etcd client.
+type healthyClient struct {
+	*clientv3.Client
+	lastHealth  time.Time
+	healthState prometheus.Gauge
+	latency     prometheus.Observer
+}
+
+// healthChecker is used to check the health of etcd endpoints. Inside the checker,
+// we will maintain a map from each available etcd endpoint to its healthyClient.
+type healthChecker struct {
+	source         string
+	tickerInterval time.Duration
+	tlsConfig      *tls.Config
+
+	// Store as endpoint(string) -> *healthyClient
+	healthyClients sync.Map
+	// evictedEps records the endpoints which are evicted from the last health patrol,
+	// the value is the count the endpoint being picked continuously after evicted.
+	// Store as endpoint(string) -> pickedCount(int)
+	evictedEps sync.Map
+	// client is the etcd client the health checker is guarding, it will be set with
+	// the checked healthy endpoints dynamically and periodically.
+	client *clientv3.Client
+
+	endpointCountState prometheus.Gauge
+}
+
+// initHealthChecker initializes the health checker for etcd client.
+func initHealthChecker(
+	tickerInterval time.Duration,
+	tlsConfig *tls.Config,
+	client *clientv3.Client,
+	source string,
+) {
+	healthChecker := &healthChecker{
+		source:             source,
+		tickerInterval:     tickerInterval,
+		tlsConfig:          tlsConfig,
+		client:             client,
+		endpointCountState: etcdStateGauge.WithLabelValues(source, endpointLabel),
+	}
+	// A health checker has the same lifetime with the given etcd client.
+	ctx := client.Ctx()
+	// Sync etcd endpoints and check the last health time of each endpoint periodically.
+	go healthChecker.syncer(ctx)
+	// Inspect the health of each endpoint by reading the health key periodically.
+	go healthChecker.inspector(ctx)
+}
+
+func (checker *healthChecker) syncer(ctx context.Context) {
+	defer logutil.LogPanic()
+	checker.update()
+	ticker := time.NewTicker(checker.tickerInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("etcd client is closed, exit the endpoint syncer goroutine",
+				zap.String("source", checker.source))
+			return
+		case <-ticker.C:
+			checker.update()
+		}
+	}
+}
+
+func (checker *healthChecker) inspector(ctx context.Context) {
+	defer logutil.LogPanic()
+	ticker := time.NewTicker(checker.tickerInterval)
+	defer ticker.Stop()
+	lastAvailable := time.Now()
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("etcd client is closed, exit the health inspector goroutine",
+				zap.String("source", checker.source))
+			checker.close()
+			return
+		case <-ticker.C:
+			lastEps, pickedEps, changed := checker.patrol(ctx)
+			if len(pickedEps) == 0 {
+				// when no endpoint could be used, try to reset endpoints to update connect rather
+				// than delete them to avoid there is no any endpoint in client.
+				// Note: reset endpoints will trigger sub-connection closed, and then trigger reconnection.
+				// Otherwise, the sub-connection will be retrying in gRPC layer and use exponential backoff,
+				// and it cannot recover as soon as possible.
+				if time.Since(lastAvailable) > etcdServerDisconnectedTimeout {
+					log.Info("no available endpoint, try to reset endpoints",
+						zap.Strings("last-endpoints", lastEps),
+						zap.String("source", checker.source))
+					resetClientEndpoints(checker.client, lastEps...)
+				}
+				continue
+			}
+			if changed {
+				oldNum, newNum := len(lastEps), len(pickedEps)
+				checker.client.SetEndpoints(pickedEps...)
+				checker.endpointCountState.Set(float64(newNum))
+				log.Info("update endpoints",
+					zap.String("num-change", fmt.Sprintf("%d->%d", oldNum, newNum)),
+					zap.Strings("last-endpoints", lastEps),
+					zap.Strings("endpoints", checker.client.Endpoints()),
+					zap.String("source", checker.source))
+			}
+			lastAvailable = time.Now()
+		}
+	}
+}
+
+func (checker *healthChecker) close() {
+	checker.healthyClients.Range(func(key, value any) bool {
+		healthyCli := value.(*healthyClient)
+		healthyCli.healthState.Set(0)
+		healthyCli.Client.Close()
+		return true
+	})
+}
+
+// Reset the etcd client endpoints to trigger reconnect.
+func resetClientEndpoints(client *clientv3.Client, endpoints ...string) {
+	client.SetEndpoints()
+	client.SetEndpoints(endpoints...)
+}
+
+type healthProbe struct {
+	ep   string
+	took time.Duration
+}
+
+// See https://github.com/etcd-io/etcd/blob/85b640cee793e25f3837c47200089d14a8392dc7/etcdctl/ctlv3/command/ep_command.go#L105-L145
+func (checker *healthChecker) patrol(ctx context.Context) ([]string, []string, bool) {
+	var (
+		count   = checker.clientCount()
+		probeCh = make(chan healthProbe, count)
+		wg      sync.WaitGroup
+	)
+	checker.healthyClients.Range(func(key, value any) bool {
+		wg.Add(1)
+		go func(key, value any) {
+			defer wg.Done()
+			defer logutil.LogPanic()
+			var (
+				ep          = key.(string)
+				healthyCli  = value.(*healthyClient)
+				client      = healthyCli.Client
+				healthState = healthyCli.healthState
+				latency     = healthyCli.latency
+				start       = time.Now()
+			)
+			// Check the health of the endpoint.
+			healthy := IsHealthy(ctx, client)
+			took := time.Since(start)
+			latency.Observe(took.Seconds())
+			if !healthy {
+				healthState.Set(0)
+				log.Warn("etcd endpoint is unhealthy",
+					zap.String("endpoint", ep),
+					zap.Duration("took", took),
+					zap.String("source", checker.source))
+				return
+			}
+			healthState.Set(1)
+			// If the endpoint is healthy, update its last health time.
+			checker.storeClient(ep, client, start)
+			// Send the healthy probe result to the channel.
+			probeCh <- healthProbe{ep, took}
+		}(key, value)
+		return true
+	})
+	wg.Wait()
+	close(probeCh)
+	var (
+		lastEps   = checker.client.Endpoints()
+		pickedEps = checker.pickEps(probeCh)
+	)
+	if len(pickedEps) > 0 {
+		checker.updateEvictedEps(lastEps, pickedEps)
+		pickedEps = checker.filterEps(pickedEps)
+	}
+	return lastEps, pickedEps, !typeutil.AreStringSlicesEquivalent(lastEps, pickedEps)
+}
+
+// Divide the acceptable latency range into several parts, and pick the endpoints which
+// are in the first acceptable latency range. Currently, we only take the latency of the
+// last health check into consideration, and maybe in the future we could introduce more
+// factors to help improving the selection strategy.
+func (checker *healthChecker) pickEps(probeCh <-chan healthProbe) []string {
+	var (
+		count     = len(probeCh)
+		pickedEps = make([]string, 0, count)
+	)
+	if count == 0 {
+		return pickedEps
+	}
+	// Consume the `probeCh` to build a reusable slice.
+	probes := make([]healthProbe, 0, count)
+	for probe := range probeCh {
+		probes = append(probes, probe)
+	}
+	// Take the default value as an example, if we have 3 endpoints with latency like:
+	//   - A: 175ms
+	//   - B: 50ms
+	//   - C: 2.5s
+	// the distribution will be like:
+	//   - [0, 1s) -> {A, B}
+	//   - [1s, 2s)
+	//   - [2s, 3s) -> {C}
+	//   - ...
+	//  - [9s, 10s)
+	// Then the picked endpoints will be {A, B} and if C is in the last used endpoints, it will be evicted later.
+	factor := int(DefaultRequestTimeout / DefaultSlowRequestTime)
+	for i := 0; i < factor; i++ {
+		minLatency, maxLatency := DefaultSlowRequestTime*time.Duration(i), DefaultSlowRequestTime*time.Duration(i+1)
+		for _, probe := range probes {
+			if minLatency <= probe.took && probe.took < maxLatency {
+				log.Debug("pick healthy etcd endpoint within acceptable latency range",
+					zap.Duration("min-latency", minLatency),
+					zap.Duration("max-latency", maxLatency),
+					zap.Duration("took", probe.took),
+					zap.String("endpoint", probe.ep),
+					zap.String("source", checker.source))
+				pickedEps = append(pickedEps, probe.ep)
+			}
+		}
+		if len(pickedEps) > 0 {
+			break
+		}
+	}
+	return pickedEps
+}
+
+func (checker *healthChecker) updateEvictedEps(lastEps, pickedEps []string) {
+	// Create a set of picked endpoints for faster lookup
+	pickedSet := make(map[string]bool, len(pickedEps))
+	for _, ep := range pickedEps {
+		pickedSet[ep] = true
+	}
+	// Reset the count to 0 if it's in evictedEps but not in the pickedEps.
+	checker.evictedEps.Range(func(key, _ any) bool {
+		ep := key.(string)
+		if !pickedSet[ep] {
+			checker.evictedEps.Store(ep, 0)
+			log.Info("reset evicted etcd endpoint picked count",
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source))
+		}
+		return true
+	})
+	// Find all endpoints which are in the lastEps but not in the pickedEps,
+	// and add them to the evictedEps.
+	for _, ep := range lastEps {
+		if pickedSet[ep] {
+			continue
+		}
+		checker.evictedEps.Store(ep, 0)
+		log.Info("evicted etcd endpoint found",
+			zap.String("endpoint", ep),
+			zap.String("source", checker.source))
+	}
+	// Find all endpoints which are in both pickedEps and evictedEps to
+	// increase their picked count.
+	for _, ep := range pickedEps {
+		if count, ok := checker.evictedEps.Load(ep); ok {
+			// Increase the count the endpoint being picked continuously.
+			checker.evictedEps.Store(ep, count.(int)+1)
+			log.Info("evicted etcd endpoint picked again",
+				zap.Int("picked-count-threshold", pickedCountThreshold),
+				zap.Int("picked-count", count.(int)+1),
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source))
+		}
+	}
+}
+
+// Filter out the endpoints that are in evictedEps and have not been continuously picked
+// for `pickedCountThreshold` times still, this is to ensure the evicted endpoints truly
+// become available before adding them back to the client.
+func (checker *healthChecker) filterEps(eps []string) []string {
+	pickedEps := make([]string, 0, len(eps))
+	for _, ep := range eps {
+		if count, ok := checker.evictedEps.Load(ep); ok {
+			if count.(int) < pickedCountThreshold {
+				continue
+			}
+			checker.evictedEps.Delete(ep)
+			log.Info("add evicted etcd endpoint back",
+				zap.Int("picked-count-threshold", pickedCountThreshold),
+				zap.Int("picked-count", count.(int)),
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source))
+		}
+		pickedEps = append(pickedEps, ep)
+	}
+	// If the pickedEps is empty, it means all endpoints are evicted,
+	// to gain better availability, just use the original picked endpoints.
+	if len(pickedEps) == 0 {
+		log.Warn("all etcd endpoints are evicted, use the picked endpoints directly",
+			zap.Strings("endpoints", eps),
+			zap.String("source", checker.source))
+		return eps
+	}
+	return pickedEps
+}
+
+func (checker *healthChecker) update() {
+	eps := checker.syncURLs()
+	if len(eps) == 0 {
+		log.Warn("no available etcd endpoint returned by etcd cluster",
+			zap.String("source", checker.source))
+		return
+	}
+	epMap := make(map[string]struct{}, len(eps))
+	for _, ep := range eps {
+		epMap[ep] = struct{}{}
+	}
+	// Check if client exists:
+	//   - If not, create one.
+	//   - If exists, check if it's offline or disconnected for a long time.
+	for ep := range epMap {
+		client := checker.loadClient(ep)
+		if client == nil {
+			checker.initClient(ep)
+			continue
+		}
+		since := time.Since(client.lastHealth)
+		// Check if it's offline for a long time and try to remove it.
+		if since > etcdServerOfflineTimeout {
+			log.Info("etcd server might be offline, try to remove it",
+				zap.Duration("since-last-health", since),
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source))
+			checker.removeClient(ep)
+			continue
+		}
+		// Check if it's disconnected for a long time and try to reconnect.
+		if since > etcdServerDisconnectedTimeout {
+			log.Info("etcd server might be disconnected, try to reconnect",
+				zap.Duration("since-last-health", since),
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source))
+			resetClientEndpoints(client.Client, ep)
+		}
+	}
+	// Clean up the stale clients which are not in the etcd cluster anymore.
+	checker.healthyClients.Range(func(key, value any) bool {
+		ep := key.(string)
+		if _, ok := epMap[ep]; !ok {
+			log.Info("remove stale etcd client",
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source))
+			checker.removeClient(ep)
+		}
+		return true
+	})
+}
+
+func (checker *healthChecker) clientCount() int {
+	count := 0
+	checker.healthyClients.Range(func(_, _ any) bool {
+		count++
+		return true
+	})
+	return count
+}
+
+func (checker *healthChecker) loadClient(ep string) *healthyClient {
+	if client, ok := checker.healthyClients.Load(ep); ok {
+		return client.(*healthyClient)
+	}
+	return nil
+}
+
+func (checker *healthChecker) initClient(ep string) {
+	client, err := newClient(checker.tlsConfig, ep)
+	if err != nil {
+		log.Error("failed to create etcd healthy client",
+			zap.String("endpoint", ep),
+			zap.String("source", checker.source),
+			zap.Error(err))
+		return
+	}
+	checker.storeClient(ep, client, time.Now())
+}
+
+func (checker *healthChecker) storeClient(ep string, client *clientv3.Client, lastHealth time.Time) {
+	checker.healthyClients.Store(ep, &healthyClient{
+		Client:      client,
+		lastHealth:  lastHealth,
+		healthState: etcdStateGauge.WithLabelValues(checker.source, ep),
+		latency:     etcdEndpointLatency.WithLabelValues(checker.source, ep),
+	})
+}
+
+func (checker *healthChecker) removeClient(ep string) {
+	if client, ok := checker.healthyClients.LoadAndDelete(ep); ok {
+		healthyCli := client.(*healthyClient)
+		healthyCli.healthState.Set(0)
+		if err := healthyCli.Close(); err != nil {
+			log.Error("failed to close etcd healthy client",
+				zap.String("endpoint", ep),
+				zap.String("source", checker.source),
+				zap.Error(err))
+		}
+	}
+	checker.evictedEps.Delete(ep)
+}
+
+// See https://github.com/etcd-io/etcd/blob/85b640cee793e25f3837c47200089d14a8392dc7/clientv3/client.go#L170-L183
+func (checker *healthChecker) syncURLs() (eps []string) {
+	resp, err := ListEtcdMembers(clientv3.WithRequireLeader(checker.client.Ctx()), checker.client)
+	if err != nil {
+		log.Error("failed to list members",
+			zap.String("source", checker.source),
+			errs.ZapError(err))
+		return nil
+	}
+	for _, m := range resp.Members {
+		if len(m.Name) == 0 || m.IsLearner {
+			continue
+		}
+		eps = append(eps, m.ClientURLs...)
+	}
+	return eps
+}
diff --git a/pkg/utils/etcdutil/metrics.go b/pkg/utils/etcdutil/metrics.go
index f78e0864b..5d0eae160 100644
--- a/pkg/utils/etcdutil/metrics.go
+++ b/pkg/utils/etcdutil/metrics.go
@@ -16,14 +16,32 @@ package etcdutil
 
 import "github.com/prometheus/client_golang/prometheus"
 
-var etcdStateGauge = prometheus.NewGaugeVec(
-	prometheus.GaugeOpts{
-		Namespace: "pd",
-		Subsystem: "server",
-		Name:      "etcd_client",
-		Help:      "Etcd client states.",
-	}, []string{"type"})
+var (
+	sourceLabel   = "source"
+	typeLabel     = "type"
+	endpointLabel = "endpoint"
+)
+
+var (
+	etcdStateGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "pd",
+			Subsystem: "server",
+			Name:      "etcd_client",
+			Help:      "Etcd client states.",
+		}, []string{sourceLabel, typeLabel})
+
+	etcdEndpointLatency = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "pd",
+			Subsystem: "server",
+			Name:      "etcd_endpoint_latency_seconds",
+			Help:      "Bucketed histogram of latency of health check.",
+			Buckets:   []float64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10},
+		}, []string{sourceLabel, endpointLabel})
+)
 
 func init() {
 	prometheus.MustRegister(etcdStateGauge)
+	prometheus.MustRegister(etcdEndpointLatency)
 }
diff --git a/server/api/member.go b/server/api/member.go
index 3016b7608..df8c0aee6 100644
--- a/server/api/member.go
+++ b/server/api/member.go
@@ -128,7 +128,7 @@ func (h *memberHandler) DeleteMemberByName(w http.ResponseWriter, r *http.Reques
 	// Get etcd ID by name.
 	var id uint64
 	name := mux.Vars(r)["name"]
-	listResp, err := etcdutil.ListEtcdMembers(client)
+	listResp, err := etcdutil.ListEtcdMembers(client.Ctx(), client)
 	if err != nil {
 		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
 		return
diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go
index ad863c992..1873c5667 100644
--- a/server/cluster/cluster.go
+++ b/server/cluster/cluster.go
@@ -2735,7 +2735,7 @@ func CheckHealth(client *http.Client, members []*pdpb.Member) map[uint64]*pdpb.M
 
 // GetMembers return a slice of Members.
 func GetMembers(etcdClient *clientv3.Client) ([]*pdpb.Member, error) {
-	listResp, err := etcdutil.ListEtcdMembers(etcdClient)
+	listResp, err := etcdutil.ListEtcdMembers(etcdClient.Ctx(), etcdClient)
 	if err != nil {
 		return nil, err
 	}
diff --git a/server/join/join.go b/server/join/join.go
index f22b61edf..d17110633 100644
--- a/server/join/join.go
+++ b/server/join/join.go
@@ -128,7 +128,7 @@ func PrepareJoinCluster(cfg *config.Config) error {
 	}
 	defer client.Close()
 
-	listResp, err := etcdutil.ListEtcdMembers(client)
+	listResp, err := etcdutil.ListEtcdMembers(client.Ctx(), client)
 	if err != nil {
 		return err
 	}
@@ -171,7 +171,7 @@ func PrepareJoinCluster(cfg *config.Config) error {
 	)
 
 	for i := 0; i < listMemberRetryTimes; i++ {
-		listResp, err = etcdutil.ListEtcdMembers(client)
+		listResp, err = etcdutil.ListEtcdMembers(client.Ctx(), client)
 		if err != nil {
 			return err
 		}
diff --git a/server/server.go b/server/server.go
index 4e30a65ac..fe1d414b4 100644
--- a/server/server.go
+++ b/server/server.go
@@ -347,36 +347,16 @@ func (s *Server) startEtcd(ctx context.Context) error {
 		return errs.ErrCancelStartEtcd.FastGenByArgs()
 	}
 
-	// start client
-	s.client, s.httpClient, err = s.startClient()
+	// Start the etcd and HTTP clients, then init the member.
+	err = s.startClient()
 	if err != nil {
 		return err
 	}
-
-	s.electionClient, err = s.startElectionClient()
+	err = s.initMember(newCtx, etcd)
 	if err != nil {
 		return err
 	}
 
-	// update advertise peer urls.
-	etcdMembers, err := etcdutil.ListEtcdMembers(s.client)
-	if err != nil {
-		return err
-	}
-	etcdServerID := uint64(etcd.Server.ID())
-	for _, m := range etcdMembers.Members {
-		if etcdServerID == m.ID {
-			etcdPeerURLs := strings.Join(m.PeerURLs, ",")
-			if s.cfg.AdvertisePeerUrls != etcdPeerURLs {
-				log.Info("update advertise peer urls", zap.String("from", s.cfg.AdvertisePeerUrls), zap.String("to", etcdPeerURLs))
-				s.cfg.AdvertisePeerUrls = etcdPeerURLs
-			}
-		}
-	}
-	failpoint.Inject("memberNil", func() {
-		time.Sleep(1500 * time.Millisecond)
-	})
-	s.member = member.NewMember(etcd, s.electionClient, etcdServerID)
 	s.initGRPCServiceLabels()
 	return nil
 }
@@ -389,29 +369,51 @@ func (s *Server) initGRPCServiceLabels() {
 	}
 }
 
-func (s *Server) startClient() (*clientv3.Client, *http.Client, error) {
+func (s *Server) startClient() error {
 	tlsConfig, err := s.cfg.Security.ToTLSConfig()
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
 	etcdCfg, err := s.cfg.GenEmbedEtcdConfig()
 	if err != nil {
-		return nil, nil, err
+		return err
 	}
-	return etcdutil.CreateClients(tlsConfig, etcdCfg.ACUrls)
-}
-
-func (s *Server) startElectionClient() (*clientv3.Client, error) {
-	tlsConfig, err := s.cfg.Security.ToTLSConfig()
+	/* Starting two different etcd clients here is to avoid the throttling. */
+	// This etcd client will be used to access the etcd cluster to read and write all kinds of meta data.
+	s.client, err = etcdutil.CreateEtcdClient(tlsConfig, etcdCfg.ACUrls, "server-etcd-client")
 	if err != nil {
-		return nil, err
+		return errs.ErrNewEtcdClient.Wrap(err).GenWithStackByCause()
 	}
-	etcdCfg, err := s.cfg.GenEmbedEtcdConfig()
+	// This etcd client will only be used to read and write the election-related data, such as leader key.
+	s.electionClient, err = etcdutil.CreateEtcdClient(tlsConfig, etcdCfg.ACUrls, "election-etcd-client")
 	if err != nil {
-		return nil, err
+		return errs.ErrNewEtcdClient.Wrap(err).GenWithStackByCause()
 	}
+	s.httpClient = etcdutil.CreateHTTPClient(tlsConfig)
+	return nil
+}
 
-	return etcdutil.CreateEtcdClient(tlsConfig, etcdCfg.ACUrls)
+func (s *Server) initMember(ctx context.Context, etcd *embed.Etcd) error {
+	// Update advertise peer URLs.
+	etcdMembers, err := etcdutil.ListEtcdMembers(ctx, s.client)
+	if err != nil {
+		return err
+	}
+	etcdServerID := uint64(etcd.Server.ID())
+	for _, m := range etcdMembers.Members {
+		if etcdServerID == m.ID {
+			etcdPeerURLs := strings.Join(m.PeerURLs, ",")
+			if s.cfg.AdvertisePeerUrls != etcdPeerURLs {
+				log.Info("update advertise peer urls", zap.String("from", s.cfg.AdvertisePeerUrls), zap.String("to", etcdPeerURLs))
+				s.cfg.AdvertisePeerUrls = etcdPeerURLs
+			}
+		}
+	}
+	failpoint.Inject("memberNil", func() {
+		time.Sleep(1500 * time.Millisecond)
+	})
+	s.member = member.NewMember(etcd, s.electionClient, etcdServerID)
+	return nil
 }
 
 // AddStartCallback adds a callback in the startServer phase.


Test patch:
diff --git a/pkg/utils/etcdutil/etcdutil_test.go b/pkg/utils/etcdutil/etcdutil_test.go
index f7fadd3bb..274037a93 100644
--- a/pkg/utils/etcdutil/etcdutil_test.go
+++ b/pkg/utils/etcdutil/etcdutil_test.go
@@ -53,7 +53,7 @@ func TestMemberHelpers(t *testing.T) {
 	etcd1, cfg1 := servers[0], servers[0].Config()
 
 	// Test ListEtcdMembers
-	listResp1, err := ListEtcdMembers(client1)
+	listResp1, err := ListEtcdMembers(client1.Ctx(), client1)
 	re.NoError(err)
 	re.Len(listResp1.Members, 1)
 	// types.ID is an alias of uint64.
@@ -74,7 +74,7 @@ func TestMemberHelpers(t *testing.T) {
 	_, err = RemoveEtcdMember(client1, uint64(etcd2.Server.ID()))
 	re.NoError(err)
 
-	listResp3, err := ListEtcdMembers(client1)
+	listResp3, err := ListEtcdMembers(client1.Ctx(), client1)
 	re.NoError(err)
 	re.Len(listResp3.Members, 1)
 	re.Equal(uint64(etcd1.Server.ID()), listResp3.Members[0].ID)
@@ -177,10 +177,8 @@ func TestEtcdClientSync(t *testing.T) {
 	etcd2 := MustAddEtcdMember(t, &cfg1, client1)
 	defer etcd2.Close()
 	checkMembers(re, client1, []*embed.Etcd{etcd1, etcd2})
-	testutil.Eventually(re, func() bool {
-		// wait for etcd client sync endpoints
-		return len(client1.Endpoints()) == 2
-	})
+	// wait for etcd client sync endpoints
+	checkEtcdEndpointNum(re, client1, 2)
 
 	// Remove the first member and close the etcd1.
 	_, err := RemoveEtcdMember(client1, uint64(etcd1.Server.ID()))
@@ -188,12 +186,21 @@ func TestEtcdClientSync(t *testing.T) {
 	etcd1.Close()
 
 	// Check the client can get the new member with the new endpoints.
+	checkEtcdEndpointNum(re, client1, 1)
+
+	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/utils/etcdutil/fastTick"))
+}
+
+func checkEtcdEndpointNum(re *require.Assertions, client *clientv3.Client, num int) {
 	testutil.Eventually(re, func() bool {
-		// wait for etcd client sync endpoints
-		return len(client1.Endpoints()) == 1
+		return len(client.Endpoints()) == num
 	})
+}
 
-	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/utils/etcdutil/fastTick"))
+func checkEtcdClientHealth(re *require.Assertions, client *clientv3.Client) {
+	testutil.Eventually(re, func() bool {
+		return IsHealthy(context.Background(), client)
+	})
 }
 
 func TestEtcdScaleInAndOut(t *testing.T) {
@@ -228,25 +235,21 @@ func TestRandomKillEtcd(t *testing.T) {
 	// Start a etcd server.
 	etcds, client1, clean := NewTestEtcdCluster(t, 3)
 	defer clean()
-	testutil.Eventually(re, func() bool {
-		return len(client1.Endpoints()) == 3
-	})
+	checkEtcdEndpointNum(re, client1, 3)
 
 	// Randomly kill an etcd server and restart it
 	cfgs := []embed.Config{etcds[0].Config(), etcds[1].Config(), etcds[2].Config()}
 	for i := 0; i < 10; i++ {
 		killIndex := rand.Intn(len(etcds))
 		etcds[killIndex].Close()
-		testutil.Eventually(re, func() bool {
-			return IsHealthy(context.Background(), client1)
-		})
+		checkEtcdEndpointNum(re, client1, 2)
+		checkEtcdClientHealth(re, client1)
 		etcd, err := embed.StartEtcd(&cfgs[killIndex])
 		re.NoError(err)
 		<-etcd.Server.ReadyNotify()
 		etcds[killIndex] = etcd
-		testutil.Eventually(re, func() bool {
-			return IsHealthy(context.Background(), client1)
-		})
+		checkEtcdEndpointNum(re, client1, 3)
+		checkEtcdClientHealth(re, client1)
 	}
 	for _, etcd := range etcds {
 		if etcd != nil {
diff --git a/pkg/utils/etcdutil/health_checker_test.go b/pkg/utils/etcdutil/health_checker_test.go
new file mode 100644
index 000000000..82dd1362b
--- /dev/null
+++ b/pkg/utils/etcdutil/health_checker_test.go
@@ -0,0 +1,363 @@
+// Copyright 2024 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 etcdutil
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+type testCase struct {
+	healthProbes       []healthProbe
+	expectedEvictedEps map[string]int
+	expectedPickedEps  []string
+}
+
+func check(re *require.Assertions, testCases []*testCase) {
+	checker := &healthChecker{}
+	lastEps := []string{}
+	for idx, tc := range testCases {
+		// Send the health probes to the channel.
+		probeCh := make(chan healthProbe, len(tc.healthProbes))
+		for _, probe := range tc.healthProbes {
+			probeCh <- probe
+		}
+		close(probeCh)
+		// Pick and filter the endpoints.
+		pickedEps := checker.pickEps(probeCh)
+		checker.updateEvictedEps(lastEps, pickedEps)
+		pickedEps = checker.filterEps(pickedEps)
+		// Check the evicted states after finishing picking.
+		count := 0
+		checker.evictedEps.Range(func(key, value any) bool {
+			count++
+			ep := key.(string)
+			times := value.(int)
+			re.Equal(tc.expectedEvictedEps[ep], times, "case %d ep %s", idx, ep)
+			return true
+		})
+		re.Len(tc.expectedEvictedEps, count, "case %d", idx)
+		re.Equal(tc.expectedPickedEps, pickedEps, "case %d", idx)
+		lastEps = pickedEps
+	}
+}
+
+// Test the endpoint picking and evicting logic.
+func TestPickEps(t *testing.T) {
+	re := require.New(t)
+	testCases := []*testCase{
+		// {} -> {A, B}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{},
+			[]string{"A", "B"},
+		},
+		// {A, B} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{},
+			[]string{"A", "B", "C"},
+		},
+		// {A, B, C} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{},
+			[]string{"A", "B", "C"},
+		},
+		// {A, B, C} -> {C}
+		{
+			[]healthProbe{
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 0, "B": 0},
+			[]string{"C"},
+		},
+		// {C} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 1, "B": 1},
+			[]string{"C"},
+		},
+		// {C} -> {B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 0, "B": 2},
+			[]string{"C"},
+		},
+		// {C} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 1},
+			[]string{"B", "C"},
+		},
+		// {B, C} -> {D}
+		{
+			[]healthProbe{
+				{
+					ep:   "D",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 0, "B": 0, "C": 0},
+			[]string{"D"},
+		},
+		// {D} -> {B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 0, "B": 1, "C": 1, "D": 0},
+			[]string{"B", "C"},
+		},
+		// {B, C} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 1, "B": 2, "C": 2, "D": 0},
+			[]string{"A", "B", "C"},
+		},
+		// {A, B, C} -> {A, C, E}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "E",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 2, "B": 0, "D": 0},
+			[]string{"C", "E"},
+		},
+	}
+	check(re, testCases)
+}
+
+func TestLatencyPick(t *testing.T) {
+	re := require.New(t)
+	testCases := []*testCase{
+		// {} -> {A, B}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{},
+			[]string{"A", "B"},
+		},
+		// {A, B} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Second,
+				},
+			},
+			map[string]int{},
+			[]string{"A", "B"},
+		},
+		// {A, B} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Second,
+				},
+				{
+					ep:   "B",
+					took: time.Second,
+				},
+				{
+					ep:   "C",
+					took: 2 * time.Second,
+				},
+			},
+			map[string]int{},
+			[]string{"A", "B"},
+		},
+		// {A, B} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Second,
+				},
+				{
+					ep:   "B",
+					took: 2 * time.Second,
+				},
+				{
+					ep:   "C",
+					took: 3 * time.Second,
+				},
+			},
+			map[string]int{"B": 0},
+			[]string{"A"},
+		},
+		// {A} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Second,
+				},
+				{
+					ep:   "B",
+					took: time.Second,
+				},
+				{
+					ep:   "C",
+					took: time.Millisecond,
+				},
+			},
+			map[string]int{"A": 0, "B": 0},
+			[]string{"C"},
+		},
+		// {C} -> {A, B, C}
+		{
+			[]healthProbe{
+				{
+					ep:   "A",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "B",
+					took: time.Millisecond,
+				},
+				{
+					ep:   "C",
+					took: time.Second,
+				},
+			},
+			map[string]int{"A": 1, "B": 1, "C": 0},
+			[]string{"A", "B"},
+		},
+	}
+	check(re, testCases)
+}
diff --git a/pkg/utils/etcdutil/testutil.go b/pkg/utils/etcdutil/testutil.go
index 54ba38b93..4e4713a48 100644
--- a/pkg/utils/etcdutil/testutil.go
+++ b/pkg/utils/etcdutil/testutil.go
@@ -70,7 +70,7 @@ func NewTestEtcdCluster(t *testing.T, count int) (servers []*embed.Etcd, etcdCli
 
 	for i := 1; i < count; i++ {
 		// Check the client can get the new member.
-		listResp, err := ListEtcdMembers(etcdClient)
+		listResp, err := ListEtcdMembers(etcdClient.Ctx(), etcdClient)
 		re.NoError(err)
 		re.Len(listResp.Members, i)
 		// Add a new member.
@@ -107,7 +107,7 @@ func MustAddEtcdMember(t *testing.T, cfg1 *embed.Config, client *clientv3.Client
 	re.NoError(err)
 	// Check the client can get the new member.
 	testutil.Eventually(re, func() bool {
-		members, err := ListEtcdMembers(client)
+		members, err := ListEtcdMembers(client.Ctx(), client)
 		re.NoError(err)
 		return len(addResp.Members) == len(members.Members)
 	})
@@ -121,7 +121,7 @@ func MustAddEtcdMember(t *testing.T, cfg1 *embed.Config, client *clientv3.Client
 
 func checkMembers(re *require.Assertions, client *clientv3.Client, etcds []*embed.Etcd) {
 	// Check the client can get the new member.
-	listResp, err := ListEtcdMembers(client)
+	listResp, err := ListEtcdMembers(client.Ctx(), client)
 	re.NoError(err)
 	re.Len(listResp.Members, len(etcds))
 	inList := func(m *etcdserverpb.Member) bool {
diff --git a/tests/pdctl/member/member_test.go b/tests/pdctl/member/member_test.go
index af3ee771e..9009e7550 100644
--- a/tests/pdctl/member/member_test.go
+++ b/tests/pdctl/member/member_test.go
@@ -87,14 +87,14 @@ func TestMember(t *testing.T) {
 	// member delete name <member_name>
 	err = svr.Destroy()
 	re.NoError(err)
-	members, err := etcdutil.ListEtcdMembers(client)
+	members, err := etcdutil.ListEtcdMembers(ctx, client)
 	re.NoError(err)
 	re.Len(members.Members, 3)
 	args = []string{"-u", pdAddr, "member", "delete", "name", name}
 	_, err = pdctl.ExecuteCommand(cmd, args...)
 	re.NoError(err)
 	testutil.Eventually(re, func() bool {
-		members, err = etcdutil.ListEtcdMembers(client)
+		members, err = etcdutil.ListEtcdMembers(ctx, client)
 		re.NoError(err)
 		return len(members.Members) == 2
 	})
@@ -104,7 +104,7 @@ func TestMember(t *testing.T) {
 	_, err = pdctl.ExecuteCommand(cmd, args...)
 	re.NoError(err)
 	testutil.Eventually(re, func() bool {
-		members, err = etcdutil.ListEtcdMembers(client)
+		members, err = etcdutil.ListEtcdMembers(ctx, client)
 		re.NoError(err)
 		return len(members.Members) == 2
 	})
diff --git a/tests/server/join/join_test.go b/tests/server/join/join_test.go
index 45eb1f162..5cdcbc090 100644
--- a/tests/server/join/join_test.go
+++ b/tests/server/join/join_test.go
@@ -47,7 +47,7 @@ func TestSimpleJoin(t *testing.T) {
 
 	pd1 := cluster.GetServer("pd1")
 	client := pd1.GetEtcdClient()
-	members, err := etcdutil.ListEtcdMembers(client)
+	members, err := etcdutil.ListEtcdMembers(ctx, client)
 	re.NoError(err)
 	re.Len(members.Members, 1)
 
@@ -58,7 +58,7 @@ func TestSimpleJoin(t *testing.T) {
 	re.NoError(err)
 	_, err = os.Stat(path.Join(pd2.GetConfig().DataDir, "join"))
 	re.False(os.IsNotExist(err))
-	members, err = etcdutil.ListEtcdMembers(client)
+	members, err = etcdutil.ListEtcdMembers(ctx, client)
 	re.NoError(err)
 	re.Len(members.Members, 2)
 	re.Equal(pd1.GetClusterID(), pd2.GetClusterID())
@@ -73,7 +73,7 @@ func TestSimpleJoin(t *testing.T) {
 	re.NoError(err)
 	_, err = os.Stat(path.Join(pd3.GetConfig().DataDir, "join"))
 	re.False(os.IsNotExist(err))
-	members, err = etcdutil.ListEtcdMembers(client)
+	members, err = etcdutil.ListEtcdMembers(ctx, client)
 	re.NoError(err)
 	re.Len(members.Members, 3)
 	re.Equal(pd1.GetClusterID(), pd3.GetClusterID())
@@ -108,7 +108,7 @@ func TestFailedAndDeletedPDJoinsPreviousCluster(t *testing.T) {
 	res := cluster.RunServer(pd3)
 	re.Error(<-res)
 
-	members, err := etcdutil.ListEtcdMembers(client)
+	members, err := etcdutil.ListEtcdMembers(ctx, client)
 	re.NoError(err)
 	re.Len(members.Members, 2)
 }
@@ -141,7 +141,7 @@ func TestDeletedPDJoinsPreviousCluster(t *testing.T) {
 	res := cluster.RunServer(pd3)
 	re.Error(<-res)
 
-	members, err := etcdutil.ListEtcdMembers(client)
+	members, err := etcdutil.ListEtcdMembers(ctx, client)
 	re.NoError(err)
 	re.Len(members.Members, 2)
 }