upstream health check config (#1796)

config/constants.go

@@ -0,0 +1,13 @@
+package config
+
+import "errors"
+
+const (
+	policyKey      = "policy"
+	toKey          = "to"
+	healthCheckKey = "health_check"
+)
+
+var (
+	errKeysMustBeStrings = errors.New("cannot convert nested map: all keys must be strings")
+)

config/custom.go

@@ -6,7 +6,10 @@ import (
 	"fmt"
 	"reflect"
 
+	envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	"github.com/mitchellh/mapstructure"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
 )
 
 // A StringSlice is a slice of strings.
@@ -104,7 +107,7 @@ func DecodeOptionsHookFunc() mapstructure.DecodeHookFunc {
 			return data, nil
 		}
 
-		ps, ok := m["policy"].([]interface{})
+		ps, ok := m[policyKey].([]interface{})
 		if !ok {
 			return data, nil
 		}
@@ -114,7 +117,15 @@ func DecodeOptionsHookFunc() mapstructure.DecodeHookFunc {
 			if !ok {
 				continue
 			}
-			rawTo, ok := pm["to"]
+			raw, ok := pm[healthCheckKey]
+			if ok {
+				hc := new(envoy_config_core_v3.HealthCheck)
+				if err := parseJSONPB(raw, hc); err != nil {
+					return nil, fmt.Errorf("%s: %w", healthCheckKey, err)
+				}
+				pm[healthCheckKey] = hc
+			}
+			rawTo, ok := pm[toKey]
 			if !ok {
 				continue
 			}
@@ -127,9 +138,57 @@ func DecodeOptionsHookFunc() mapstructure.DecodeHookFunc {
 			if err != nil {
 				return nil, err
 			}
-			pm["to"] = slc
+			pm[toKey] = slc
 		}
 
 		return data, nil
 	}
 }
+
+// parseJSONPB takes an intermediate representation and parses it using protobuf parser
+// that correctly handles oneof and other data types
+func parseJSONPB(raw interface{}, dst proto.Message) error {
+	ms, err := serializable(raw)
+	if err != nil {
+		return err
+	}
+
+	data, err := json.Marshal(ms)
+	if err != nil {
+		return err
+	}
+
+	return protojson.Unmarshal(data, dst)
+}
+
+// serializable converts mapstructure nested map into map[string]interface{} that is serializable to JSON
+func serializable(in interface{}) (interface{}, error) {
+	switch typed := in.(type) {
+	case map[interface{}]interface{}:
+		m := make(map[string]interface{})
+		for k, v := range typed {
+			kstr, ok := k.(string)
+			if !ok {
+				return nil, errKeysMustBeStrings
+			}
+			val, err := serializable(v)
+			if err != nil {
+				return nil, err
+			}
+			m[kstr] = val
+		}
+		return m, nil
+	case []interface{}:
+		out := make([]interface{}, 0, len(typed))
+		for _, elem := range typed {
+			val, err := serializable(elem)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, val)
+		}
+		return out, nil
+	default:
+		return in, nil
+	}
+}

config/custom_test.go

@@ -1,10 +1,12 @@
 package config
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v2"
 )
 
@@ -37,3 +39,18 @@ func TestStringSlice_UnmarshalYAML(t *testing.T) {
 		assert.Equal(t, NewStringSlice("a", "b", "c"), slc)
 	})
 }
+func TestSerializable(t *testing.T) {
+	data, err := base64.StdEncoding.DecodeString("aGVhbHRoX2NoZWNrOgogIHRpbWVvdXQ6IDVzCiAgaW50ZXJ2YWw6IDYwcwogIGhlYWx0aHlUaHJlc2hvbGQ6IDEKICB1bmhlYWx0aHlUaHJlc2hvbGQ6IDIKICBodHRwX2hlYWx0aF9jaGVjazogCiAgICBob3N0OiAiaHR0cDovL2xvY2FsaG9zdDo4MDgwIgogICAgcGF0aDogIi8iCg==")
+	require.NoError(t, err, "decode")
+
+	var mi map[interface{}]interface{}
+
+	err = yaml.Unmarshal(data, &mi)
+	require.NoError(t, err, "unmarshal")
+
+	ms, err := serializable(mi)
+	require.NoError(t, err, "serializable")
+
+	_, err = json.Marshal(ms)
+	require.NoError(t, err, "json marshal")
+}

config/options.go

