connect: add gRPC keep-alive (#4961)

internal/zero/api/api.go

@@ -6,6 +6,9 @@ import (
 	"fmt"
 	"time"
 
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
+
 	"github.com/pomerium/pomerium/internal/zero/apierror"
 	connect_mux "github.com/pomerium/pomerium/internal/zero/connect-mux"
 	"github.com/pomerium/pomerium/internal/zero/grpcconn"
@@ -34,6 +37,13 @@ const (
 	minTelemetryTokenTTL = time.Minute * 5
 )
 
+// see https://github.com/pomerium/pomerium-zero/issues/1711
+var connectClientKeepaliveParams = keepalive.ClientParameters{
+	Time:                time.Minute, // send pings every minute
+	Timeout:             time.Minute, // wait 1 minute for ping ack
+	PermitWithoutStream: false,
+}
+
 // WatchOption defines which events to watch for
 type WatchOption = connect_mux.WatchOption
 
@@ -60,7 +70,7 @@ func NewAPI(ctx context.Context, opts ...Option) (*API, error) {
 
 	connectGRPCConn, err := grpcconn.New(ctx, cfg.connectAPIEndpoint, func(ctx context.Context) (string, error) {
 		return tokenCache.GetToken(ctx, minConnectTokenTTL)
-	})
+	}, grpc.WithKeepaliveParams(connectClientKeepaliveParams))
 	if err != nil {
 		return nil, fmt.Errorf("error creating connect grpc client: %w", err)
 	}

internal/zero/grpcconn/client.go

@@ -25,8 +25,9 @@ func New(
 	ctx context.Context,
 	endpoint string,
 	tokenProvider TokenProviderFn,
+	dialOpts ...grpc.DialOption,
 ) (*grpc.ClientConn, error) {
-	cfg, err := getConfig(endpoint)
+	cfg, err := getConfig(endpoint, dialOpts...)
 	if err != nil {
 		return nil, err
 	}

internal/zero/grpcconn/config.go

@@ -25,8 +25,11 @@ type config struct {
 // NewConfig returns a new Config from an endpoint string, that has to be in a URL format.
 // The endpoint can be either http:// or https:// that will be used to determine whether TLS should be used.
 // if port is not specified, it will be inferred from the scheme (80 for http, 443 for https).
-func getConfig(endpoint string) (*config, error) {
-	c := new(config)
+func getConfig(
+	endpoint string,
+	opts ...grpc.DialOption,
+) (*config, error) {
+	c := &config{opts: opts}
 	err := c.parseEndpoint(endpoint)
 	if err != nil {
 		return nil, fmt.Errorf("invalid endpoint: %w", err)