feat: configure default grpc server keepalive enforcement policy (#589)

### Motivation

configure default grpc server keepalive enforcement policy

### Modification

- DefaultGrpcKeepAliveMinTime        = 5 * time.Second
- DefaultGrpcKeepPermitWithoutStream = true
- DefaultGrpcClientKeepAliveTime       = time.Second * 10
- DefaultGrpcClientKeepAliveTimeout    = time.Second * 5
- DefaultGrpcClientPermitWithoutStream = true

common/client_pool.go

@@ -17,6 +17,7 @@ package common
 import (
 	"context"
 	"crypto/tls"
+	"google.golang.org/grpc/keepalive"
 	"io"
 	"log/slog"
 	"strings"
@@ -39,6 +40,11 @@ import (
 
 const DefaultRpcTimeout = 30 * time.Second
 const AddressSchemaTLS = "tls://"
+const (
+	defaultGrpcClientKeepAliveTime       = time.Second * 10
+	defaultGrpcClientKeepAliveTimeout    = time.Second * 5
+	defaultGrpcClientPermitWithoutStream = true
+)
 
 type ClientPool interface {
 	io.Closer
@@ -178,6 +184,11 @@ func (cp *clientPool) newConnection(target string) (*grpc.ClientConn, error) {
 		grpc.WithTransportCredentials(tcs),
 		grpc.WithStreamInterceptor(grpcprometheus.StreamClientInterceptor),
 		grpc.WithUnaryInterceptor(grpcprometheus.UnaryClientInterceptor),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			PermitWithoutStream: defaultGrpcClientPermitWithoutStream,
+			Time:                defaultGrpcClientKeepAliveTime,
+			Timeout:             defaultGrpcClientKeepAliveTimeout,
+		}),
 	}
 	if cp.authentication != nil {
 		options = append(options, grpc.WithPerRPCCredentials(cp.authentication))

common/container/container.go

@@ -17,10 +17,12 @@ package container
 import (
 	"context"
 	"crypto/tls"
+	"google.golang.org/grpc/keepalive"
 	"io"
 	"log/slog"
 	"net"
 	"os"
+	"time"
 
 	"github.com/streamnative/oxia/server/auth"
 
@@ -34,9 +36,10 @@ import (
 )
 
 const (
-	maxGrpcFrameSize = 256 * 1024 * 1024
-
-	ReadinessProbeService = "oxia-readiness"
+	maxGrpcFrameSize                         = 256 * 1024 * 1024
+	defaultGrpcServerKeepAliveMinTime        = 5 * time.Second
+	defaultGrpcServerKeepPermitWithoutStream = true
+	ReadinessProbeService                    = "oxia-readiness"
 )
 
 type GrpcServer interface {
@@ -102,6 +105,10 @@ func newDefaultGrpcProvider(name, bindAddress string, registerFunc func(grpc.Ser
 			grpc.ChainStreamInterceptor(streamInterceptors...),
 			grpc.ChainUnaryInterceptor(unaryInterceptors...),
 			grpc.MaxRecvMsgSize(maxGrpcFrameSize),
+			grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
+				MinTime:             defaultGrpcServerKeepAliveMinTime,
+				PermitWithoutStream: defaultGrpcServerKeepPermitWithoutStream,
+			}),
 		),
 	}
 	registerFunc(c.server)