Add basic connection management and keep alive parameters (#1334)

temporalio · Mar 30, 2021 · df0c869 · df0c869
1 parent c2d3f6f
commit df0c869
Show file tree

Hide file tree

Showing 2 changed files with 71 additions and 0 deletions.
diff --git 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
@@ -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,