Increase the allowed concurrent gRPC streams (#16327)

* Increase the allowed concurrent gRPC streams

* Add a env override for the max streams setting

* Add changelog

* go fmt

* fix builds on 32bit systems

changelog/16327.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+core: Increase the allowed concurrent gRPC streams over the cluster port.
+```

vault/cluster/cluster.go

@@ -6,15 +6,18 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
-	"github.com/hashicorp/vault/sdk/helper/certutil"
-	"github.com/hashicorp/vault/sdk/helper/tlsutil"
+	"math"
 	"net"
 	"net/url"
 	"os"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/hashicorp/vault/sdk/helper/certutil"
+	"github.com/hashicorp/vault/sdk/helper/tlsutil"
+
 	log "github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/vault/sdk/helper/consts"
 	"golang.org/x/net/http2"
@@ -72,6 +75,17 @@ type Listener struct {
 }
 
 func NewListener(networkLayer NetworkLayer, cipherSuites []uint16, logger log.Logger, idleTimeout time.Duration) *Listener {
+	var maxStreams uint32 = math.MaxUint32
+	if override := os.Getenv("VAULT_GRPC_MAX_STREAMS"); override != "" {
+		i, err := strconv.ParseUint(override, 10, 32)
+		if err != nil {
+			logger.Warn("vault grpc max streams override must be an uint32 integer", "value", override)
+		} else {
+			maxStreams = uint32(i)
+			logger.Info("overriding grpc max streams", "value", i)
+		}
+	}
+
 	// Create the HTTP/2 server that will be shared by both RPC and regular
 	// duties. Doing it this way instead of listening via the server and gRPC
 	// allows us to re-use the same port via ALPN. We can just tell the server
@@ -80,6 +94,10 @@ func NewListener(networkLayer NetworkLayer, cipherSuites []uint16, logger log.Lo
 		// Our forwarding connections heartbeat regularly so anything else we
 		// want to go away/get cleaned up pretty rapidly
 		IdleTimeout: idleTimeout,
+
+		// By default this is 250 which can be too small on high traffic
+		// clusters with many forwarded or replication gRPC connections.
+		MaxConcurrentStreams: maxStreams,
 	}
 
 	return &Listener{