enable gRPC keepalive to detect dead TCP connections (#217) (#337)

When a TCP connection dies, i.e. no more packet from peer, gRPC
does not close the connection by default.  We need to enable gRPC
keepalive to detect and close dead TCP connections.

Author: mechpen

cmd/kiam/agent.go

@@ -82,7 +82,7 @@ func (opts *agentCommand) Run() {
 	ctxGateway, cancelCtxGateway := context.WithTimeout(context.Background(), opts.timeoutKiamGateway)
 	defer cancelCtxGateway()
 
-	gateway, err := kiamserver.NewGateway(ctxGateway, opts.serverAddress, opts.caPath, opts.certificatePath, opts.keyPath)
+	gateway, err := kiamserver.NewGateway(ctxGateway, opts.serverAddress, opts.caPath, opts.certificatePath, opts.keyPath, opts.keepaliveParams)
 	if err != nil {
 		log.Fatalf("error creating server gateway: %s", err.Error())
 	}

cmd/kiam/health.go

@@ -43,7 +43,7 @@ func (opts *healthCommand) Run() {
 	ctxGateway, cancelCtxGateway := context.WithTimeout(context.Background(), opts.timeoutKiamGateway)
 	defer cancelCtxGateway()
 
-	gateway, err := kiamserver.NewGateway(ctxGateway, opts.serverAddress, opts.caPath, opts.certificatePath, opts.keyPath)
+	gateway, err := kiamserver.NewGateway(ctxGateway, opts.serverAddress, opts.caPath, opts.certificatePath, opts.keyPath, opts.keepaliveParams)
 	if err != nil {
 		log.Fatalf("error creating server gateway: %s", err.Error())
 	}

cmd/kiam/opts.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc/keepalive"
 	"github.com/uswitch/kiam/pkg/pprof"
 	"github.com/uswitch/kiam/pkg/prometheus"
 	"github.com/uswitch/kiam/pkg/statsd"
@@ -109,9 +110,13 @@ type clientOptions struct {
 	serverAddress        string
 	serverAddressRefresh time.Duration
 	timeoutKiamGateway   time.Duration
+	keepaliveParams      keepalive.ClientParameters
 }
 
 func (o *clientOptions) bind(parser parser) {
+	parser.Flag("grpc-keepalive-time-ms", "gRPC keepalive time").Default("10s").DurationVar(&o.keepaliveParams.Time)
+	parser.Flag("grpc-keepalive-timeout-ms", "gRPC keepalive timeout").Default("2s").DurationVar(&o.keepaliveParams.Timeout)
+	parser.Flag("grpc-keepalive-permit-without-stream", "gRPC keepalive ping even with no RPC").BoolVar(&o.keepaliveParams.PermitWithoutStream)
 	parser.Flag("server-address", "gRPC address to Kiam server service").Default("localhost:9610").StringVar(&o.serverAddress)
 	parser.Flag("server-address-refresh", "Interval to refresh server service endpoints ( deprecated )").Default("0s").DurationVar(&o.serverAddressRefresh)
 	parser.Flag("gateway-timeout-creation", "Timeout to create the kiam gateway ").Default("1s").DurationVar(&o.timeoutKiamGateway)

pkg/server/gateway.go

@@ -31,6 +31,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/balancer/roundrobin"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/keepalive"
 
 	status "google.golang.org/grpc/status"
 )
@@ -53,7 +54,7 @@ const (
 )
 
 // NewGateway constructs a gRPC client to talk to the server
-func NewGateway(ctx context.Context, address string, caFile, certificateFile, keyFile string) (*KiamGateway, error) {
+func NewGateway(ctx context.Context, address string, caFile, certificateFile, keyFile string, keepaliveParams keepalive.ClientParameters) (*KiamGateway, error) {
 	callOpts := []retry.CallOption{
 		retry.WithBackoff(retry.BackoffLinear(RetryInterval)),
 	}
@@ -85,6 +86,7 @@ func NewGateway(ctx context.Context, address string, caFile, certificateFile, ke
 	dialAddress := fmt.Sprintf("dns:///%s", address)
 
 	dialOpts := []grpc.DialOption{
+		grpc.WithKeepaliveParams(keepaliveParams),
 		grpc.WithTransportCredentials(creds),
 		grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(retry.UnaryClientInterceptor(callOpts...), grpc_prometheus.UnaryClientInterceptor)),
 		grpc.WithBalancerName(roundrobin.Name),