client: add sheme in pd http client (#7761)

ref #7576

Signed-off-by: Cabinfever_B <cabinfeveroier@gmail.com>

client/client.go

@@ -381,7 +381,7 @@ func createClientWithKeyspace(
 		ctx:                     clientCtx,
 		cancel:                  clientCancel,
 		keyspaceID:              keyspaceID,
-		svrUrls:                 addrsToUrls(svrAddrs),
+		svrUrls:                 svrAddrs,
 		tlsCfg:                  tlsCfg,
 		option:                  newOption(),
 	}
@@ -500,7 +500,7 @@ func newClientWithKeyspaceName(
 		updateTokenConnectionCh: make(chan struct{}, 1),
 		ctx:                     clientCtx,
 		cancel:                  clientCancel,
-		svrUrls:                 addrsToUrls(svrAddrs),
+		svrUrls:                 svrAddrs,
 		tlsCfg:                  tlsCfg,
 		option:                  newOption(),
 	}
@@ -1388,19 +1388,6 @@ func (c *client) scatterRegionsWithOptions(ctx context.Context, regionsID []uint
 	return resp, nil
 }
 
-func addrsToUrls(addrs []string) []string {
-	// Add default schema "http://" to addrs.
-	urls := make([]string, 0, len(addrs))
-	for _, addr := range addrs {
-		if strings.Contains(addr, "://") {
-			urls = append(urls, addr)
-		} else {
-			urls = append(urls, "http://"+addr)
-		}
-	}
-	return urls
-}
-
 // IsLeaderChange will determine whether there is a leader change.
 func IsLeaderChange(err error) bool {
 	if err == errs.ErrClientTSOStreamClosed {

client/pd_service_discovery.go

@@ -480,18 +480,7 @@ func NewDefaultPDServiceDiscovery(
 	urls []string, tlsCfg *tls.Config,
 ) *pdServiceDiscovery {
 	var wg sync.WaitGroup
-	pdsd := &pdServiceDiscovery{
-		checkMembershipCh: make(chan struct{}, 1),
-		ctx:               ctx,
-		cancel:            cancel,
-		wg:                &wg,
-		apiCandidateNodes: [apiKindCount]*pdServiceBalancer{newPDServiceBalancer(emptyErrorFn), newPDServiceBalancer(regionAPIErrorFn)},
-		keyspaceID:        defaultKeyspaceID,
-		tlsCfg:            tlsCfg,
-		option:            newOption(),
-	}
-	pdsd.urls.Store(urls)
-	return pdsd
+	return newPDServiceDiscovery(ctx, cancel, &wg, nil, nil, defaultKeyspaceID, urls, tlsCfg, newOption())
 }
 
 // newPDServiceDiscovery returns a new PD service discovery-based client.
@@ -515,6 +504,7 @@ func newPDServiceDiscovery(
 		tlsCfg:              tlsCfg,
 		option:              option,
 	}
+	urls = addrsToUrls(urls)
 	pdsd.urls.Store(urls)
 	return pdsd
 }
@@ -1155,3 +1145,16 @@ func (c *pdServiceDiscovery) switchTSOAllocatorLeaders(allocatorMap map[string]*
 func (c *pdServiceDiscovery) GetOrCreateGRPCConn(addr string) (*grpc.ClientConn, error) {
 	return grpcutil.GetOrCreateGRPCConn(c.ctx, &c.clientConns, addr, c.tlsCfg, c.option.gRPCDialOptions...)
 }
+
+func addrsToUrls(addrs []string) []string {
+	// Add default schema "http://" to addrs.
+	urls := make([]string, 0, len(addrs))
+	for _, addr := range addrs {
+		if strings.Contains(addr, "://") {
+			urls = append(urls, addr)
+		} else {
+			urls = append(urls, "http://"+addr)
+		}
+	}
+	return urls
+}

tests/integrations/client/http_client_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"math"
 	"net/http"
+	"net/url"
 	"sort"
 	"strings"
 	"testing"
@@ -97,7 +98,10 @@ func (suite *httpClientTestSuite) SetupSuite() {
 			endpoints   = make([]string, 0, len(testServers))
 		)
 		for _, s := range testServers {
-			endpoints = append(endpoints, s.GetConfig().AdvertiseClientUrls)
+			addr := s.GetConfig().AdvertiseClientUrls
+			url, err := url.Parse(addr)
+			re.NoError(err)
+			endpoints = append(endpoints, url.Host)
 		}
 		env.endpoints = endpoints
 		env.cluster = cluster