remove unnecessary ctx

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

pkg/mcs/resourcemanager/server/server.go

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

pkg/mcs/tso/server/server.go

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

pkg/utils/etcdutil/etcdutil.go

@@ -224,7 +224,7 @@ func newClient(tlsConfig *tls.Config, acURL ...string) (*clientv3.Client, error)
 }
 
 // CreateEtcdClient creates etcd v3 client with detecting endpoints.
-func CreateEtcdClient(ctx context.Context, tlsConfig *tls.Config, acURLs []url.URL) (*clientv3.Client, error) {
+func CreateEtcdClient(tlsConfig *tls.Config, acURLs []url.URL) (*clientv3.Client, error) {
 	urls := make([]string, 0, len(acURLs))
 	for _, u := range acURLs {
 		urls = append(urls, u.String())
@@ -236,7 +236,7 @@ func CreateEtcdClient(ctx context.Context, tlsConfig *tls.Config, acURLs []url.U
 	checker := &healthyChecker{
 		tlsConfig: tlsConfig,
 	}
-	eps := syncUrls(ctx, client)
+	eps := syncUrls(client)
 	checker.update(eps)
 	tickerInterval := defaultDialKeepAliveTime
 	failpoint.Inject("fastTick", func() {
@@ -249,7 +249,7 @@ func CreateEtcdClient(ctx context.Context, tlsConfig *tls.Config, acURLs []url.U
 		return client, err
 	}
 
-	go func(ctx context.Context, client *clientv3.Client) {
+	go func(client *clientv3.Client) {
 		defer logutil.LogPanic()
 		ticker := time.NewTicker(tickerInterval)
 		defer ticker.Stop()
@@ -259,11 +259,9 @@ func CreateEtcdClient(ctx context.Context, tlsConfig *tls.Config, acURLs []url.U
 			select {
 			case <-client.Ctx().Done():
 				return
-			case <-ctx.Done():
-				return
 			case <-ticker.C:
 				usedEps := client.Endpoints()
-				healthyEps := checker.patrol(ctx)
+				healthyEps := checker.patrol(client.Ctx())
 				if len(healthyEps) == 0 {
 					// when all endpoints are unhealthy, try to reset endpoints rather than delete them
 					// to avoid blocking there is no any endpoint in client.
@@ -285,25 +283,23 @@ func CreateEtcdClient(ctx context.Context, tlsConfig *tls.Config, acURLs []url.U
 				}
 			}
 		}
-	}(ctx, client)
+	}(client)
 
 	// Notes: use another goroutine to update endpoints to avoid blocking health check in the first goroutine.
-	go func(ctx context.Context, client *clientv3.Client) {
+	go func(client *clientv3.Client) {
 		defer logutil.LogPanic()
 		ticker := time.NewTicker(tickerInterval)
 		defer ticker.Stop()
 		for {
 			select {
 			case <-client.Ctx().Done():
 				return
-			case <-ctx.Done():
-				return
 			case <-ticker.C:
-				eps := syncUrls(ctx, client)
+				eps := syncUrls(client)
 				checker.update(eps)
 			}
 		}
-	}(ctx, client)
+	}(client)
 
 	return client, err
 }
@@ -393,9 +389,9 @@ func (checker *healthyChecker) addClient(ep string, lastHealth time.Time) {
 	})
 }
 
-func syncUrls(ctx context.Context, client *clientv3.Client) []string {
+func syncUrls(client *clientv3.Client) []string {
 	// See https://github.com/etcd-io/etcd/blob/85b640cee793e25f3837c47200089d14a8392dc7/clientv3/client.go#L170
-	ctx, cancel := context.WithTimeout(clientv3.WithRequireLeader(ctx), DefaultRequestTimeout)
+	ctx, cancel := context.WithTimeout(clientv3.WithRequireLeader(client.Ctx()), DefaultRequestTimeout)
 	defer cancel()
 	now := time.Now()
 	mresp, err := client.MemberList(ctx)
@@ -430,8 +426,8 @@ func isEqual(a, b []string) bool {
 }
 
 // CreateClients creates etcd v3 client and http client.
-func CreateClients(ctx context.Context, tlsConfig *tls.Config, acUrls []url.URL) (*clientv3.Client, *http.Client, error) {
-	client, err := CreateEtcdClient(ctx, tlsConfig, acUrls)
+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()
 	}

pkg/utils/etcdutil/etcdutil_test.go

@@ -234,8 +234,6 @@ func TestInitClusterID(t *testing.T) {
 func TestEtcdClientSync(t *testing.T) {
 	re := require.New(t)
 	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/utils/etcdutil/fastTick", "return(true)"))
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := sync.WaitGroup{}
 
 	// Start a etcd server.
 	cfg1 := NewTestSingleConfig(t)
@@ -249,7 +247,7 @@ func TestEtcdClientSync(t *testing.T) {
 	ep1 := cfg1.LCUrls[0].String()
 	urls, err := types.NewURLs([]string{ep1})
 	re.NoError(err)
-	client1, err := CreateEtcdClient(ctx, nil, urls)
+	client1, err := CreateEtcdClient(nil, urls)
 	defer func() {
 		client1.Close()
 	}()
@@ -273,8 +271,6 @@ func TestEtcdClientSync(t *testing.T) {
 		return err == nil && len(listResp.Members) == 1 && listResp.Members[0].ID == uint64(etcd2.Server.ID())
 	})
 
-	cancel()
-	wg.Wait()
 	re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/utils/etcdutil/fastTick"))
 }
 
@@ -296,8 +292,6 @@ func TestEtcdWithHangLeaderEnableCheck(t *testing.T) {
 
 func TestEtcdScaleInAndOutWithoutMultiPoint(t *testing.T) {
 	re := require.New(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := sync.WaitGroup{}
 	// Start a etcd server.
 	cfg1 := NewTestSingleConfig(t)
 	etcd1, err := embed.StartEtcd(cfg1)
@@ -311,12 +305,12 @@ func TestEtcdScaleInAndOutWithoutMultiPoint(t *testing.T) {
 	// Create two etcd clients with etcd1 as endpoint.
 	urls, err := types.NewURLs([]string{ep1})
 	re.NoError(err)
-	client1, err := CreateEtcdClient(ctx, nil, urls) // execute member change operation with this client
+	client1, err := CreateEtcdClient(nil, urls) // execute member change operation with this client
 	defer func() {
 		client1.Close()
 	}()
 	re.NoError(err)
-	client2, err := CreateEtcdClient(ctx, nil, urls) // check member change with this client
+	client2, err := CreateEtcdClient(nil, urls) // check member change with this client
 	defer func() {
 		client2.Close()
 	}()
@@ -333,14 +327,10 @@ func TestEtcdScaleInAndOutWithoutMultiPoint(t *testing.T) {
 	_, err = RemoveEtcdMember(client1, uint64(etcd1.Server.ID()))
 	re.NoError(err)
 	checkMembers(re, client2, []*embed.Etcd{etcd2})
-	cancel()
-	wg.Wait()
 }
 
 func checkEtcdWithHangLeader(t *testing.T) error {
 	re := require.New(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := sync.WaitGroup{}
 	// Start a etcd server.
 	cfg1 := NewTestSingleConfig(t)
 	etcd1, err := embed.StartEtcd(cfg1)
@@ -359,7 +349,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(ctx, nil, urls)
+	client1, err := CreateEtcdClient(nil, urls)
 	defer func() {
 		client1.Close()
 	}()
@@ -375,8 +365,6 @@ func checkEtcdWithHangLeader(t *testing.T) error {
 	enableDiscard.Store(true)
 	time.Sleep(defaultDialKeepAliveTime + defaultDialKeepAliveTimeout*2)
 	_, err = EtcdKVGet(client1, "test/key1")
-	cancel()
-	wg.Wait()
 	return err
 }
 
@@ -494,7 +482,7 @@ func (suite *loopWatcherTestSuite) SetupSuite() {
 	ep1 := suite.config.LCUrls[0].String()
 	urls, err := types.NewURLs([]string{ep1})
 	suite.NoError(err)
-	suite.client, err = CreateEtcdClient(suite.ctx, nil, urls)
+	suite.client, err = CreateEtcdClient(nil, urls)
 	suite.NoError(err)
 	suite.cleans = append(suite.cleans, func() {
 		suite.client.Close()
@@ -697,23 +685,23 @@ func (suite *loopWatcherTestSuite) TestWatcherBreak() {
 
 	// Case2: close the etcd client and put a new value after watcher restarts
 	suite.client.Close()
-	suite.client, err = CreateEtcdClient(suite.ctx, nil, suite.config.LCUrls)
+	suite.client, err = CreateEtcdClient(nil, suite.config.LCUrls)
 	suite.NoError(err)
 	watcher.updateClientCh <- suite.client
 	suite.put("TestWatcherBreak", "2")
 	checkCache("2")
 
 	// Case3: close the etcd client and put a new value before watcher restarts
 	suite.client.Close()
-	suite.client, err = CreateEtcdClient(suite.ctx, nil, suite.config.LCUrls)
+	suite.client, err = CreateEtcdClient(nil, suite.config.LCUrls)
 	suite.NoError(err)
 	suite.put("TestWatcherBreak", "3")
 	watcher.updateClientCh <- suite.client
 	checkCache("3")
 
 	// Case4: close the etcd client and put a new value with compact
 	suite.client.Close()
-	suite.client, err = CreateEtcdClient(suite.ctx, nil, suite.config.LCUrls)
+	suite.client, err = CreateEtcdClient(nil, suite.config.LCUrls)
 	suite.NoError(err)
 	suite.put("TestWatcherBreak", "4")
 	resp, err := EtcdKVGet(suite.client, "TestWatcherBreak")

server/server.go

@@ -379,7 +379,7 @@ func (s *Server) startClient() (*clientv3.Client, *http.Client, error) {
 	if err != nil {
 		return nil, nil, err
 	}
-	return etcdutil.CreateClients(s.ctx, tlsConfig, etcdCfg.ACUrls)
+	return etcdutil.CreateClients(tlsConfig, etcdCfg.ACUrls)
 }
 
 func (s *Server) startElectionClient() (*clientv3.Client, error) {
@@ -392,7 +392,7 @@ func (s *Server) startElectionClient() (*clientv3.Client, error) {
 		return nil, err
 	}
 
-	return etcdutil.CreateEtcdClient(s.ctx, tlsConfig, etcdCfg.ACUrls)
+	return etcdutil.CreateEtcdClient(tlsConfig, etcdCfg.ACUrls)
 }
 
 // AddStartCallback adds a callback in the startServer phase.