GRPC Improvements

docs/docs/CHANGELOG.md

@@ -5,6 +5,11 @@
 ### New
 
 - Add ability to set client certificates for downstream connections. [GH-259]
+- GRPC Improvements. [#261](https://github.com/pomerium/pomerium/pull/261) and [#69](https://github.com/pomerium/pomerium/issues/69)
+  - Enable WaitForReady to allow background retries through transient failures
+  - Expose a configurable timeout for backend requests to Authorize and Authenticate
+  - Enable DNS round_robin load balancing to Authorize and Authenticate services by default
+
 
 ### Fixed
 

docs/docs/reference/reference.md

@@ -144,6 +144,31 @@ Timeouts set the global server timeouts. For route-specific timeouts, see [polic
 
 > For a deep dive on timeout values see [these](https://blog.cloudflare.com/the-complete-guide-to-golang-net-http-timeouts/) [two](https://blog.cloudflare.com/exposing-go-on-the-internet/) excellent blog posts.
 
+## GRPC Options
+
+These settings control upstream connections to the Authorize and Authenticate services.
+
+### GRPC Client Timeout
+
+Maxmimum time before canceling an upstream RPC request.  During transient failures, the proxy will retry upstreams for this duration, if possible.  You should leave this high enough to handle backend service restart and rediscovery so that client requests do not fail.
+
+- Environmental Variable: `GRPC_CLIENT_TIMEOUT` 
+- Config File Key: `grpc_client_timeout`
+- Type: [Go Duration](https://golang.org/pkg/time/#Duration.String) `string`
+- Default: `10s`
+
+### GRPC Client DNS RoundRobin
+
+Enable grpc DNS based round robin load balancing.  This method uses DNS to resolve endpoints and does client side load balancing of _all_ addresses returned by the DNS record.  Do not disable unless you have a specific use case.
+
+- Environmental Variable: `GRPC_CLIENT_DNS_ROUNDROBIN`
+- Config File Key: `grpc_client_dns_roundrobin`
+- Type: `bool`
+- Default: `true`
+
+
+
+
 ## HTTP Redirect Address
 
 - Environmental Variable: `HTTP_REDIRECT_ADDR`

internal/config/options.go

@@ -145,6 +145,10 @@ type Options struct {
 	// AgentEndpoint instructs exporter to send spans to jaeger-agent at this address.
 	// For example, localhost:6831.
 	TracingJaegerAgentEndpoint string `mapstructure:"tracing_jaeger_agent_endpoint"`
+
+	// GRPC Service Settings
+	GRPCClientTimeout       time.Duration `mapstructure:"grpc_client_timeout"`
+	GRPCClientDNSRoundRobin bool          `mapstructure:"grpc_client_dns_roundrobin"`
 }
 
 var defaultOptions = Options{
@@ -163,14 +167,16 @@ var defaultOptions = Options{
 		"X-XSS-Protection":          "1; mode=block",
 		"Strict-Transport-Security": "max-age=31536000; includeSubDomains; preload",
 	},
-	Addr:              ":https",
+	Addr:                    ":https",
-	CertFile:          filepath.Join(fileutil.Getwd(), "cert.pem"),
+	CertFile:                filepath.Join(fileutil.Getwd(), "cert.pem"),
-	KeyFile:           filepath.Join(fileutil.Getwd(), "privkey.pem"),
+	KeyFile:                 filepath.Join(fileutil.Getwd(), "privkey.pem"),
-	ReadHeaderTimeout: 10 * time.Second,
+	ReadHeaderTimeout:       10 * time.Second,
-	ReadTimeout:       30 * time.Second,
+	ReadTimeout:             30 * time.Second,
-	WriteTimeout:      0, // support streaming by default
+	WriteTimeout:            0, // support streaming by default
-	IdleTimeout:       5 * time.Minute,
+	IdleTimeout:             5 * time.Minute,
-	RefreshCooldown:   5 * time.Minute,
+	RefreshCooldown:         5 * time.Minute,
+	GRPCClientTimeout:       10 * time.Second, // Try to withstand transient service failures for a single request
+	GRPCClientDNSRoundRobin: true,
 }
 
 // NewOptions returns a minimal options configuration built from default options.

internal/config/options_test.go

@@ -337,7 +337,7 @@ func TestNewOptions(t *testing.T) {
 
 func TestOptionsFromViper(t *testing.T) {
 	opts := []cmp.Option{
-		cmpopts.IgnoreFields(Options{}, "AuthenticateInternalAddr", "DefaultUpstreamTimeout", "CookieRefresh", "CookieExpire", "Services", "Addr", "RefreshCooldown", "LogLevel", "KeyFile", "CertFile", "SharedKey", "ReadTimeout", "ReadHeaderTimeout", "IdleTimeout"),
+		cmpopts.IgnoreFields(Options{}, "AuthenticateInternalAddr", "DefaultUpstreamTimeout", "CookieRefresh", "CookieExpire", "Services", "Addr", "RefreshCooldown", "LogLevel", "KeyFile", "CertFile", "SharedKey", "ReadTimeout", "ReadHeaderTimeout", "IdleTimeout", "GRPCClientTimeout", "GRPCClientDNSRoundRobin"),
 		cmpopts.IgnoreFields(Policy{}, "Source", "Destination"),
 	}
 

proxy/clients/clients.go

@@ -1,6 +1,7 @@
 package clients // import "github.com/pomerium/pomerium/proxy/clients"
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/base64"
@@ -9,13 +10,14 @@ import (
 	"io/ioutil"
 	"net/url"
 	"strings"
+	"time"
 
 	"github.com/pomerium/pomerium/internal/log"
 	"github.com/pomerium/pomerium/internal/middleware"
 	"github.com/pomerium/pomerium/internal/telemetry/metrics"
-
 	"go.opencensus.io/plugin/ocgrpc"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer/roundrobin"
 	"google.golang.org/grpc/credentials"
 )
 
@@ -38,6 +40,10 @@ type Options struct {
 	CA string
 	// CAFile specifies the TLS certificate authority file to use.
 	CAFile string
+	// RequestTimeout specifies the timeout for individual RPC calls
+	RequestTimeout time.Duration
+	// ClientDNSRoundRobin enables or disables DNS resolver based load balancing
+	ClientDNSRoundRobin bool
 }
 
 // NewGRPCClientConn returns a new gRPC pomerium service client connection.
@@ -102,11 +108,36 @@ func NewGRPCClientConn(opts *Options) (*grpc.ClientConn, error) {
 			return nil, err
 		}
 	}
-	return grpc.Dial(
+
-		connAddr,
+	dialOptions := []grpc.DialOption{
 		grpc.WithTransportCredentials(cert),
 		grpc.WithPerRPCCredentials(grpcAuth),
-		grpc.WithUnaryInterceptor(metrics.GRPCClientInterceptor("proxy")),
+		grpc.WithChainUnaryInterceptor(metrics.GRPCClientInterceptor("proxy"), grpcTimeoutInterceptor(opts.RequestTimeout)),
 		grpc.WithStatsHandler(&ocgrpc.ClientHandler{}),
+		grpc.WithDefaultCallOptions([]grpc.CallOption{grpc.WaitForReady(true)}...),
+	}
+
+	if opts.ClientDNSRoundRobin {
+		dialOptions = append(dialOptions, grpc.WithBalancerName(roundrobin.Name))
+		connAddr = fmt.Sprintf("dns:///%s", connAddr)
+	}
+
+	return grpc.Dial(
+		connAddr,
+		dialOptions...,
 	)
 }
+
+// grpcTimeoutInterceptor enforces per-RPC request timeouts
+func grpcTimeoutInterceptor(timeout time.Duration) grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		if timeout <= 0 {
+			return invoker(ctx, method, req, reply, cc, opts...)
+		}
+
+		ctx, cancel := context.WithTimeout(ctx, timeout)
+		defer cancel()
+		return invoker(ctx, method, req, reply, cc, opts...)
+
+	}
+}

proxy/clients/clients_test.go

@@ -0,0 +1,37 @@
+package clients // import "github.com/pomerium/pomerium/proxy/clients"
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"google.golang.org/grpc"
+)
+
+func Test_grpcTimeoutInterceptor(t *testing.T) {
+
+	mockInvoker := func(sleepTime time.Duration, wantFail bool) grpc.UnaryInvoker {
+		return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, opts ...grpc.CallOption) error {
+			time.Sleep(sleepTime)
+			select {
+			case <-ctx.Done():
+				if !wantFail {
+					t.Error("Deadline should not have been exceeded")
+				}
+				return nil
+			default:
+				if wantFail {
+					t.Error("Deadline not exceeded but should have been")
+				}
+			}
+			return nil
+		}
+	}
+
+	timeOut := 5 * time.Millisecond
+	to := grpcTimeoutInterceptor(timeOut)
+
+	to(context.Background(), "test", nil, nil, nil, mockInvoker(timeOut*2, true))
+	to(context.Background(), "test", nil, nil, nil, mockInvoker(timeOut/2, false))
+
+}

proxy/proxy.go

@@ -182,6 +182,8 @@ func New(opts config.Options) (*Proxy, error) {
 			SharedSecret:            opts.SharedKey,
 			CA:                      opts.CA,
 			CAFile:                  opts.CAFile,
+			RequestTimeout:          opts.GRPCClientTimeout,
+			ClientDNSRoundRobin:     opts.GRPCClientDNSRoundRobin,
 		})
 	if err != nil {
 		return nil, err
@@ -193,6 +195,8 @@ func New(opts config.Options) (*Proxy, error) {
 			SharedSecret:            opts.SharedKey,
 			CA:                      opts.CA,
 			CAFile:                  opts.CAFile,
+			RequestTimeout:          opts.GRPCClientTimeout,
+			ClientDNSRoundRobin:     opts.GRPCClientDNSRoundRobin,
 		})
 	return p, err
 }