client: support to set timeout in http client

client/http/client.go

@@ -42,8 +42,9 @@
 	httpsScheme          = "https"
 	networkErrorStatus   = "network error"
 
-	defaultMembersInfoUpdateInterval = time.Minute
-	defaultTimeout                   = 30 * time.Second
+	// The value of defaultTimeout is the maximum value.
+	// If you want to shorten timeout, declare it when creating the client
+	defaultTimeout = 300 * time.Second
 )
 
 // respHandleFunc is the function to handle the HTTP response.
@@ -77,7 +78,7 @@
 func (ci *clientInner) init(sd pd.ServiceDiscovery) {
 	// Init the HTTP client if it's not configured.
 	if ci.cli == nil {
-		ci.cli = &http.Client{Timeout: defaultTimeout}
+		ci.cli = &http.Client{}
 		if ci.tlsConf != nil {
 			transport := http.DefaultTransport.(*http.Transport).Clone()
 			transport.TLSClientConfig = ci.tlsConf
@@ -172,6 +173,7 @@
 		body        = reqInfo.body
 		res         = reqInfo.res
 		respHandler = reqInfo.respHandler
+		cancel      context.CancelFunc
 	)
 	logFields := []zap.Field{
 		zap.String("source", source),
@@ -181,6 +183,8 @@
 		zap.String("caller-id", callerID),
 	}
 	log.Debug("[pd] request the http url", logFields...)
+	ctx, cancel = context.WithTimeout(ctx, reqInfo.timeout)
+	defer cancel()
 	req, err := http.NewRequestWithContext(ctx, method, url, bytes.NewBuffer(body))
 	if err != nil {
 		log.Error("[pd] create http request failed", append(logFields, zap.Error(err))...)
@@ -244,6 +248,7 @@
 	callerID    string
 	respHandler respHandleFunc
 	bo          *retry.Backoffer
+	timeout     time.Duration
 }
 
 // ClientOption configures the HTTP client.
@@ -256,6 +261,13 @@
 	}
 }
 
+// WithTimeout configures the client with the given timeout config.
+func WithTimeout(dur time.Duration) ClientOption {
+	return func(c *client) {
+		c.timeout = dur
+	}
+}
+
 // WithTLSConfig configures the client with the given TLS config.
 // This option won't work if the client is configured with WithHTTPClient.
 func WithTLSConfig(tlsConf *tls.Config) ClientOption {
@@ -342,6 +354,13 @@
 	return &newClient
 }
 
+// WithTimeout sets and returns a new client with the given timeout config.
+func (c *client) WithTimeout(dur time.Duration) Client {
+	newClient := *c
+	newClient.timeout = dur
+	return &newClient
+}
+
 // Header key definition constants.
 const (
 	pdAllowFollowerHandleKey = "PD-Allow-Follower-Handle"
@@ -362,7 +381,8 @@
 	return c.inner.requestWithRetry(ctx, reqInfo.
 		WithCallerID(c.callerID).
 		WithRespHandler(c.respHandler).
-		WithBackoffer(c.bo),
+		WithBackoffer(c.bo).
+		WithTimeout(c.timeout),
 		headerOpts...)
 }
 

client/http/interface.go

@@ -21,6 +21,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/kvproto/pkg/metapb"
@@ -105,6 +106,8 @@ type Client interface {
 	WithRespHandler(func(resp *http.Response, res any) error) Client
 	// WithBackoffer sets and returns a new client with the given backoffer.
 	WithBackoffer(*retry.Backoffer) Client
+	// WithTimeout sets and returns a new client with the timeout config.
+	WithTimeout(time.Duration) Client
 	// Close gracefully closes the HTTP client.
 	Close()
 }

client/http/request_info.go

@@ -16,6 +16,7 @@ package http
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/tikv/pd/client/retry"
 )
@@ -88,11 +89,14 @@ type requestInfo struct {
 	res         any
 	respHandler respHandleFunc
 	bo          *retry.Backoffer
+	timeout     time.Duration
 }
 
 // newRequestInfo creates a new request info.
 func newRequestInfo() *requestInfo {
-	return &requestInfo{}
+	return &requestInfo{
+		timeout: defaultTimeout,
+	}
 }
 
 // WithCallerID sets the caller ID of the request.
