Support configuring MaxConcurrentStreams for http2

Backport https://github.com/etcd-io/etcd/pull/14219 to 3.4

Signed-off-by: Benjamin Wang <wachao@vmware.com>

embed/config.go

@@ -18,6 +18,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -55,6 +56,7 @@ const (
 	DefaultMaxTxnOps             = uint(128)
 	DefaultWarningApplyDuration  = 100 * time.Millisecond
 	DefaultMaxRequestBytes       = 1.5 * 1024 * 1024
+	DefaultMaxConcurrentStreams  = math.MaxUint32
 	DefaultGRPCKeepAliveMinTime  = 5 * time.Second
 	DefaultGRPCKeepAliveInterval = 2 * time.Hour
 	DefaultGRPCKeepAliveTimeout  = 20 * time.Second
@@ -177,6 +179,10 @@ type Config struct {
 	MaxTxnOps         uint  `json:"max-txn-ops"`
 	MaxRequestBytes   uint  `json:"max-request-bytes"`
 
+	// MaxConcurrentStreams specifies the maximum number of concurrent
+	// streams that each client can open at a time.
+	MaxConcurrentStreams uint32 `json:"max-concurrent-streams"`
+
 	LPUrls, LCUrls []url.URL
 	APUrls, ACUrls []url.URL
 	ClientTLSInfo  transport.TLSInfo
@@ -274,7 +280,7 @@ type Config struct {
 	AuthToken  string `json:"auth-token"`
 	BcryptCost uint   `json:"bcrypt-cost"`
 
-	//The AuthTokenTTL in seconds of the simple token
+	// AuthTokenTTL specifies the TTL in seconds of the simple token
 	AuthTokenTTL uint `json:"auth-token-ttl"`
 
 	ExperimentalInitialCorruptCheck bool          `json:"experimental-initial-corrupt-check"`
@@ -394,6 +400,7 @@ func NewConfig() *Config {
 
 		MaxTxnOps:                        DefaultMaxTxnOps,
 		MaxRequestBytes:                  DefaultMaxRequestBytes,
+		MaxConcurrentStreams:             DefaultMaxConcurrentStreams,
 		ExperimentalWarningApplyDuration: DefaultWarningApplyDuration,
 
 		GRPCKeepAliveMinTime:  DefaultGRPCKeepAliveMinTime,

embed/etcd.go

@@ -188,6 +188,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 		BackendBatchInterval:        cfg.BackendBatchInterval,
 		MaxTxnOps:                   cfg.MaxTxnOps,
 		MaxRequestBytes:             cfg.MaxRequestBytes,
+		MaxConcurrentStreams:        cfg.MaxConcurrentStreams,
 		StrictReconfigCheck:         cfg.StrictReconfigCheck,
 		ClientCertAuthEnabled:       cfg.ClientTLSInfo.ClientCertAuth,
 		AuthToken:                   cfg.AuthToken,
@@ -331,7 +332,10 @@ func print(lg *zap.Logger, ec Config, sc etcdserver.ServerConfig, memberInitiali
 			zap.String("initial-cluster", sc.InitialPeerURLsMap.String()),
 			zap.String("initial-cluster-state", ec.ClusterState),
 			zap.String("initial-cluster-token", sc.InitialClusterToken),
-			zap.Int64("quota-size-bytes", quota),
+			zap.Int64("quota-backend-bytes", quota),
+			zap.Uint("max-request-bytes", sc.MaxRequestBytes),
+			zap.Uint32("max-concurrent-streams", sc.MaxConcurrentStreams),
+
 			zap.Bool("pre-vote", sc.PreVote),
 			zap.Bool("initial-corrupt-check", sc.InitialCorruptCheck),
 			zap.String("corrupt-check-time-interval", sc.CorruptCheckTime.String()),

embed/serve.go

@@ -17,6 +17,7 @@ package embed
 import (
 	"context"
 	"fmt"
+	"golang.org/x/net/http2"
 	"io/ioutil"
 	defaultLog "log"
 	"math"
@@ -132,6 +133,10 @@ func (sctx *serveCtx) serve(
 			Handler:  createAccessController(sctx.lg, s, httpmux),
 			ErrorLog: logger, // do not log user error
 		}
+		if err = configureHttpServer(srvhttp, s.Cfg); err != nil {
+			sctx.lg.Error("Configure http server failed", zap.Error(err))
+			return err
+		}
 		httpl := m.Match(cmux.HTTP1())
 		go func() { errHandler(srvhttp.Serve(httpl)) }()
 
@@ -185,6 +190,10 @@ func (sctx *serveCtx) serve(
 			TLSConfig: tlscfg,
 			ErrorLog:  logger, // do not log user error
 		}
+		if err = configureHttpServer(srv, s.Cfg); err != nil {
+			sctx.lg.Error("Configure https server failed", zap.Error(err))
+			return err
+		}
 		go func() { errHandler(srv.Serve(tlsl)) }()
 
 		sctx.serversC <- &servers{secure: true, grpc: gs, http: srv}
@@ -202,6 +211,13 @@ func (sctx *serveCtx) serve(
 	return m.Serve()
 }
 
+func configureHttpServer(srv *http.Server, cfg etcdserver.ServerConfig) error {
+	// todo (ahrtr): should we support configuring other parameters in the future as well?
+	return http2.ConfigureServer(srv, &http2.Server{
+		MaxConcurrentStreams: cfg.MaxConcurrentStreams,
+	})
+}
+
 // grpcHandlerFunc returns an http.Handler that delegates to grpcServer on incoming gRPC
 // connections or otherHandler otherwise. Given in gRPC docs.
 func grpcHandlerFunc(grpcServer *grpc.Server, otherHandler http.Handler) http.Handler {