diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 13e0f4f2d76..ac1063f09fa 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -118,6 +118,13 @@ var Keys = map[Key]string{ VisibilityArchivalQueryMaxQPS: "frontend.visibilityArchivalQueryMaxQPS", EnableServerVersionCheck: "frontend.enableServerVersionCheck", EnableTokenNamespaceEnforcement: "frontend.enableTokenNamespaceEnforcement", + KeepAliveMinTime: "frontend.keepAliveMinTime", + KeepAlivePermitWithoutStream: "frontend.keepAlivePermitWithoutStream", + KeepAliveMaxConnectionIdle: "frontend.keepAliveMaxConnectionIdle", + KeepAliveMaxConnectionAge: "frontend.keepAliveMaxConnectionAge", + KeepAliveMaxConnectionAgeGrace: "frontend.keepAliveMaxConnectionAgeGrace", + KeepAliveTime: "frontend.keepAliveTime", + KeepAliveTimeout: "frontend.keepAliveTimeout", // matching settings MatchingRPS: "matching.rps", @@ -437,6 +444,32 @@ const ( EnableServerVersionCheck // EnableTokenNamespaceEnforcement enables enforcement that namespace in completion token matches namespace of the request EnableTokenNamespaceEnforcement + // KeepAliveMinTime is the minimum amount of time a client should wait before sending a keepalive ping. + KeepAliveMinTime + // KeepAlivePermitWithoutStream If true, server allows keepalive pings even when there are no active + // streams(RPCs). If false, and client sends ping when there are no active + // streams, server will send GOAWAY and close the connection. + KeepAlivePermitWithoutStream + // KeepAliveMaxConnectionIdle is a duration for the amount of time after which an + // idle connection would be closed by sending a GoAway. Idleness duration is + // defined since the most recent time the number of outstanding RPCs became + // zero or the connection establishment. + KeepAliveMaxConnectionIdle + // KeepAliveMaxConnectionAge is 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. + KeepAliveMaxConnectionAge + // KeepAliveMaxConnectionAgeGrace is an additive period after MaxConnectionAge after + // which the connection will be forcibly closed. + KeepAliveMaxConnectionAgeGrace + // KeepAliveTime After a duration of this time if the server doesn't see any activity it + // pings the client to see if the transport is still alive. + // If set below 1s, a minimum value of 1s will be used instead. + KeepAliveTime + // KeepAliveTimeout After having pinged for keepalive check, the server waits for a duration + // of Timeout and if no activity is seen even after that the connection is closed. + KeepAliveTimeout // key for matching diff --git a/service/frontend/service.go b/service/frontend/service.go index eb44a7bf90c..9f4203f0558 100644 --- a/service/frontend/service.go +++ b/service/frontend/service.go @@ -32,6 +32,7 @@ import ( "go.temporal.io/api/workflowservice/v1" "google.golang.org/grpc" healthpb "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/keepalive" "google.golang.org/grpc/reflection" "go.temporal.io/server/common/config" @@ -112,6 +113,22 @@ type Config struct { // EnableTokenNamespaceEnforcement enables enforcement that namespace in completion token matches namespace of the request EnableTokenNamespaceEnforcement dynamicconfig.BoolPropertyFn + + // gRPC keep alive options + // If a client pings too frequently, terminate the connection. + KeepAliveMinTime dynamicconfig.DurationPropertyFn + // Allow pings even when there are no active streams (RPCs) + KeepAlivePermitWithoutStream dynamicconfig.BoolPropertyFn + // Close the connection if a client is idle. + KeepAliveMaxConnectionIdle dynamicconfig.DurationPropertyFn + // Close the connection if it is too old. + KeepAliveMaxConnectionAge dynamicconfig.DurationPropertyFn + // Additive period after MaxConnectionAge after which the connection will be forcibly closed. + KeepAliveMaxConnectionAgeGrace dynamicconfig.DurationPropertyFn + // Ping the client if it is idle to ensure the connection is still active. + KeepAliveTime dynamicconfig.DurationPropertyFn + // Wait for the ping ack before assuming the connection is dead. + KeepAliveTimeout dynamicconfig.DurationPropertyFn } // NewConfig returns new service config with default values @@ -152,6 +169,13 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int32, enableReadF DefaultWorkflowTaskTimeout: dc.GetDurationPropertyFilteredByNamespace(dynamicconfig.DefaultWorkflowTaskTimeout, common.DefaultWorkflowTaskTimeout), EnableServerVersionCheck: dc.GetBoolProperty(dynamicconfig.EnableServerVersionCheck, os.Getenv("TEMPORAL_VERSION_CHECK_DISABLED") == ""), EnableTokenNamespaceEnforcement: dc.GetBoolProperty(dynamicconfig.EnableTokenNamespaceEnforcement, false), + KeepAliveMinTime: dc.GetDurationProperty(dynamicconfig.KeepAliveMinTime, 10*time.Second), + KeepAlivePermitWithoutStream: dc.GetBoolProperty(dynamicconfig.KeepAlivePermitWithoutStream, true), + KeepAliveMaxConnectionIdle: dc.GetDurationProperty(dynamicconfig.KeepAliveMaxConnectionIdle, 2*time.Minute), + KeepAliveMaxConnectionAge: dc.GetDurationProperty(dynamicconfig.KeepAliveMaxConnectionAge, 5*time.Minute), + KeepAliveMaxConnectionAgeGrace: dc.GetDurationProperty(dynamicconfig.KeepAliveMaxConnectionAgeGrace, 70*time.Second), + KeepAliveTime: dc.GetDurationProperty(dynamicconfig.KeepAliveTime, 1*time.Minute), + KeepAliveTimeout: dc.GetDurationProperty(dynamicconfig.KeepAliveTimeout, 10*time.Second), } } @@ -268,11 +292,25 @@ func (s *Service) Start() { ) opts, err := s.params.RPCFactory.GetFrontendGRPCServerOptions() + kep := keepalive.EnforcementPolicy{ + MinTime: s.config.KeepAliveMinTime(), + PermitWithoutStream: s.config.KeepAlivePermitWithoutStream(), + } + var kp = keepalive.ServerParameters{ + MaxConnectionIdle: s.config.KeepAliveMaxConnectionIdle(), + MaxConnectionAge: s.config.KeepAliveMaxConnectionAge(), + MaxConnectionAgeGrace: s.config.KeepAliveMaxConnectionAgeGrace(), + Time: s.config.KeepAliveTime(), + Timeout: s.config.KeepAliveTimeout(), + } + if err != nil { logger.Fatal("creating grpc server options failed", tag.Error(err)) } opts = append( opts, + grpc.KeepaliveParams(kp), + grpc.KeepaliveEnforcementPolicy(kep), grpc.ChainUnaryInterceptor( rpc.ServiceErrorInterceptor, metricsInterceptor.Intercept,