refactor(server): refactor http connect pool when comminute with master.

Signed-off-by: Victor1319 <zengxuewei@oppo.com>

datanode/server.go

@@ -381,6 +381,10 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
 		log.LogErrorf("parseConfig: masters addrs format err[%v]", masters)
 		return err
 	}
+	poolSize := cfg.GetInt64(proto.CfgHttpPoolSize)
+	syslog.Printf("parseConfig: http pool size %d", poolSize)
+	MasterClient.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{PoolSize: int(poolSize)}))
+
 	if err = MasterClient.Start(); err != nil {
 		return err
 	}

master/cluster.go

@@ -386,6 +386,9 @@ func newCluster(name string, leaderInfo *LeaderInfo, fsm *MetadataFsm, partition
 	c.DecommissionLimit = defaultDecommissionParallelLimit
 	c.checkAutoCreateDataPartition = false
 	c.masterClient = masterSDK.NewMasterClient(nil, false)
+	c.masterClient.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{
+		PoolSize: int(cfg.httpPoolSize),
+	}))
 	c.inodeCountNotEqualMP = new(sync.Map)
 	c.maxInodeNotEqualMP = new(sync.Map)
 	c.dentryCountNotEqualMP = new(sync.Map)

master/config.go

@@ -55,6 +55,8 @@ const (
 
 	cfgVolForceDeletion           = "volForceDeletion"
 	cfgVolDeletionDentryThreshold = "volDeletionDentryThreshold"
+
+	cfgHttpReversePoolSize = "httpReversePoolSize"
 )
 
 // default value
@@ -98,7 +100,7 @@ const (
 	defaultMaxConcurrentLcNodes                        = 3
 	metaPartitionInodeUsageThreshold           float64 = 0.75 // inode usage threshold on a meta partition
 	lowerLimitRWMetaPartition                          = 3    // lower limit of RW meta partition, equal defaultReplicaNum
-	// defaultIntervalToCheckDelVerTaskExpiration         = 3
+	defaultHttpReversePoolSize                         = 1024
 )
 
 // AddrDatabase is a map that stores the address of a given host (e.g., the leader)
