feat(injector): Set probe timeouts based on pod deployment spec (#4149)

Fixes #4137

Overall we aim to maintain the following invariance:

For all health probes we want the following to be true:

All Envoy (implicit or explicit) timeouts >= the timeout specified in
the pod deployment spec.

Changes:

* Remove connect timeouts from all clusters
* For health probe clusters we set the route timeout to be equal to the
timeout set on the kubernetes pod. This is only applied on HTTP routes
since plain tcp routes have infinite timeout.

Signed-off-by: Sotiris Nanopoulos <[email protected]>

Author: Sotiris Nanopoulos

pkg/envoy/cds/cluster.go

@@ -23,11 +23,6 @@ import (
 	"github.com/openservicemesh/osm/pkg/trafficpolicy"
 )
 
-const (
-	// clusterConnectTimeout is the timeout duration used by Envoy to timeout connections to the cluster
-	clusterConnectTimeout = 1 * time.Second
-)
-
 // replacer used to configure an Envoy cluster's altStatName
 var replacer = strings.NewReplacer(".", "_", ":", "_")
 
@@ -49,7 +44,6 @@ func getUpstreamServiceCluster(downstreamIdentity identity.ServiceIdentity, conf
 
 	remoteCluster := &xds_cluster.Cluster{
 		Name:                          config.Name,
-		ConnectTimeout:                ptypes.DurationProto(clusterConnectTimeout),
 		TypedExtensionProtocolOptions: HTTP2ProtocolOptions,
 		TransportSocket: &xds_core.TransportSocket{
 			Name: wellknown.TransportSocketTls,
@@ -78,8 +72,7 @@ func getMulticlusterGatewayUpstreamServiceCluster(catalog catalog.MeshCataloger,
 	}
 
 	remoteCluster := &xds_cluster.Cluster{
-		Name:           upstreamSvc.ServerName(),
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
+		Name: upstreamSvc.ServerName(),
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_STRICT_DNS,
 		},
@@ -142,11 +135,10 @@ func getLocalServiceCluster(config trafficpolicy.MeshClusterConfig) *xds_cluster
 
 	return &xds_cluster.Cluster{
 		// The name must match the domain being cURLed in the demo
-		Name:           config.Name,
-		AltStatName:    config.Name,
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
-		LbPolicy:       xds_cluster.Cluster_ROUND_ROBIN,
-		RespectDnsTtl:  true,
+		Name:          config.Name,
+		AltStatName:   config.Name,
+		LbPolicy:      xds_cluster.Cluster_ROUND_ROBIN,
+		RespectDnsTtl: true,
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_STRICT_DNS,
 		},
@@ -180,9 +172,8 @@ func getLocalServiceCluster(config trafficpolicy.MeshClusterConfig) *xds_cluster
 // getPrometheusCluster returns an Envoy Cluster responsible for scraping metrics by Prometheus
 func getPrometheusCluster() *xds_cluster.Cluster {
 	return &xds_cluster.Cluster{
-		Name:           constants.EnvoyMetricsCluster,
-		AltStatName:    constants.EnvoyMetricsCluster,
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
+		Name:        constants.EnvoyMetricsCluster,
+		AltStatName: constants.EnvoyMetricsCluster,
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_STATIC,
 		},
@@ -259,9 +250,8 @@ func getDNSResolvableEgressCluster(config *trafficpolicy.EgressClusterConfig) (*
 	}
 
 	return &xds_cluster.Cluster{
-		Name:           config.Name,
-		AltStatName:    formatAltStatNameForPrometheus(config.Name),
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
+		Name:        config.Name,
+		AltStatName: formatAltStatNameForPrometheus(config.Name),
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_STRICT_DNS,
 		},
@@ -295,8 +285,7 @@ func getOriginalDestinationEgressCluster(name string) (*xds_cluster.Cluster, err
 	}
 
 	return &xds_cluster.Cluster{
-		Name:           name,
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
+		Name: name,
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_ORIGINAL_DST,
 		},

pkg/envoy/cds/cluster_test.go

@@ -2,13 +2,11 @@ package cds
 
 import (
 	"testing"
-	"time"
 
 	xds_cluster "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	xds_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	xds_endpoint "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	"github.com/golang/mock/gomock"
-	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/wrappers"
 	tassert "github.com/stretchr/testify/assert"
 
@@ -188,7 +186,6 @@ func TestGetLocalServiceCluster(t *testing.T) {
 			} else {
 				assert.Equal(tc.clusterConfig.Name, cluster.Name)
 				assert.Equal(tc.clusterConfig.Name, cluster.AltStatName)
-				assert.Equal(ptypes.DurationProto(clusterConnectTimeout), cluster.ConnectTimeout)
 				assert.Equal(xds_cluster.Cluster_ROUND_ROBIN, cluster.LbPolicy)
 				assert.Equal(&xds_cluster.Cluster_Type{Type: xds_cluster.Cluster_STRICT_DNS}, cluster.ClusterDiscoveryType)
 				assert.Equal(true, cluster.RespectDnsTtl)
@@ -209,7 +206,6 @@ func TestGetPrometheusCluster(t *testing.T) {
 		AltStatName:            constants.EnvoyMetricsCluster,
 		ClusterDiscoveryType:   &xds_cluster.Cluster_Type{Type: xds_cluster.Cluster_STATIC},
 		EdsClusterConfig:       nil,
-		ConnectTimeout:         ptypes.DurationProto(1 * time.Second),
 		LoadAssignment: &xds_endpoint.ClusterLoadAssignment{
 			ClusterName: constants.EnvoyMetricsCluster,
 			Endpoints: []*xds_endpoint.LocalityLbEndpoints{
@@ -261,8 +257,7 @@ func TestGetOriginalDestinationEgressCluster(t *testing.T) {
 			name:        "foo cluster",
 			clusterName: "foo",
 			expected: &xds_cluster.Cluster{
-				Name:           "foo",
-				ConnectTimeout: ptypes.DurationProto(1 * time.Second),
+				Name: "foo",
 				ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 					Type: xds_cluster.Cluster_ORIGINAL_DST,
 				},
@@ -274,8 +269,7 @@ func TestGetOriginalDestinationEgressCluster(t *testing.T) {
 			name:        "bar cluster",
 			clusterName: "bar",
 			expected: &xds_cluster.Cluster{
-				Name:           "bar",
-				ConnectTimeout: ptypes.DurationProto(1 * time.Second),
+				Name: "bar",
 				ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 					Type: xds_cluster.Cluster_ORIGINAL_DST,
 				},
@@ -371,9 +365,8 @@ func TestGetDNSResolvableEgressCluster(t *testing.T) {
 				Port: 80,
 			},
 			expectedCluster: &xds_cluster.Cluster{
-				Name:           "foo.com:80",
-				AltStatName:    "foo_com_80",
-				ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
+				Name:        "foo.com:80",
+				AltStatName: "foo_com_80",
 				ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 					Type: xds_cluster.Cluster_STRICT_DNS,
 				},

pkg/envoy/cds/response_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"testing"
-	"time"
 
 	xds_cluster "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	xds_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
@@ -145,7 +144,6 @@ func TestNewResponse(t *testing.T) {
 		AltStatName:                   "default/bookbuyer|8080|local",
 		ClusterDiscoveryType:          &xds_cluster.Cluster_Type{Type: xds_cluster.Cluster_STRICT_DNS},
 		EdsClusterConfig:              nil,
-		ConnectTimeout:                ptypes.DurationProto(1 * time.Second),
 		RespectDnsTtl:                 true,
 		DnsLookupFamily:               xds_cluster.Cluster_V4_ONLY,
 		TypedExtensionProtocolOptions: HTTP2ProtocolOptions,
@@ -199,7 +197,6 @@ func TestNewResponse(t *testing.T) {
 			},
 			ServiceName: "",
 		},
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
 		TransportSocket: &xds_core.TransportSocket{
 			Name: wellknown.TransportSocketTls,
 			ConfigType: &xds_core.TransportSocket_TypedConfig{
@@ -228,7 +225,6 @@ func TestNewResponse(t *testing.T) {
 			},
 			ServiceName: "",
 		},
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
 		TransportSocket: &xds_core.TransportSocket{
 			Name: wellknown.TransportSocketTls,
 			ConfigType: &xds_core.TransportSocket_TypedConfig{
@@ -308,7 +304,6 @@ func TestNewResponse(t *testing.T) {
 		AltStatName:            constants.EnvoyMetricsCluster,
 		ClusterDiscoveryType:   &xds_cluster.Cluster_Type{Type: xds_cluster.Cluster_STATIC},
 		EdsClusterConfig:       nil,
-		ConnectTimeout:         ptypes.DurationProto(1 * time.Second),
 		LoadAssignment: &xds_endpoint.ClusterLoadAssignment{
 			ClusterName: constants.EnvoyMetricsCluster,
 			Endpoints: []*xds_endpoint.LocalityLbEndpoints{

pkg/envoy/cds/tracing.go

@@ -3,7 +3,6 @@ package cds
 import (
 	xds_cluster "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	xds_endpoint "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
-	"github.com/golang/protobuf/ptypes"
 
 	"github.com/openservicemesh/osm/pkg/configurator"
 	"github.com/openservicemesh/osm/pkg/constants"
@@ -12,9 +11,8 @@ import (
 
 func getTracingCluster(cfg configurator.Configurator) *xds_cluster.Cluster {
 	return &xds_cluster.Cluster{
-		Name:           constants.EnvoyTracingCluster,
-		AltStatName:    constants.EnvoyTracingCluster,
-		ConnectTimeout: ptypes.DurationProto(clusterConnectTimeout),
+		Name:        constants.EnvoyTracingCluster,
+		AltStatName: constants.EnvoyTracingCluster,
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_LOGICAL_DNS,
 		},

pkg/injector/envoy_config_health_probes.go

@@ -5,7 +5,6 @@ import (
 
 	"github.com/envoyproxy/go-control-plane/pkg/wellknown"
 	"github.com/golang/protobuf/ptypes"
-	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/openservicemesh/osm/pkg/constants"
@@ -56,8 +55,7 @@ func getStartupCluster(originalProbe *healthProbe) *xds_cluster.Cluster {
 
 func getProbeCluster(clusterName string, port int32) *xds_cluster.Cluster {
 	return &xds_cluster.Cluster{
-		Name:           clusterName,
-		ConnectTimeout: durationpb.New(time.Second),
+		Name: clusterName,
 		ClusterDiscoveryType: &xds_cluster.Cluster_Type{
 			Type: xds_cluster.Cluster_STATIC,
 		},
@@ -128,7 +126,7 @@ func getProbeListener(listenerName, clusterName, newPath string, port int32, ori
 				RouteConfig: &xds_route.RouteConfiguration{
 					Name: "local_route",
 					VirtualHosts: []*xds_route.VirtualHost{
-						getVirtualHost(newPath, clusterName, originalProbe.path),
+						getVirtualHost(newPath, clusterName, originalProbe.path, originalProbe.timeout),
 					},
 				},
 			},
@@ -204,7 +202,13 @@ func getProbeListener(listenerName, clusterName, newPath string, port int32, ori
 	}, nil
 }
 
-func getVirtualHost(newPath, clusterName, originalProbePath string) *xds_route.VirtualHost {
+func getVirtualHost(newPath, clusterName, originalProbePath string, routeTimeout time.Duration) *xds_route.VirtualHost {
+	if routeTimeout < 1*time.Second {
+		// This should never happen in practice because the minimum value in Kubernetes
+		// is set to 1. However it is easy to check and setting the timeout to 0 will lead
+		// to leaks.
+		routeTimeout = 1 * time.Second
+	}
 	return &xds_route.VirtualHost{
 		Name: "local_service",
 		Domains: []string{
@@ -223,6 +227,7 @@ func getVirtualHost(newPath, clusterName, originalProbePath string) *xds_route.V
 							Cluster: clusterName,
 						},
 						PrefixRewrite: originalProbePath,
+						Timeout:       ptypes.DurationProto(routeTimeout),
 					},
 				},
 			},

pkg/injector/envoy_config_health_probes_test.go

@@ -2,6 +2,7 @@ package injector
 
 import (
 	"testing"
+	"time"
 
 	xds_cluster "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	xds_listener "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
@@ -14,17 +15,21 @@ import (
 
 var _ = ginkgo.Describe("Test functions creating Envoy config and rewriting the Pod's health probes to pass through Envoy", func() {
 
-	liveness := &healthProbe{path: "/liveness", port: 81, isHTTP: true}
-	livenessNonHTTP := &healthProbe{port: 81, isHTTP: false}
-	readiness := &healthProbe{path: "/readiness", port: 82, isHTTP: true}
-	startup := &healthProbe{path: "/startup", port: 83, isHTTP: true}
+	timeout := 42 * time.Second
+	liveness := &healthProbe{path: "/liveness", port: 81, isHTTP: true, timeout: timeout}
+	livenessNonHTTP := &healthProbe{port: 81, isHTTP: false, timeout: timeout}
+	readiness := &healthProbe{path: "/readiness", port: 82, isHTTP: true, timeout: timeout}
+	startup := &healthProbe{path: "/startup", port: 83, isHTTP: true, timeout: timeout}
 
 	// Listed below are the functions we are going to test.
 	// The key in the map is the name of the function -- must match what's in the value of the map.
 	// The key (function name) is used to locate and load the YAML file with the expected return for this function.
 	clusterFunctionsToTest := map[string]func() protoreflect.ProtoMessage{
 		"getVirtualHosts": func() protoreflect.ProtoMessage {
-			return getVirtualHost("/some/path", "-cluster-name-", "/original/probe/path")
+			return getVirtualHost("/some/path", "-cluster-name-", "/original/probe/path", timeout)
+		},
+		"getVirtualHostsDefault": func() protoreflect.ProtoMessage {
+			return getVirtualHost("/some/path", "-cluster-name-", "/original/probe/path", 0*time.Second)
 		},
 		"getProbeCluster":     func() protoreflect.ProtoMessage { return getProbeCluster("cluster-name", 12341234) },
 		"getLivenessCluster":  func() protoreflect.ProtoMessage { return getLivenessCluster(liveness) },

pkg/injector/health_probes.go

@@ -2,6 +2,7 @@ package injector
 
 import (
 	"errors"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -20,8 +21,9 @@ const (
 var errNoMatchingPort = errors.New("no matching port")
 
 type healthProbe struct {
-	path string
-	port int32
+	path    string
+	port    int32
+	timeout time.Duration
 
 	// isHTTP corresponds to an httpGet probe with a scheme of HTTP or undefined.
 	// This helps inform what kind of Envoy config to add to the pod.
@@ -89,6 +91,7 @@ func rewriteProbe(probe *corev1.Probe, probeType, path string, port int32, conta
 		log.Error().Err(err).Msgf("Error finding a matching port for %+v on container %+v", *definedPort, containerPorts)
 	}
 	*definedPort = intstr.IntOrString{Type: intstr.Int, IntVal: port}
+	originalProbe.timeout = time.Duration(probe.TimeoutSeconds) * time.Second
 
 	log.Debug().Msgf(
 		"Rewriting %s probe (:%d%s) to :%d%s",