br: configure the httpclient for external storage

br/pkg/storage/azblob.go

@@ -157,10 +157,12 @@
  cred *azblob.SharedKeyCredential
  accountName string
  serviceURL string
+
+ clientOptions *azblob.ClientOptions
 }
 
 func (b *sharedKeyClientBuilder) GetServiceClient() (*azblob.Client, error) {
- return azblob.NewClientWithSharedKeyCredential(b.serviceURL, b.cred, getDefaultClientOptions())
+ return azblob.NewClientWithSharedKeyCredential(b.serviceURL, b.cred, b.clientOptions)
 }
 
 func (b *sharedKeyClientBuilder) GetAccountName() string {
@@ -172,10 +174,12 @@
  accountName string
  // Example of serviceURL: https://<account>.blob.core.windows.net/?<sas token>
  serviceURL string
+
+ clientOptions *azblob.ClientOptions
 }
 
 func (b *sasClientBuilder) GetServiceClient() (*azblob.Client, error) {
- return azblob.NewClientWithNoCredential(b.serviceURL, getDefaultClientOptions())
+ return azblob.NewClientWithNoCredential(b.serviceURL, b.clientOptions)
 }
 
 func (b *sasClientBuilder) GetAccountName() string {
@@ -187,10 +191,12 @@
  cred *azidentity.ClientSecretCredential
  accountName string
  serviceURL string
+
+ clientOptions *azblob.ClientOptions
 }
 
 func (b *tokenClientBuilder) GetServiceClient() (*azblob.Client, error) {
- return azblob.NewClient(b.serviceURL, b.cred, getDefaultClientOptions())
+ return azblob.NewClient(b.serviceURL, b.cred, b.clientOptions)
 }
 
 func (b *tokenClientBuilder) GetAccountName() string {
@@ -209,6 +215,11 @@
  return nil, errors.New("bucket(container) cannot be empty to access azure blob storage")
  }
 
+ clientOptions := getDefaultClientOptions()
+ if opts.HTTPClient != nil {
+ clientOptions.Transport = opts.HTTPClient
+ }
+
  if len(options.AccountName) > 0 && len(options.AccessSig) > 0 {
  serviceURL := options.Endpoint
  if len(serviceURL) == 0 {
@@ -221,6 +232,8 @@
  return &sasClientBuilder{
  options.AccountName,
  serviceURL,
+
+ clientOptions,
  }, nil
  }
 
@@ -237,6 +250,8 @@
  cred,
  options.AccountName,
  serviceURL,
+
+ clientOptions,
  }, nil
  }
 
@@ -265,6 +280,8 @@
  cred,
  accountName,
  serviceURL,
+
+ clientOptions,
  }, nil
  }
  log.Warn("Failed to get azure token credential but environment variables exist, try to use shared key.", zap.String("tenantId", tenantID), zap.String("clientId", clientID), zap.String("clientSecret", "?"))
@@ -292,6 +309,8 @@
  cred,
  accountName,
  serviceURL,
+
+ clientOptions,
  }, nil
 }
 

br/pkg/storage/storage.go

@@ -5,7 +5,9 @@
 import (
  "context"
  "io"
+ "net"
  "net/http"
+ "time"
 
  "github.com/aws/aws-sdk-go/aws/request"
  "github.com/pingcap/errors"
@@ -27,6 +29,8 @@
  GetObject Permission = "GetObject"
  // PutObject represents PutObject permission
  PutObject Permission = "PutObject"
+
+ DefaultRequestConcurrency uint = 128
 )
 
 // WalkOption is the option of storage.WalkDir.
@@ -169,6 +173,9 @@
  if opts == nil {
  opts = &ExternalStorageOptions{}
  }
+ if opts.HTTPClient == nil {
+ opts.HTTPClient = GetDefaultHttpClient(DefaultRequestConcurrency)
+ }
  switch backend := backend.Backend.(type) {
  case *backuppb.StorageBackend_Local:
  if backend.Local == nil {
@@ -198,3 +205,29 @@
  return nil, errors.Annotatef(berrors.ErrStorageInvalidConfig, "storage %T is not supported yet", backend)
  }
 }
+
+// copy from `http.defaultTransportDialContext`
+func defaultTransportDialContext(dialer *net.Dialer) func(context.Context, string, string) (net.Conn, error) {
+ return dialer.DialContext
+}
+
+// Different from `http.DefaultTransport`, set the `MaxIdleConns` and `MaxIdleConnsPerHost`
+// to the actual request concurrency to reuse tcp connection as much as possible.
+func GetDefaultHttpClient(concurrency uint) *http.Client {
+ return &http.Client{
+ // copy from `http.DefaultTransport`
+ Transport: &http.Transport{
+ Proxy: http.ProxyFromEnvironment,
+ DialContext: defaultTransportDialContext(&net.Dialer{
+ Timeout: 30 * time.Second,
+ KeepAlive: 30 * time.Second,
+ }),
+ ForceAttemptHTTP2: true,
+ MaxIdleConns: int(concurrency),
+ MaxIdleConnsPerHost: int(concurrency),
+ IdleConnTimeout: 90 * time.Second,
+ TLSHandshakeTimeout: 10 * time.Second,
+ ExpectContinueTimeout: 1 * time.Second,
+ },
+ }
+}

br/pkg/task/stream.go

@@ -135,6 +135,7 @@
  opts := storage.ExternalStorageOptions{
  NoCredentials: cfg.NoCreds,
  SendCredentials: cfg.SendCreds,
+ HTTPClient: storage.GetDefaultHttpClient(cfg.MetadataDownloadBatchSize),
  }
  storage, err := storage.New(ctx, u, &opts)
  if err != nil {
@@ -1471,6 +1472,7 @@
  opts := storage.ExternalStorageOptions{
  NoCredentials: cfg.NoCreds,
  SendCredentials: cfg.SendCreds,
+ HTTPClient: storage.GetDefaultHttpClient(cfg.MetadataDownloadBatchSize),
  }
  if err = client.SetStorage(ctx, u, &opts); err != nil {
  return nil, errors.Trace(err)