etcdutil: revert etcd client in pd legacy server (tikv#6182) (tikv#6186)

close tikv#6124, ref tikv#6182

Signed-off-by: lhy1024 <[email protected]>

Co-authored-by: lhy1024 <[email protected]>
Signed-off-by: Cabinfever_B <[email protected]>

.github/workflows/pd-tests.yaml

@@ -6,12 +6,14 @@ on:
       - release-4.0
       - release-5.*
       - release-6.*
+      - release-7.*
   pull_request:
     branches:
       - master
       - release-4.0
       - release-5.*
       - release-6.*
+      - release-7.*
 concurrency:
   group: ${{ github.ref }}-${{ github.workflow }}
   cancel-in-progress: true

.github/workflows/tso-function-test.yaml

@@ -5,11 +5,13 @@ on:
       - master
       - release-5.*
       - release-6.*
+      - release-7.*
   pull_request:
     branches:
       - master
       - release-5.*
       - release-6.*
+      - release-7.*
 concurrency:
   group: ${{ github.ref }}-${{ github.workflow }}
   cancel-in-progress: true

pkg/mcs/resource_manager/server/server.go

@@ -263,7 +263,7 @@ func (s *Server) initClient() error {
 	if err != nil {
 		return err
 	}
-	s.etcdClient, s.httpClient, err = etcdutil.CreateClients(tlsConfig, []url.URL(u))
+	s.etcdClient, s.httpClient, err = etcdutil.CreateClientsWithMultiEndpoint(tlsConfig, []url.URL(u))
 	return err
 }
 

pkg/mcs/tso/server/server.go

@@ -466,7 +466,7 @@ func (s *Server) initClient() error {
 	if err != nil {
 		return err
 	}
-	s.etcdClient, s.httpClient, err = etcdutil.CreateClients(tlsConfig, s.backendUrls)
+	s.etcdClient, s.httpClient, err = etcdutil.CreateClientsWithMultiEndpoint(tlsConfig, s.backendUrls)
 	return err
 }
 

pkg/utils/etcdutil/etcdutil.go

@@ -193,22 +193,30 @@ func EtcdKVPutWithTTL(ctx context.Context, c *clientv3.Client, key string, value
 	return kv.Put(ctx, key, value, clientv3.WithLease(grantResp.ID))
 }
 
+// CreateClientsWithMultiEndpoint creates etcd v3 client and http client.
+func CreateClientsWithMultiEndpoint(tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client, *http.Client, error) {
+	client, err := createEtcdClientWithMultiEndpoint(tlsConfig, acUrls)
+	if err != nil {
+		return nil, nil, errs.ErrNewEtcdClient.Wrap(err).GenWithStackByCause()
+	}
+	httpClient := createHTTPClient(tlsConfig)
+	return client, httpClient, nil
+}
+
 // CreateClients creates etcd v3 client and http client.
-func CreateClients(tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client, *http.Client, error) {
+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 := &http.Client{
-		Transport: &http.Transport{
-			DisableKeepAlives: true,
-			TLSClientConfig:   tlsConfig,
-		},
-	}
+	httpClient := createHTTPClient(tlsConfig)
 	return client, httpClient, nil
 }
 
-func createEtcdClient(tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client, error) {
+// createEtcdClientWithMultiEndpoint creates etcd v3 client.
+// Note: it will be used by micro service server and support multi etcd endpoints.
+// FIXME: But it cannot switch etcd endpoints as soon as possible when one of endpoints is with io hang.
+func createEtcdClientWithMultiEndpoint(tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client, error) {
 	if len(acUrls) == 0 {
 		return nil, errs.ErrNewEtcdClient.FastGenByArgs("no available etcd address")
 	}
@@ -244,6 +252,32 @@ func createEtcdClient(tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client
 	return client, err
 }
 
+// createEtcdClient creates etcd v3 client.
+// Note: it will be used by legacy pd-server, and only connect to leader only.
+func createEtcdClient(tlsConfig *tls.Config, acURL url.URL) (*clientv3.Client, error) {
+	lgc := zap.NewProductionConfig()
+	lgc.Encoding = log.ZapEncodingName
+	client, err := clientv3.New(clientv3.Config{
+		Endpoints:   []string{acURL.String()},
+		DialTimeout: defaultEtcdClientTimeout,
+		TLS:         tlsConfig,
+		LogConfig:   &lgc,
+	})
+	if err == nil {
+		log.Info("create etcd v3 client", zap.String("endpoints", acURL.String()))
+	}
+	return client, err
+}
+
+func createHTTPClient(tlsConfig *tls.Config) *http.Client {
+	return &http.Client{
+		Transport: &http.Transport{
+			DisableKeepAlives: true,
+			TLSClientConfig:   tlsConfig,
+		},
+	}
+}
+
 // InitClusterID creates a cluster ID for the given key if it hasn't existed.
 // This function assumes the cluster ID has already existed and always use a
 // cheaper read to retrieve it; if it doesn't exist, invoke the more expensive

pkg/utils/etcdutil/etcdutil_test.go

@@ -220,7 +220,7 @@ func TestEtcdClientSync(t *testing.T) {
 	ep1 := cfg1.LCUrls[0].String()
 	urls, err := types.NewURLs([]string{ep1})
 	re.NoError(err)
-	client1, err := createEtcdClient(nil, urls)
+	client1, err := createEtcdClientWithMultiEndpoint(nil, urls)
 	re.NoError(err)
 	<-etcd1.Server.ReadyNotify()
 
@@ -277,7 +277,7 @@ func checkEtcdWithHangLeader(t *testing.T) error {
 	// Create a etcd client with etcd1 as endpoint.
 	urls, err := types.NewURLs([]string{proxyAddr})
 	re.NoError(err)
-	client1, err := createEtcdClient(nil, urls)
+	client1, err := createEtcdClientWithMultiEndpoint(nil, urls)
 	re.NoError(err)
 
 	// Add a new member and set the client endpoints to etcd1 and etcd2.

server/server.go

@@ -349,7 +349,7 @@ func startClient(cfg *config.Config) (*clientv3.Client, *http.Client, error) {
 	if err != nil {
 		return nil, nil, err
 	}
-	return etcdutil.CreateClients(tlsConfig, etcdCfg.ACUrls)
+	return etcdutil.CreateClients(tlsConfig, etcdCfg.ACUrls[0])
 }
 
 // AddStartCallback adds a callback in the startServer phase.