@@ -143,6 +147,14 @@ func (ri *requestInfo) WithBackoffer(bo *retry.Backoffer) *requestInfo {
 	return ri
 }
 
+// WithTimeout sets the timeout of the request.
+func (ri *requestInfo) WithTimeout(dur time.Duration) *requestInfo {
+	if dur > 0 {
+		ri.timeout = dur
+	}
+	return ri
+}
+
 func (ri *requestInfo) getURL(addr string) string {
 	return fmt.Sprintf("%s%s", addr, ri.uri)
 }

server/api/version.go

@@ -16,7 +16,9 @@ package api
 
 import (
 	"net/http"
+	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/tikv/pd/pkg/versioninfo"
 	"github.com/unrolled/render"
 )
@@ -47,5 +49,9 @@ func (h *versionHandler) GetVersion(w http.ResponseWriter, r *http.Request) {
 	version := &version{
 		Version: versioninfo.PDReleaseVersion,
 	}
+	failpoint.Inject("slowGetVersion", func(val failpoint.Value) {
+		//nolint:durationcheck
+		time.Sleep(time.Millisecond * time.Duration(val.(int)))
+	})
 	h.rd.JSON(w, http.StatusOK, version)
 }

tests/integrations/client/http_client_test.go

@@ -16,6 +16,7 @@ package client_test
 
 import (
 	"context"
+	"fmt"
 	"math"
 	"net/http"
 	"net/url"
@@ -25,6 +26,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
@@ -118,7 +120,7 @@ func (suite *httpClientTestSuite) TearDownSuite() {
 }
 
 // RunTestInTwoModes is to run test in two modes.
-func (suite *httpClientTestSuite) RunTestInTwoModes(test func(mode mode, client pd.Client)) {
+func (suite *httpClientTestSuite) RunTestInTwoModes(test func(mode mode, client pd.Client), opts ...pd.ClientOption) {
 	// Run test with specific service discovery.
 	cli := setupCli(suite.Require(), suite.env[specificServiceDiscovery].ctx, suite.env[specificServiceDiscovery].endpoints)
 	sd := cli.GetServiceDiscovery()
@@ -680,6 +682,35 @@ func (suite *httpClientTestSuite) checkWithBackoffer(mode mode, client pd.Client
 	re.Nil(rule)
 }
 
+func (suite *httpClientTestSuite) TestWithTimeout() {
+	suite.RunTestInTwoModes(suite.checkWithTimeout, pd.WithTimeout(1*time.Second))
+}
+
+func (suite *httpClientTestSuite) checkWithTimeout(mode mode, client pd.Client) {
+	re := suite.Require()
+	env := suite.env[mode]
+
+	re.NoError(failpoint.Enable("github.com/tikv/pd/server/api/slowGetVersion", fmt.Sprintf("return(%d)", 700)))
+	ctx, cancel := context.WithTimeout(env.ctx, 500*time.Millisecond)
+	_, err := client.GetPDVersion(ctx)
+	re.ErrorIs(err, context.DeadlineExceeded)
+	cancel()
+
+	newClient := client.WithTimeout(500 * time.Millisecond)
+	_, err = newClient.GetPDVersion(env.ctx)
+	re.ErrorIs(err, context.DeadlineExceeded)
+
+	version, err := client.GetPDVersion(env.ctx)
+	re.NoError(err)
+	re.Equal("None", version)
+	re.NoError(failpoint.Disable("github.com/tikv/pd/server/api/slowGetVersion"))
+
+	re.NoError(failpoint.Enable("github.com/tikv/pd/server/api/slowGetVersion", fmt.Sprintf("return(%d)", 1200)))
+	_, err = client.GetPDVersion(ctx)
+	re.ErrorIs(err, context.DeadlineExceeded)
+	re.NoError(failpoint.Disable("github.com/tikv/pd/server/api/slowGetVersion"))
+}
+
 func (suite *httpClientTestSuite) TestRedirectWithMetrics() {
 	re := suite.Require()
 	env := suite.env[defaultServiceDiscovery]