Add grpc keep alive server and client config for server and client initialization (#7483)

## What changed?
<!-- Describe what has changed in this PR -->
1. Add grpc keepalive config in the yaml config.
## Why?
<!-- Tell your future self why have you made these changes -->
To support configuring grpc keep alive for service and client.
## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->

## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->

Author: xwduan

client/clientfactory.go

@@ -128,7 +128,6 @@ func (cf *rpcClientFactory) NewHistoryClientWithTimeout(timeout time.Duration) (
 	if err != nil {
 		return nil, err
 	}
-
 	client := history.NewClient(
 		cf.dynConfig,
 		resolver,
@@ -155,7 +154,7 @@ func (cf *rpcClientFactory) NewMatchingClientWithTimeout(
 
 	keyResolver := newServiceKeyResolver(resolver)
 	clientProvider := func(clientKey string) (interface{}, error) {
-		connection := cf.rpcFactory.CreateInternodeGRPCConnection(clientKey)
+		connection := cf.rpcFactory.CreateMatchingGRPCConnection(clientKey)
 		return matchingservice.NewMatchingServiceClient(connection), nil
 	}
 	client := matching.NewClient(

client/history/client_test.go

@@ -185,7 +185,7 @@ func (s *testHistoryService) DeleteDLQTasks(
 	return &historyservice.DeleteDLQTasksResponse{}, nil
 }
 
-func (t *testRPCFactory) CreateInternodeGRPCConnection(rpcAddress string) *grpc.ClientConn {
+func (t *testRPCFactory) CreateHistoryGRPCConnection(rpcAddress string) *grpc.ClientConn {
 	t.dialedAddresses = append(t.dialedAddresses, rpcAddress)
-	return t.base.CreateInternodeGRPCConnection(rpcAddress)
+	return t.base.CreateHistoryGRPCConnection(rpcAddress)
 }

client/history/connections.go

@@ -54,7 +54,7 @@ type (
 
 	// RPCFactory is a subset of the [go.temporal.io/server/common/rpc.RPCFactory] interface to make testing easier.
 	RPCFactory interface {
-		CreateInternodeGRPCConnection(rpcAddress string) *grpc.ClientConn
+		CreateHistoryGRPCConnection(rpcAddress string) *grpc.ClientConn
 	}
 
 	connectionPool interface {
@@ -90,8 +90,7 @@ func (c *connectionPoolImpl) getOrCreateClientConn(addr rpcAddress) clientConnec
 	if cc, ok = c.mu.conns[addr]; ok {
 		return cc
 	}
-
-	grpcConn := c.rpcFactory.CreateInternodeGRPCConnection(string(addr))
+	grpcConn := c.rpcFactory.CreateHistoryGRPCConnection(string(addr))
 	cc = clientConnection{
 		historyClient: historyservice.NewHistoryServiceClient(grpcConn),
 		grpcConn:      grpcConn,

client/history/connections_mock.go

@@ -27,6 +27,7 @@ package config
 import (
 	"bytes"
 	"fmt"
+	"math"
 	"strings"
 	"time"
 
@@ -40,9 +41,14 @@ import (
 	"go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/telemetry"
+	"google.golang.org/grpc/keepalive"
 	"gopkg.in/yaml.v3"
 )
 
+const (
+	infinity = time.Duration(math.MaxInt64)
+)
+
 type (
 	// Config contains the configuration for a set of temporal services
 	Config struct {
@@ -107,6 +113,39 @@ type (
 		// forwarded from HTTP to gRPC. Any value with a trailing * will match the prefix before
 		// the asterisk (eg. `x-internal-*`)
 		HTTPAdditionalForwardedHeaders []string `yaml:"httpAdditionalForwardedHeaders"`
+		// KeepAliveServerConfig keep alive configuration for the server
+		KeepAliveServerConfig KeepAliveServerConfig `yaml:"keepAliveServerConfig"`
+		// ClientConnectionConfig defines the connection config used by other services
+		// when they create a gRPC client connection to this service.
+		ClientConnectionConfig ClientConnectionConfig `yaml:"clientConnectionConfig"`
+	}
+
+	KeepAliveServerParameters struct {
+		MaxConnectionIdle     *time.Duration `yaml:"maxConnectionIdle"`
+		MaxConnectionAge      *time.Duration `yaml:"maxConnectionAge"`
+		MaxConnectionAgeGrace *time.Duration `yaml:"maxConnectionAgeGrace"`
+		Time                  *time.Duration `yaml:"keepAliveTime"`
+		Timeout               *time.Duration `yaml:"keepAliveTimeout"`
+	}
+
+	KeepAliveClientParameters struct {
+		Time                *time.Duration `yaml:"keepAliveTime"`
+		Timeout             *time.Duration `yaml:"keepAliveTimeout"`
+		PermitWithoutStream *bool          `yaml:"keepAlivePermitWithoutStream"`
+	}
+
+	ClientConnectionConfig struct {
+		KeepAliveClientConfig *KeepAliveClientParameters `yaml:"keepAliveClientParameters"`
+	}
+
+	KeepAliveServerEnforcementPolicy struct {
+		MinTime             *time.Duration `yaml:"minTime"`
+		PermitWithoutStream *bool          `yaml:"permitWithoutStream"`
+	}
+
+	KeepAliveServerConfig struct {
+		KeepAliveServerParameters  *KeepAliveServerParameters        `yaml:"keepAliveServerParameters"`
+		KeepAliveEnforcementPolicy *KeepAliveServerEnforcementPolicy `yaml:"keepAliveEnforcementPolicy"`
 	}
 
 	// Global contains config items that apply process-wide to all services
@@ -652,3 +691,78 @@ func (p *JWTKeyProvider) HasSourceURIsConfigured() bool {
 	}
 	return false
 }
+
+func (k *KeepAliveServerConfig) GetKeepAliveServerParameters() keepalive.ServerParameters {
+	// the default config is same as grpc default config, same for the below client config and enforcement policy
+	defaultConfig := keepalive.ServerParameters{
+		MaxConnectionIdle:     infinity,
+		MaxConnectionAge:      infinity,
+		MaxConnectionAgeGrace: infinity,
+		Time:                  2 * time.Hour,
+		Timeout:               20 * time.Second,
+	}
+	if k == nil || k.KeepAliveServerParameters == nil {
+		return defaultConfig
+	}
+	kp := k.KeepAliveServerParameters
+	if kp.MaxConnectionIdle != nil {
+		defaultConfig.MaxConnectionIdle = *kp.MaxConnectionIdle
+	}
+	if kp.MaxConnectionAge != nil {
+		defaultConfig.MaxConnectionAge = *kp.MaxConnectionAge
+	}
+	if kp.MaxConnectionAgeGrace != nil {
+		defaultConfig.MaxConnectionAgeGrace = *kp.MaxConnectionAgeGrace
+	}
+	if kp.Time != nil {
+		defaultConfig.Time = *kp.Time
+	}
+	if kp.Timeout != nil {
+		defaultConfig.Timeout = *kp.Timeout
+	}
+	return defaultConfig
+}
+
+func (c *ClientConnectionConfig) GetKeepAliveClientParameters() keepalive.ClientParameters {
+	defaultConfig := keepalive.ClientParameters{
+		Time:                infinity,
+		Timeout:             20 * time.Second,
+		PermitWithoutStream: false,
+	}
+
+	if c == nil || c.KeepAliveClientConfig == nil {
+		return defaultConfig
+	}
+
+	if c.KeepAliveClientConfig.Time != nil {
+		defaultConfig.Time = *c.KeepAliveClientConfig.Time
+	}
+	if c.KeepAliveClientConfig.Timeout != nil {
+		defaultConfig.Timeout = *c.KeepAliveClientConfig.Timeout
+	}
+	if c.KeepAliveClientConfig.PermitWithoutStream != nil {
+		defaultConfig.PermitWithoutStream = *c.KeepAliveClientConfig.PermitWithoutStream
+	}
+
+	return defaultConfig
+}
+
+func (k *KeepAliveServerConfig) GetKeepAliveEnforcementPolicy() keepalive.EnforcementPolicy {
+	defaultConfig := keepalive.EnforcementPolicy{
+		MinTime:             5 * time.Minute,
+		PermitWithoutStream: false,
+	}
+
+	if k == nil || k.KeepAliveEnforcementPolicy == nil {
+		return defaultConfig
+	}
+
+	if k.KeepAliveEnforcementPolicy.MinTime != nil {
+		defaultConfig.MinTime = *k.KeepAliveEnforcementPolicy.MinTime
+	}
+	if k.KeepAliveEnforcementPolicy.PermitWithoutStream != nil {
+		defaultConfig.PermitWithoutStream = *k.KeepAliveEnforcementPolicy.PermitWithoutStream
+	}
+
+	return defaultConfig
+}

common/config/config.go

@@ -395,7 +395,6 @@ func RPCFactoryProvider(
 	tracingStatsHandler telemetry.ClientStatsHandler,
 	monitor membership.Monitor,
 ) (common.RPCFactory, error) {
-	svcCfg := cfg.Services[string(svcName)]
 	frontendURL, frontendHTTPURL, frontendHTTPPort, frontendTLSConfig, err := getFrontendConnectionDetails(cfg, tlsConfigProvider, resolver)
 	if err != nil {
 		return nil, err
@@ -405,9 +404,8 @@ func RPCFactoryProvider(
 	if tracingStatsHandler != nil {
 		options = append(options, grpc.WithStatsHandler(tracingStatsHandler))
 	}
-
 	return rpc.NewFactory(
-		&svcCfg.RPC,
+		cfg,
 		svcName,
 		logger,
 		tlsConfigProvider,

common/resource/fx.go

@@ -40,7 +40,8 @@ type RPCFactory interface {
 	GetGRPCListener() net.Listener
 	CreateRemoteFrontendGRPCConnection(rpcAddress string) *grpc.ClientConn
 	CreateLocalFrontendGRPCConnection() *grpc.ClientConn
-	CreateInternodeGRPCConnection(rpcAddress string) *grpc.ClientConn
+	CreateHistoryGRPCConnection(rpcAddress string) *grpc.ClientConn
+	CreateMatchingGRPCConnection(rpcAddress string) *grpc.ClientConn
 	CreateLocalFrontendHTTPClient() (*FrontendHTTPClient, error)
 }
 

common/rpc.go

@@ -53,7 +53,7 @@ var _ common.RPCFactory = (*RPCFactory)(nil)
 
 // RPCFactory is an implementation of common.RPCFactory interface
 type RPCFactory struct {
-	config      *config.RPC
+	config      *config.Config
 	serviceName primitives.ServiceName
 	logger      log.Logger
 
@@ -74,7 +74,7 @@ type RPCFactory struct {
 // NewFactory builds a new RPCFactory
 // conforming to the underlying configuration
 func NewFactory(
-	cfg *config.RPC,
+	cfg *config.Config,
 	sName primitives.ServiceName,
 	logger log.Logger,
 	tlsProvider encryption.TLSConfigProvider,
@@ -150,6 +150,11 @@ func (d *RPCFactory) GetInternodeGRPCServerOptions() ([]grpc.ServerOption, error
 		opts = append(opts, grpc.Creds(credentials.NewTLS(serverConfig)))
 	}
 
+	rpcConfig := d.config.Services[string(d.serviceName)].RPC
+	kep := rpcConfig.KeepAliveServerConfig.GetKeepAliveEnforcementPolicy()
+	kp := rpcConfig.KeepAliveServerConfig.GetKeepAliveServerParameters()
+	opts = append(opts, grpc.KeepaliveEnforcementPolicy(kep), grpc.KeepaliveParams(kp))
+
 	return opts, nil
 }
 
@@ -167,7 +172,8 @@ func (d *RPCFactory) GetGRPCListener() net.Listener {
 }
 
 func (d *RPCFactory) createGRPCListener() net.Listener {
-	hostAddress := net.JoinHostPort(getListenIP(d.config, d.logger).String(), convert.IntToString(d.config.GRPCPort))
+	rpcConfig := d.config.Services[string(d.serviceName)].RPC
+	hostAddress := net.JoinHostPort(getListenIP(&rpcConfig, d.logger).String(), convert.IntToString(rpcConfig.GRPCPort))
 
 	grpcListener, err := net.Listen("tcp", hostAddress)
 	if err != nil || grpcListener == nil || grpcListener.Addr() == nil {
@@ -220,17 +226,18 @@ func (d *RPCFactory) CreateRemoteFrontendGRPCConnection(rpcAddress string) *grpc
 			return nil
 		}
 	}
+	keepAliveOption := d.getClientKeepAliveConfig(primitives.FrontendService)
 
-	return d.dial(rpcAddress, tlsClientConfig)
+	return d.dial(rpcAddress, tlsClientConfig, keepAliveOption)
 }
 
 // CreateLocalFrontendGRPCConnection creates connection for internal frontend calls
 func (d *RPCFactory) CreateLocalFrontendGRPCConnection() *grpc.ClientConn {
 	return d.dial(d.frontendURL, d.frontendTLSConfig)
 }
 
-// CreateInternodeGRPCConnection creates connection for gRPC calls
-func (d *RPCFactory) CreateInternodeGRPCConnection(hostName string) *grpc.ClientConn {
+// createInternodeGRPCConnection creates connection for gRPC calls
+func (d *RPCFactory) createInternodeGRPCConnection(hostName string, serviceName primitives.ServiceName) *grpc.ClientConn {
 	if c, ok := d.interNodeGrpcConnections.Get(hostName).(*grpc.ClientConn); ok {
 		return c
 	}
@@ -243,13 +250,22 @@ func (d *RPCFactory) CreateInternodeGRPCConnection(hostName string) *grpc.Client
 			return nil
 		}
 	}
-	c := d.dial(hostName, tlsClientConfig)
+	c := d.dial(hostName, tlsClientConfig, d.getClientKeepAliveConfig(serviceName))
 	d.interNodeGrpcConnections.Put(hostName, c)
 	return c
 }
 
-func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config) *grpc.ClientConn {
-	connection, err := Dial(hostName, tlsClientConfig, d.logger, d.dialOptions...)
+func (d *RPCFactory) CreateHistoryGRPCConnection(rpcAddress string) *grpc.ClientConn {
+	return d.createInternodeGRPCConnection(rpcAddress, primitives.HistoryService)
+}
+
+func (d *RPCFactory) CreateMatchingGRPCConnection(rpcAddress string) *grpc.ClientConn {
+	return d.createInternodeGRPCConnection(rpcAddress, primitives.MatchingService)
+}
+
+func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config, dialOptions ...grpc.DialOption) *grpc.ClientConn {
+	dialOptions = append(d.dialOptions, dialOptions...)
+	connection, err := Dial(hostName, tlsClientConfig, d.logger, dialOptions...)
 	if err != nil {
 		d.logger.Fatal("Failed to create gRPC connection", tag.Error(err))
 		return nil
@@ -258,6 +274,11 @@ func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config) *grpc.Cl
 	return connection
 }
 
+func (d *RPCFactory) getClientKeepAliveConfig(serviceName primitives.ServiceName) grpc.DialOption {
+	serviceConfig := d.config.Services[string(serviceName)]
+	return grpc.WithKeepaliveParams(serviceConfig.RPC.ClientConnectionConfig.GetKeepAliveClientParameters())
+}
+
 func (d *RPCFactory) GetTLSConfigProvider() encryption.TLSConfigProvider {
 	return d.tlsFactory
 }

common/rpc/rpc.go

@@ -76,6 +76,13 @@ var (
 		MembershipPort: 7600,
 		BindOnIP:       localhostIPv4,
 	}
+	cfg = &config.Config{
+		Services: map[string]config.Service{
+			"frontend": {
+				RPC: *rpcTestCfgDefault,
+			},
+		},
+	}
 	serverCfgInsecure = &config.Global{
 		Membership: config.Membership{
 			MaxJoinDuration:  5,