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

Fixes #4137

* Remove connect timeouts from all clusters
* For health probe clusters we set the idle_timeout to be equal to the
timeout set on the kubernetes pod.

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",