@@ -114,6 +116,8 @@ type clusterConfig struct {
 	IntervalToAlarmMissingDataPartition int64
 	PeriodToLoadALLDataPartitions       int64
 	metaNodeReservedMem                 uint64
+	httpProxyPoolSize                   uint64
+	httpPoolSize                        uint64
 	IntervalToCheckDataPartition        int // seconds
 	IntervalToCheckQos                  int // seconds
 	numberOfDataPartitionsToFree        int

master/http_server.go

@@ -880,10 +880,17 @@ func ErrResponse(w http.ResponseWriter, err error) {
 }
 
 func (m *Server) newReverseProxy() *httputil.ReverseProxy {
-	return &httputil.ReverseProxy{Director: func(request *http.Request) {
-		request.URL.Scheme = "http"
-		request.URL.Host = m.leaderInfo.addr
-	}}
+	tr := proto.GetHttpTransporter(&proto.HttpCfg{
+		PoolSize: int(m.config.httpProxyPoolSize),
+	})
+
+	return &httputil.ReverseProxy{
+		Director: func(request *http.Request) {
+			request.URL.Scheme = "http"
+			request.URL.Host = m.leaderInfo.addr
+		},
+		Transport: tr,
+	}
 }
 
 func (m *Server) proxy(w http.ResponseWriter, r *http.Request) {

master/server.go

@@ -136,11 +136,12 @@ func (m *Server) Start(cfg *config.Config) (err error) {
 	m.config = newClusterConfig()
 	gConfig = m.config
 	m.leaderInfo = &LeaderInfo{}
-	m.reverseProxy = m.newReverseProxy()
+
 	if err = m.checkConfig(cfg); err != nil {
 		log.LogError(errors.Stack(err))
 		return
 	}
+	m.reverseProxy = m.newReverseProxy()
 
 	if m.rocksDBStore, err = raftstore_db.NewRocksDBStoreAndRecovery(m.storeDir, LRUCacheSize, WriteBufferSize); err != nil {
 		return
@@ -269,6 +270,14 @@ func (m *Server) checkConfig(cfg *config.Config) (err error) {
 		m.config.nodeSetCapacity = defaultNodeSetCapacity
 	}
 
+	m.config.httpProxyPoolSize = uint64(cfg.GetInt64(cfgHttpReversePoolSize))
+	if m.config.httpProxyPoolSize < defaultHttpReversePoolSize {
+		m.config.httpProxyPoolSize = defaultHttpReversePoolSize
+	}
+	m.config.httpPoolSize = uint64(cfg.GetInt64(proto.CfgHttpPoolSize))
+
+	syslog.Printf("http reverse pool size %d, http pool size %d\n", m.config.httpProxyPoolSize, m.config.httpPoolSize)
+
 	m.config.DefaultNormalZoneCnt = defaultNodeSetGrpBatchCnt
 	m.config.DomainBuildAsPossible = cfg.GetBoolWithDefault(cfgDomainBuildAsPossible, false)
 	domainBatchGrpCnt := cfg.GetString(cfgDomainBatchGrpCnt)

metanode/metanode.go

@@ -329,6 +329,10 @@ func (m *MetaNode) parseConfig(cfg *config.Config) (err error) {
 		log.LogErrorf("parseConfig: masters addrs format err[%v]", masters)
 		return err
 	}
+	poolSize := cfg.GetInt64(proto.CfgHttpPoolSize)
+	syslog.Printf("parseConfig: http pool size %d", poolSize)
+	masterClient.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{PoolSize: int(poolSize)}))
+
 	if err = masterClient.Start(); err != nil {
 		return err
 	}

objectnode/server.go

@@ -292,6 +292,10 @@ func (o *ObjectNode) loadConfig(cfg *config.Config) (err error) {
 	o.disableCreateBucketByS3 = cfg.GetBool(disableCreateBucketByS3)
 
 	o.mc = master.NewMasterClient(masters, false)
+	poolSize := cfg.GetInt64(proto.CfgHttpPoolSize)
+	log.LogWarnf("loadConfig: http pool size %d", poolSize)
+	o.mc.SetTransport(proto.GetHttpTransporter(&proto.HttpCfg{PoolSize: int(poolSize)}))
+
 	o.vm = NewVolumeManager(masters, strict)
 	o.userStore = NewUserInfoStore(masters, strict)
 

proto/admin_proto.go

@@ -17,6 +17,8 @@ package proto
 import (
 	"encoding/json"
 	"fmt"
+	"net"
+	"net/http"
 	"strconv"
 	"time"
 
@@ -378,6 +380,35 @@ const (
 	MinDirChildrenNumLimit     = 1000000
 )
 
+const (
+	CfgHttpPoolSize     = "httpPoolSize"
+	defaultHttpPoolSize = 128
+)
+
+type HttpCfg struct {
+	PoolSize int
+}
+
+func GetHttpTransporter(cfg *HttpCfg) *http.Transport {
+	if cfg.PoolSize < defaultHttpPoolSize {
+		cfg.PoolSize = defaultHttpPoolSize
+	}
+
+	return &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		ForceAttemptHTTP2:     true,
+		MaxIdleConns:          cfg.PoolSize,
+		MaxIdleConnsPerHost:   cfg.PoolSize / 2,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+	}
+}
+
 // HTTPReply uniform response structure
 type HTTPReply struct {
 	Code int32       `json:"code"`
@@ -1030,7 +1061,7 @@ const (
 )
 
 const (
-	QosDefaultBurst                   = 16000000
+	QosDefaultBurst                   = 1600000
 	QosDefaultClientCnt        uint32 = 100
 	QosDefaultDiskMaxFLowLimit int    = 0x7FFFFFFF
 	QosDefaultDiskMaxIoLimit   int    = 100000

sdk/master/client.go

@@ -59,6 +59,7 @@ type MasterClient struct {
 	leaderAddr  string
 	timeout     time.Duration
 	clientIDKey string
+	client      *http.Client
 
 	adminAPI  *AdminAPI
 	clientAPI *ClientAPI
@@ -66,6 +67,12 @@ type MasterClient struct {
 	userAPI   *UserAPI
 }
 
+func (c *MasterClient) SetTransport(tr http.RoundTripper) {
+	c.client = &http.Client{
+		Transport: tr,
+	}
+}
+
 func (c *MasterClient) ReplaceMasterAddresses(addrs []string) {
 	c.Lock()
 	defer c.Unlock()
@@ -248,6 +255,13 @@ func (c *MasterClient) prepareRequest() (addr string, nodes []string) {
 
 func (c *MasterClient) httpRequest(method, url string, r *request) (resp *http.Response, err error) {
 	client := http.DefaultClient
+	if c.client != nil {
+		if log.EnableDebug() {
+			log.LogDebug("use client config from config")
+		}
+		client = c.client
+	}
+
 	if !r.noTimeout {
 		client.Timeout = c.timeout
 	}