diff --git a/README.md b/README.md index f2e8d6df..29bf743d 100644 --- a/README.md +++ b/README.md @@ -923,6 +923,7 @@ The behavior can be fixed by configuring the following env variables for the rat - `GRPC_MAX_CONNECTION_AGE`: a duration for the maximum amount of time a connection may exist before it will be closed by sending a GoAway. A random jitter of +/-10% will be added to MaxConnectionAge to spread out connection storms. - `GRPC_MAX_CONNECTION_AGE_GRACE`: an additive period after MaxConnectionAge after which the connection will be forcibly closed. +- `GRPC_MAX_CONCURRENT_STREAMS`: caps the maximum number of concurrent gRPC streams the server allows **per HTTP/2 connection**. Defaults to `0`, which leaves the grpc-go server default (unlimited, `math.MaxUint32`) in place. This is a per-connection limit, so the aggregate number of concurrent streams a pod can process is roughly `(number of active connections) x GRPC_MAX_CONCURRENT_STREAMS`. Setting this gives operators a lever to bound the number of handler goroutines spawned during a downstream stall (e.g. a slow Redis), preventing unbounded goroutine growth and OOMs. ## Health-check diff --git a/src/server/server_impl.go b/src/server/server_impl.go index a82f2821..51497812 100644 --- a/src/server/server_impl.go +++ b/src/server/server_impl.go @@ -231,6 +231,15 @@ func NewServer(s settings.Settings, name string, statsManager stats.Manager, loc return newServer(s, name, statsManager, localCache, opts...) } +// maxConcurrentStreamsOptions returns the ServerOption that caps concurrent gRPC +// streams per HTTP/2 connection, or no options when the cap is disabled (0). +func maxConcurrentStreamsOptions(maxStreams uint32) []grpc.ServerOption { + if maxStreams > 0 { + return []grpc.ServerOption{grpc.MaxConcurrentStreams(maxStreams)} + } + return nil +} + func newServer(s settings.Settings, name string, statsManager stats.Manager, localCache *freecache.Cache, opts ...settings.Option) *server { for _, opt := range opts { opt(&s) @@ -258,6 +267,7 @@ func newServer(s settings.Settings, name string, statsManager stats.Manager, loc ), grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()), } + grpcOptions = append(grpcOptions, maxConcurrentStreamsOptions(s.GrpcMaxConcurrentStreams)...) if s.GrpcServerUseTLS { grpcServerTlsConfig := s.GrpcServerTlsConfig ret.grpcCertProvider = provider.NewCertProvider(s, ret.store, s.GrpcServerTlsCert, s.GrpcServerTlsKey) diff --git a/src/server/server_impl_internal_test.go b/src/server/server_impl_internal_test.go new file mode 100644 index 00000000..10227a39 --- /dev/null +++ b/src/server/server_impl_internal_test.go @@ -0,0 +1,17 @@ +package server + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestMaxConcurrentStreamsOptions_Disabled(t *testing.T) { + opts := maxConcurrentStreamsOptions(0) + assert.Empty(t, opts) +} + +func TestMaxConcurrentStreamsOptions_Enabled(t *testing.T) { + opts := maxConcurrentStreamsOptions(100) + assert.Len(t, opts, 1) +} diff --git a/src/settings/settings.go b/src/settings/settings.go index 2129cf80..fd2ac04b 100644 --- a/src/settings/settings.go +++ b/src/settings/settings.go @@ -33,6 +33,9 @@ type Settings struct { GrpcMaxConnectionAge time.Duration `envconfig:"GRPC_MAX_CONNECTION_AGE" default:"24h" description:"Duration a connection may exist before it will be closed by sending a GoAway."` // GrpcMaxConnectionAgeGrace is an additive period after MaxConnectionAge after which the connection will be forcibly closed. GrpcMaxConnectionAgeGrace time.Duration `envconfig:"GRPC_MAX_CONNECTION_AGE_GRACE" default:"1h" description:"Period after MaxConnectionAge after which the connection will be forcibly closed."` + // GrpcMaxConcurrentStreams caps the maximum number of concurrent gRPC streams the server allows PER HTTP/2 connection. + // 0 disables the cap, leaving the grpc-go default (unlimited) in place. + GrpcMaxConcurrentStreams uint32 `envconfig:"GRPC_MAX_CONCURRENT_STREAMS" default:"0" description:"Maximum number of concurrent gRPC streams allowed per HTTP/2 connection. 0 disables the cap (grpc-go default, unlimited)."` // GrpcServerUseTLS enables gprc connections to server over TLS GrpcServerUseTLS bool `envconfig:"GRPC_SERVER_USE_TLS" default:"false"` // Allow to set the server certificate and key for TLS connections. diff --git a/src/settings/settings_test.go b/src/settings/settings_test.go index c6ab2fd0..888d1b70 100644 --- a/src/settings/settings_test.go +++ b/src/settings/settings_test.go @@ -128,6 +128,24 @@ func TestRedisClusterPipelineParallelism_Auto(t *testing.T) { assert.Equal(t, 0, settings.RedisClusterPipelineParallelism) } +// Tests for GrpcMaxConcurrentStreams +func TestGrpcMaxConcurrentStreams_Default(t *testing.T) { + os.Unsetenv("GRPC_MAX_CONCURRENT_STREAMS") + + settings := NewSettings() + + assert.Equal(t, uint32(0), settings.GrpcMaxConcurrentStreams) +} + +func TestGrpcMaxConcurrentStreams_Configured(t *testing.T) { + os.Setenv("GRPC_MAX_CONCURRENT_STREAMS", "100") + defer os.Unsetenv("GRPC_MAX_CONCURRENT_STREAMS") + + settings := NewSettings() + + assert.Equal(t, uint32(100), settings.GrpcMaxConcurrentStreams) +} + // Test both pools can be configured independently func TestRedisPoolOnEmptyBehavior_IndependentConfiguration(t *testing.T) { os.Setenv("REDIS_POOL_ON_EMPTY_BEHAVIOR", "ERROR")