@@ -399,7 +399,8 @@ func (o *Options) parsePolicy() error {
 	}
 	// Finish initializing policies
 	for i := range o.Policies {
-		if err := (&o.Policies[i]).Validate(); err != nil {
+		p := &o.Policies[i]
+		if err := p.Validate(); err != nil {
 			return err
 		}
 	}

config/policy.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	"github.com/golang/protobuf/ptypes"
 
 	"github.com/pomerium/pomerium/internal/hashutil"
@@ -141,6 +142,9 @@ type Policy struct {
 	// OutlierDetection configures outlier detection for the upstream cluster.
 	OutlierDetection *PolicyOutlierDetection `mapstructure:"outlier_detection" yaml:"outlier_detection,omitempty" json:"outlier_detection,omitempty"`
 
+	// HealthCheck defines active health checks. See https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/core/v3/health_check.proto
+	HealthCheck *envoy_config_core_v3.HealthCheck `mapstructure:"health_check" yaml:"health_check,omitempty" json:"health_check,omitempty"`
+
 	SubPolicies []SubPolicy `mapstructure:"sub_policies" yaml:"sub_policies,omitempty" json:"sub_policies,omitempty"`
 }
 
@@ -429,6 +433,12 @@ func (p *Policy) Validate() error {
 		return fmt.Errorf("config: only prefix_rewrite or regex_rewrite_pattern can be specified, but not both")
 	}
 
+	if p.HealthCheck != nil {
+		if err := p.HealthCheck.Validate(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 

docs/reference/readme.md

@@ -1370,6 +1370,15 @@ When enabled, this option will pass identity headers to upstream applications. T
 If set, enables proxying of SPDY protocol upgrades.
 
 
+### Health Check
+- Config File Key: `health_check`
+- Type: `object`
+- Optional
+
+When defined, will issue periodic health check requests to upstream servers.
+See [Envoy documentation](https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/upstream/health_checking) for a list of supported parameters.
+
+
 ### Websocket Connections
 - Config File Key: `allow_websockets`
 - Type: `bool`

docs/reference/settings.yaml

@@ -1502,6 +1502,15 @@ settings:
           - Default: `false`
         doc: |
           If set, enables proxying of SPDY protocol upgrades.
+      - name: "Health Check"
+        keys: ["health_check"]
+        attributes: |
+          - Config File Key: `health_check`
+          - Type: `object`
+          - Optional
+        doc: |
+          When defined, will issue periodic health check requests to upstream servers.
+          See [Envoy documentation](https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/upstream/health_checking) for a list of supported parameters.
       - name: "Websocket Connections"
         keys: ["allow_websockets"]
         attributes: |

internal/controlplane/xds_cluster_test.go

@@ -214,7 +214,7 @@ func Test_buildCluster(t *testing.T) {
 		})
 		cluster := buildCluster("example", endpoints, true,
 			config.GetEnvoyDNSLookupFamily(config.DNSLookupFamilyV4Only),
-			nil)
+			nil, nil)
 		testutil.AssertProtoJSONEqual(t, `
 			{
 				"name": "example",
@@ -253,7 +253,7 @@ func Test_buildCluster(t *testing.T) {
 		})
 		cluster := buildCluster("example", endpoints, true,
 			config.GetEnvoyDNSLookupFamily(config.DNSLookupFamilyAuto),
-			nil)
+			nil, nil)
 		testutil.AssertProtoJSONEqual(t, `
 			{
 				"name": "example",
@@ -344,7 +344,7 @@ func Test_buildCluster(t *testing.T) {
 		})
 		cluster := buildCluster("example", endpoints, true,
 			config.GetEnvoyDNSLookupFamily(config.DNSLookupFamilyAuto),
-			nil)
+			nil, nil)
 		testutil.AssertProtoJSONEqual(t, `
 			{
 				"name": "example",
@@ -379,7 +379,7 @@ func Test_buildCluster(t *testing.T) {
 		})
 		cluster := buildCluster("example", endpoints, true,
 			config.GetEnvoyDNSLookupFamily(config.DNSLookupFamilyAuto),
-			nil)
+			nil, nil)
 		testutil.AssertProtoJSONEqual(t, `
 			{
 				"name": "example",
@@ -417,7 +417,7 @@ func Test_buildCluster(t *testing.T) {
 			&envoy_config_cluster_v3.OutlierDetection{
 				EnforcingConsecutive_5Xx:       wrapperspb.UInt32(17),
 				SplitExternalLocalOriginErrors: true,
-			})
+			}, nil)
 		testutil.AssertProtoJSONEqual(t, `
 			{
 				"name": "example",

internal/controlplane/xds_clusters.go

@@ -78,7 +78,7 @@ func (srv *Server) buildClusters(options *config.Options) []*envoy_config_cluste
 func (srv *Server) buildInternalCluster(options *config.Options, name string, dst *url.URL, forceHTTP2 bool) *envoy_config_cluster_v3.Cluster {
 	endpoints := []Endpoint{NewEndpoint(dst, srv.buildInternalTransportSocket(options, dst))}
 	dnsLookupFamily := config.GetEnvoyDNSLookupFamily(options.DNSLookupFamily)
-	return buildCluster(name, endpoints, forceHTTP2, dnsLookupFamily, nil)
+	return buildCluster(name, endpoints, forceHTTP2, dnsLookupFamily, nil, nil)
 }
 
 func (srv *Server) buildPolicyCluster(options *config.Options, policy *config.Policy) *envoy_config_cluster_v3.Cluster {
@@ -89,7 +89,7 @@ func (srv *Server) buildPolicyCluster(options *config.Options, policy *config.Po
 		dnsLookupFamily = envoy_config_cluster_v3.Cluster_V4_ONLY
 	}
 	outlierDetection := (*envoy_config_cluster_v3.OutlierDetection)(policy.OutlierDetection)
-	return buildCluster(name, endpoints, false, dnsLookupFamily, outlierDetection)
+	return buildCluster(name, endpoints, false, dnsLookupFamily, outlierDetection, policy.HealthCheck)
 }
 
 func (srv *Server) buildPolicyEndpoints(policy *config.Policy) []Endpoint {
@@ -235,6 +235,7 @@ func buildCluster(
 	forceHTTP2 bool,
 	dnsLookupFamily envoy_config_cluster_v3.Cluster_DnsLookupFamily,
 	outlierDetection *envoy_config_cluster_v3.OutlierDetection,
+	healthCheck *envoy_config_core_v3.HealthCheck,
 ) *envoy_config_cluster_v3.Cluster {
 	if len(endpoints) == 0 {
 		return nil
@@ -256,6 +257,10 @@ func buildCluster(
 		OutlierDetection:       outlierDetection,
 	}
 
+	if healthCheck != nil {
+		cluster.HealthChecks = append(cluster.HealthChecks, healthCheck)
+	}
+
 	if forceHTTP2 {
 		cluster.Http2ProtocolOptions = &envoy_config_core_v3.Http2ProtocolOptions{
 			AllowConnect: true,