proxy: add idle timeout (#2319)

config/envoyconfig/routes.go

@@ -495,9 +495,10 @@ func getRequestHeadersToRemove(options *config.Options, policy *config.Policy) [
 
 func getRouteTimeout(options *config.Options, policy *config.Policy) *durationpb.Duration {
 	var routeTimeout *durationpb.Duration
-	if policy.UpstreamTimeout != 0 {
-		routeTimeout = durationpb.New(policy.UpstreamTimeout)
+	if policy.UpstreamTimeout != nil {
+		routeTimeout = durationpb.New(*policy.UpstreamTimeout)
 	} else if shouldDisableTimeouts(policy) {
+		// a non-zero value would conflict with idleTimeout and/or websocket / tcp calls
 		routeTimeout = durationpb.New(0)
 	} else {
 		routeTimeout = durationpb.New(options.DefaultUpstreamTimeout)
@@ -507,14 +508,17 @@ func getRouteTimeout(options *config.Options, policy *config.Policy) *durationpb
 
 func getRouteIdleTimeout(policy *config.Policy) *durationpb.Duration {
 	var idleTimeout *durationpb.Duration
-	if shouldDisableTimeouts(policy) {
+	if policy.IdleTimeout != nil {
+		idleTimeout = durationpb.New(*policy.IdleTimeout)
+	} else if shouldDisableTimeouts(policy) {
 		idleTimeout = durationpb.New(0)
 	}
 	return idleTimeout
 }
 
 func shouldDisableTimeouts(policy *config.Policy) bool {
-	return policy.AllowWebsockets ||
+	return policy.IdleTimeout != nil ||
+		policy.AllowWebsockets ||
 		urlutil.IsTCP(policy.Source.URL) ||
 		policy.IsForKubernetes() // disable for kubernetes so that tailing logs works (#2182)
 }

config/envoyconfig/routes_test.go

@@ -211,12 +211,69 @@ func Test_buildControlPlanePrefixRoute(t *testing.T) {
 	`, route)
 }
 
+func TestTimeouts(t *testing.T) {
+	defer func(f func(*config.Policy) string) {
+		getClusterID = f
+	}(getClusterID)
+	getClusterID = func(*config.Policy) string { return "policy" }
+
+	getDuration := func(txt string) *time.Duration {
+		if txt == "" {
+			return nil
+		}
+		d, err := time.ParseDuration(txt)
+		require.NoError(t, err, txt)
+		return &d
+	}
+
+	testCases := []struct {
+		upstream, idle, expect string
+	}{
+		{"", "", `"timeout": "3s"`},
+		{"", "0s", `"timeout": "0s","idleTimeout": "0s"`},
+		{"", "5s", `"timeout": "0s","idleTimeout": "5s"`},
+		{"5s", "", `"timeout": "5s"`},
+		{"5s", "4s", `"timeout": "5s","idleTimeout": "4s"`},
+		{"0s", "4s", `"timeout": "0s","idleTimeout": "4s"`},
+	}
+
+	for _, tc := range testCases {
+		b := &Builder{filemgr: filemgr.NewManager()}
+		routes, err := b.buildPolicyRoutes(&config.Options{
+			CookieName:             "pomerium",
+			DefaultUpstreamTimeout: time.Second * 3,
+			Policies: []config.Policy{
+				{
+					Source:          &config.StringURL{URL: mustParseURL(t, "https://example.com")},
+					Path:            "/test",
+					UpstreamTimeout: getDuration(tc.upstream),
+					IdleTimeout:     getDuration(tc.idle),
+				}},
+		}, "example.com")
+		if !assert.NoError(t, err, "%v", tc) || !assert.Len(t, routes, 1, tc) || !assert.NotNil(t, routes[0].GetRoute(), "%v", tc) {
+			continue
+		}
+		testutil.AssertProtoJSONEqual(t, fmt.Sprintf(`{
+			%s,
+			"autoHostRewrite": true,
+			"cluster": "policy",
+			"upgradeConfigs": [
+				{ "enabled": false, "upgradeType": "websocket"},
+				{ "enabled": false, "upgradeType": "spdy/3.1"}
+			]
+		}`, tc.expect), routes[0].GetRoute())
+	}
+}
+
 func Test_buildPolicyRoutes(t *testing.T) {
 	defer func(f func(*config.Policy) string) {
 		getClusterID = f
 	}(getClusterID)
 	getClusterID = policyNameFunc()
 
+	oneMinute := time.Minute
+	ten := time.Second * 10
+
 	b := &Builder{filemgr: filemgr.NewManager()}
 	routes, err := b.buildPolicyRoutes(&config.Options{
 		CookieName:             "pomerium",
@@ -241,7 +298,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 				Source:              &config.StringURL{URL: mustParseURL(t, "https://example.com")},
 				Prefix:              "/some/prefix/",
 				SetRequestHeaders:   map[string]string{"HEADER-KEY": "HEADER-VALUE"},
-				UpstreamTimeout:     time.Minute,
+				UpstreamTimeout:     &oneMinute,
 				PassIdentityHeaders: true,
 			},
 			{
@@ -253,7 +310,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 				Source:               &config.StringURL{URL: mustParseURL(t, "https://example.com")},
 				Prefix:               "/some/prefix/",
 				RemoveRequestHeaders: []string{"HEADER-KEY"},
-				UpstreamTimeout:      time.Minute,
+				UpstreamTimeout:      &oneMinute,
 				PassIdentityHeaders:  true,
 			},
 			{
@@ -277,7 +334,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 				AllowWebsockets:     true,
 				PreserveHostHeader:  true,
 				PassIdentityHeaders: true,
-				UpstreamTimeout:     time.Second * 10,
+				UpstreamTimeout:     &ten,
 			},
 		},
 	}, "example.com")
@@ -599,7 +656,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 				{
 					Source:              &config.StringURL{URL: mustParseURL(t, "tcp+https://example.com:22")},
 					PassIdentityHeaders: true,
-					UpstreamTimeout:     time.Second * 10,
+					UpstreamTimeout:     &ten,
 				},
 			},
 		}, "example.com:22")

config/policy.go

@@ -13,7 +13,7 @@ import (
 	"time"
 
 	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
-	"github.com/golang/protobuf/ptypes"
+	"google.golang.org/protobuf/types/known/durationpb"
 
 	"github.com/pomerium/pomerium/internal/hashutil"
 	"github.com/pomerium/pomerium/internal/identity"
@@ -69,8 +69,13 @@ type Policy struct {
 	AllowAnyAuthenticatedUser bool `mapstructure:"allow_any_authenticated_user" yaml:"allow_any_authenticated_user,omitempty"`
 
 	// UpstreamTimeout is the route specific timeout. Must be less than the global
-	// timeout. If unset,  route will fallback to the proxy's DefaultUpstreamTimeout.
-	UpstreamTimeout time.Duration `mapstructure:"timeout" yaml:"timeout,omitempty"`
+	// timeout. If unset, route will fallback to the proxy's DefaultUpstreamTimeout.
+	UpstreamTimeout *time.Duration `mapstructure:"timeout" yaml:"timeout,omitempty"`
+
+	// IdleTimeout is distinct from UpstreamTimeout and defines period of time there may be no data over this connection
+	// value of zero completely disables this setting
+	// see https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/route/v3/route_components.proto#envoy-v3-api-field-config-route-v3-routeaction-idle-timeout
+	IdleTimeout *time.Duration `mapstructure:"idle_timeout" yaml:"idle_timeout,omitempty"`
 
 	// Enable proxying of websocket connections by removing the default timeout handler.
 	// Caution: Enabling this feature could result in abuse via DOS attacks.
@@ -188,7 +193,16 @@ type PolicyRedirect struct {
 
 // NewPolicyFromProto creates a new Policy from a protobuf policy config route.
 func NewPolicyFromProto(pb *configpb.Route) (*Policy, error) {
-	timeout, _ := ptypes.Duration(pb.GetTimeout())
+	var timeout *time.Duration
+	if pb.GetTimeout() != nil {
+		t := pb.GetTimeout().AsDuration()
+		timeout = &t
+	}
+	var idleTimeout *time.Duration
+	if pb.GetIdleTimeout() != nil {
+		t := pb.GetIdleTimeout().AsDuration()
+		idleTimeout = &t
+	}
 
 	p := &Policy{
 		From:                             pb.GetFrom(),
@@ -206,6 +220,7 @@ func NewPolicyFromProto(pb *configpb.Route) (*Policy, error) {
 		AllowPublicUnauthenticatedAccess: pb.GetAllowPublicUnauthenticatedAccess(),
 		AllowAnyAuthenticatedUser:        pb.GetAllowAnyAuthenticatedUser(),
 		UpstreamTimeout:                  timeout,
+		IdleTimeout:                      idleTimeout,
 		AllowWebsockets:                  pb.GetAllowWebsockets(),
 		TLSSkipVerify:                    pb.GetTlsSkipVerify(),
 		TLSServerName:                    pb.GetTlsServerName(),
@@ -278,7 +293,14 @@ func NewPolicyFromProto(pb *configpb.Route) (*Policy, error) {
 
 // ToProto converts the policy to a protobuf type.
 func (p *Policy) ToProto() (*configpb.Route, error) {
-	timeout := ptypes.DurationProto(p.UpstreamTimeout)
+	var timeout *durationpb.Duration
+	if p.UpstreamTimeout == nil {
+		timeout = durationpb.New(defaultOptions.DefaultUpstreamTimeout)
+	}
+	var idleTimeout *durationpb.Duration
+	if p.IdleTimeout != nil {
+		idleTimeout = durationpb.New(*p.IdleTimeout)
+	}
 	sps := make([]*configpb.Policy, 0, len(p.SubPolicies))
 	for _, sp := range p.SubPolicies {
 		sps = append(sps, &configpb.Policy{
@@ -309,6 +331,7 @@ func (p *Policy) ToProto() (*configpb.Route, error) {
 		AllowPublicUnauthenticatedAccess: p.AllowPublicUnauthenticatedAccess,
 		AllowAnyAuthenticatedUser:        p.AllowAnyAuthenticatedUser,
 		Timeout:                          timeout,
+		IdleTimeout:                      idleTimeout,
 		AllowWebsockets:                  p.AllowWebsockets,
 		TlsSkipVerify:                    p.TLSSkipVerify,
 		TlsServerName:                    p.TLSServerName,

docs/reference/readme.md

@@ -1334,6 +1334,19 @@ See Envoy [documentation](https://www.envoyproxy.io/docs/envoy/latest/intro/arch
 Policy timeout establishes the per-route timeout value. Cannot exceed global timeout values.
 
 
+### Idle Timeout
+- `yaml`/`json` setting: `idle_timeout`
+- Type: [Go Duration](https://golang.org/pkg/time/#Duration.String) `string`
+- Optional
+- Default: `5m`
+
+If you are proxying long-lived requests that employ streaming calls such as websockets or gRPC,
+set this to either a maximum value there may be no data exchange over a connection (recommended),
+or set it to unlimited (`0s`). If `idle_timeout` is specified, and `timeout` is not
+explicitly set, then `timeout` would be unlimited (`0s`). You still may specify maximum lifetime
+of the connection using `timeout` value (i.e. to 1 day).
+
+
 ### Set Request Headers
 - Config File Key: `set_request_headers`
 - Type: map of `strings` key value pairs

docs/reference/settings.yaml

@@ -1469,6 +1469,19 @@ settings:
           - Default: `30s`
         doc: |
           Policy timeout establishes the per-route timeout value. Cannot exceed global timeout values.
+      - name: "Idle Timeout"
+        keys: ["idle_timeout"]
+        attributes: |
+          - `yaml`/`json` setting: `idle_timeout`
+          - Type: [Go Duration](https://golang.org/pkg/time/#Duration.String) `string`
+          - Optional
+          - Default: `5m`
+        doc: |
+          If you are proxying long-lived requests that employ streaming calls such as websockets or gRPC,
+          set this to either a maximum value there may be no data exchange over a connection (recommended),
+          or set it to unlimited (`0s`). If `idle_timeout` is specified, and `timeout` is not
+          explicitly set, then `timeout` would be unlimited (`0s`). You still may specify maximum lifetime
+          of the connection using `timeout` value (i.e. to 1 day).
       - name: "Set Request Headers"
         keys: ["set_request_headers"]
         attributes: |

internal/testutil/testutil.go

@@ -16,6 +16,7 @@ import (
 // AssertProtoJSONEqual asserts that a protobuf message matches the given JSON. The protoMsg can also be a slice
 // of protobuf messages.
 func AssertProtoJSONEqual(t *testing.T, expected string, protoMsg interface{}, msgAndArgs ...interface{}) bool {
+	t.Helper()
 	protoMsgVal := reflect.ValueOf(protoMsg)
 	if protoMsgVal.Kind() == reflect.Slice {
 		var protoMsgs []json.RawMessage

pkg/grpc/audit/audit.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: audit.proto
 

pkg/grpc/config/config.proto

@@ -66,6 +66,7 @@ message Route {
   bool allow_public_unauthenticated_access = 11;
   bool allow_any_authenticated_user = 33;
   google.protobuf.Duration timeout = 12;
+  google.protobuf.Duration idle_timeout = 43;
   bool allow_websockets = 13;
 
   bool tls_skip_verify = 14;

pkg/grpc/crypt/crypt.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: crypt.proto
 

pkg/grpc/databroker/databroker.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: databroker.proto
 

pkg/grpc/directory/directory.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: directory.proto
 

pkg/grpc/events/xds.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: xds.proto
 

pkg/grpc/registry/registry.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: registry.proto
 

pkg/grpc/session/session.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: session.proto
 

pkg/grpc/user/user.pb.go

@@ -1,6 +1,6 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.26.0
+// 	protoc-gen-go v1.27.0
 // 	protoc        v3.14.0
 // source: user.proto
 

proxy/proxy_test.go

@@ -156,7 +156,9 @@ func Test_UpdateOptions(t *testing.T) {
 	allowWebSockets := testOptions(t)
 	allowWebSockets.Policies = []config.Policy{{To: toFoo, From: "http://bar.example", AllowWebsockets: true}}
 	customTimeout := testOptions(t)
-	customTimeout.Policies = []config.Policy{{To: toFoo, From: "http://bar.example", UpstreamTimeout: 10 * time.Second}}
+
+	ten := 10 * time.Second
+	customTimeout.Policies = []config.Policy{{To: toFoo, From: "http://bar.example", UpstreamTimeout: &ten}}
 	corsPreflight := testOptions(t)
 	corsPreflight.Policies = []config.Policy{{To: toFoo, From: "http://bar.example", CORSAllowPreflight: true}}
 	disableAuth := testOptions(t)