wip: passive health checks

Signed-off-by: Roman Zavodskikh <roman.zavodskikh@zalando.de>

config/config.go

@@ -283,6 +283,9 @@ type Config struct {
 	OpenPolicyAgentEnvoyMetadata   string        `yaml:"open-policy-agent-envoy-metadata"`
 	OpenPolicyAgentCleanerInterval time.Duration `yaml:"open-policy-agent-cleaner-interval"`
 	OpenPolicyAgentStartupTimeout  time.Duration `yaml:"open-policy-agent-startup-timeout"`
+
+	// Passive Health Checks
+	PassiveHealthCheckParams mapFlags `yaml:"passive-health-check-params"`
 }
 
 const (
@@ -567,6 +570,9 @@ func NewConfig() *Config {
 	flag.Var(cfg.LuaModules, "lua-modules", "comma separated list of lua filter modules. Use <module>.<symbol> to selectively enable module symbols, for example: package,base._G,base.print,json")
 	flag.Var(cfg.LuaSources, "lua-sources", `comma separated list of lua input types for the lua() filter. Valid sources "", "file", "inline", "file,inline" and "none". Use "file" to only allow lua file references in lua filter. Default "" is the same as "file","inline". Use "none" to disable lua filters.`)
 
+	// Passive Health Checks
+	flag.Var(&cfg.PassiveHealthCheckParams, "passive-health-check-params", "sets the parameters for passive health check feature")
+
 	cfg.flags = flag
 	return cfg
 }
@@ -906,6 +912,9 @@ func (c *Config) ToOptions() skipper.Options {
 		OpenPolicyAgentEnvoyMetadata:   c.OpenPolicyAgentEnvoyMetadata,
 		OpenPolicyAgentCleanerInterval: c.OpenPolicyAgentCleanerInterval,
 		OpenPolicyAgentStartupTimeout:  c.OpenPolicyAgentStartupTimeout,
+
+		// Passive Health Checks
+		PassiveHealthCheckParams: c.PassiveHealthCheckParams.values,
 	}
 	for _, rcci := range c.CloneRoute {
 		eskipClone := eskip.NewClone(rcci.Reg, rcci.Repl)

docs/operation/operation.md

@@ -893,6 +893,38 @@ to get the results paginated or getting all of them at the same time.
 curl localhost:9911/routes?offset=200&limit=100
 ```
 
+## Passive health check
+
+Skipper has an option to automatically detect and mitigate faulty backend endpoints, this feature is called
+Passive Health Check(PHC).
+
+PHC works the following way: the entire uptime is divided in chunks of `stats-reset-period`, per every period Skipper calculates
+the total amount of requests and amount of requests for which HTTP round trip had failed per every endpoint. While next period is going on,
+the Skipper takes a look at previous period and if for the given endpoint:
++ the amount of requests in the previous period is more than `total-requests-threshold`
+AND
++ the share of requests with failed HTTP round trip is more than `failed-round-trips-ratio-threshold`
+then Skipper will send reduced (multiplied by `unhealthy-endpoints-requests-discount`, but not less than `total-requests-threshold`)
+amount of requests compared to amount sent without PHC.
+
+Having this, we expect less requests to fail because a lot of them would be sent to endpoints that seem to be healthy instead.
+
+To enable this feature, you need to provide `-passive-health-check-params` having all forementioned parameters
+(`stats-reset-period`, `total-requests-threshold`, `failed-round-trips-ratio-threshold`, `unhealthy-endpoints-requests-discount`) defined,
+for instance: `-passive-health-check-params=stats-reset-period=1m,total-requests-threshold=10,failed-round-trips-ratio-threshold=0.2,unhealthy-endpoints-requests-discount=0.9`.
+
+You need to define all parameters on your side, there are no defaults, and skipper will not run if PHC params are passed only partially.
+
+However, Skipper will run without this feature, if no `-passive-health-check-params` is provided at all.
+
+The parameters of `-passive-health-check-params` are:
++ `stats-reset-period=<duration>` - the duration of stats reset period
++ `total-requests-threshold=<int>` - the minimum number of requests per minute per backend endpoint required to activate PHC for this endpoint
++ `failed-round-trips-ratio-threshold=<float more than 0 and less than 1>` - the minimum ratio of failed requests with failed round trip
+for every endpoint to make PHC avoid sending requests to this endpoint
++ `unhealthy-endpoints-requests-discount=<float more than 0 and less than 1>` - the discount multiplier which defines how huge is reduction of requests
+sent to endpoints considered unhealthy
+
 ## Memory consumption
 
 While Skipper is generally not memory bound, some features may require

metricsinit_test.go

@@ -53,6 +53,12 @@ func TestInitOrderAndDefault(t *testing.T) {
 			SwarmRedisURLs:                []string{fmt.Sprintf("localhost:%d", redisPort)},
 			EnableRatelimiters:            true,
 			SwarmRedisConnMetricsInterval: ringMetricsUpdatePeriod,
+			PassiveHealthCheckParams: map[string]string{
+				"stats-reset-period":                    "1m",
+				"total-requests-threshold":              "10",
+				"failed-round-trips-ratio-threshold":    "0.2",
+				"unhealthy-endpoints-requests-discount": "0.9",
+			},
 		}
 
 		tornDown := make(chan struct{})

proxy/healthy_endpoints.go

@@ -0,0 +1,34 @@
+package proxy
+
+import (
+	"math/rand"
+
+	"github.com/zalando/skipper/routing"
+)
+
+type healthyEndpoints struct {
+	rnd              *rand.Rand
+	endpointRegistry *routing.EndpointRegistry
+}
+
+func (h *healthyEndpoints) filterHealthyEndpoints(endpoints []routing.LBEndpoint, rt *routing.Route) []routing.LBEndpoint {
+	if h == nil {
+		return endpoints
+	}
+
+	p := h.rnd.Float64()
+
+	filtered := make([]routing.LBEndpoint, 0, len(endpoints))
+	for _, e := range endpoints {
+		if p < e.Metrics.HealthCheckDropProbability() {
+			/* drop */
+		} else {
+			filtered = append(filtered, e)
+		}
+	}
+
+	if len(filtered) == 0 {
+		return endpoints
+	}
+	return filtered
+}

proxy/healthy_endpoints_test.go

@@ -0,0 +1,136 @@
+package proxy
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/zalando/skipper/eskip"
+	"github.com/zalando/skipper/loadbalancer"
+	"github.com/zalando/skipper/routing"
+)
+
+func initialize(hosts []string) (*routing.EndpointRegistry, *routing.Route) {
+	endpointRegistry := routing.NewEndpointRegistry(routing.RegistryOptions{
+		PassiveHealthCheckEnabled:          true,
+		TotalRequestsThreshold:             10,
+		FailedRoundTripsRatioThreshold:     0.1,
+		UnhealthyEndpointsRequestsDiscount: 1.0,
+	})
+	eskipRoute := eskip.Route{BackendType: eskip.LBBackend, LBAlgorithm: loadbalancer.PowerOfRandomNChoices.String()}
+	for i := range hosts {
+		eskipRoute.LBEndpoints = append(eskipRoute.LBEndpoints, fmt.Sprintf("http://%s", hosts[i]))
+	}
+
+	route := &routing.Route{
+		Route:       eskipRoute,
+		LBEndpoints: []routing.LBEndpoint{},
+	}
+	rt := loadbalancer.NewAlgorithmProvider().Do([]*routing.Route{route})
+	route = rt[0]
+	endpointRegistry.Do([]*routing.Route{route})
+
+	return endpointRegistry, route
+}
+
+func TestFilterHealthyEndpointsReturnsAllEndpointsWhenAllAreHealthy(t *testing.T) {
+	hosts := []string{"host1.test:80", "host2.test:80", "host3.test:80"}
+	endpointRegistry, route := initialize(hosts)
+
+	for i := range hosts {
+		metrics := endpointRegistry.GetMetrics(hosts[i])
+		for j := 0; j < 50; j++ {
+			metrics.IncTotalRequests()
+		}
+	}
+	endpointRegistry.Do([]*routing.Route{route})
+
+	proxy := &Proxy{
+		registry:         endpointRegistry,
+		heathlyEndpoints: &healthyEndpoints{rnd: rand.New(loadbalancer.NewLockedSource()), endpointRegistry: endpointRegistry},
+	}
+
+	const nRequests = 10_000
+	nHosts := map[int]int{}
+	for i := 0; i < nRequests; i++ {
+		endpoints := proxy.heathlyEndpoints.filterHealthyEndpoints(route.LBEndpoints, route)
+		v := nHosts[len(endpoints)]
+		nHosts[len(endpoints)] = v + 1
+	}
+
+	assert.Equal(t, nRequests, nHosts[3])
+	assert.Equal(t, 0, nHosts[2])
+	assert.Equal(t, 0, nHosts[1])
+}
+
+func TestFilterHealthyEndpointsReturnsAllEndpointsWhenAllAreUnhealthy(t *testing.T) {
+	hosts := []string{"host1.test:80", "host2.test:80", "host3.test:80"}
+	endpointRegistry, route := initialize(hosts)
+
+	for i := range hosts {
+		metrics := endpointRegistry.GetMetrics(hosts[i])
+		for j := 0; j < 50; j++ {
+			metrics.IncTotalRequests()
+		}
+		for j := 0; j < 30; j++ {
+			metrics.IncTotalFailedRoundTrips()
+		}
+	}
+
+	proxy := &Proxy{
+		registry:         endpointRegistry,
+		heathlyEndpoints: &healthyEndpoints{rnd: rand.New(loadbalancer.NewLockedSource()), endpointRegistry: endpointRegistry},
+	}
+
+	const nRequests = 10_000
+	nHosts := map[int]int{}
+	for i := 0; i < nRequests; i++ {
+		endpoints := proxy.heathlyEndpoints.filterHealthyEndpoints(route.LBEndpoints, route)
+		v := nHosts[len(endpoints)]
+		nHosts[len(endpoints)] = v + 1
+	}
+
+	assert.Equal(t, nRequests, nHosts[3])
+	assert.Equal(t, 0, nHosts[2])
+	assert.Equal(t, 0, nHosts[1])
+}
+
+func TestFilterHealthyEndpointsFiltersUnhealthyEndpoints(t *testing.T) {
+	hosts := []string{"host1.test:80", "host2.test:80", "host3.test:80"}
+	endpointRegistry, route := initialize(hosts)
+
+	for i := range hosts {
+		metrics := endpointRegistry.GetMetrics(hosts[i])
+		for j := 0; j < 50; j++ {
+			metrics.IncTotalRequests()
+		}
+	}
+	for i := 0; i < 30; i++ {
+		metrics := endpointRegistry.GetMetrics(hosts[0])
+		metrics.IncTotalFailedRoundTrips()
+	}
+	endpointRegistry.Do([]*routing.Route{route})
+
+	proxy := &Proxy{
+		registry:         endpointRegistry,
+		heathlyEndpoints: &healthyEndpoints{rnd: rand.New(loadbalancer.NewLockedSource()), endpointRegistry: endpointRegistry},
+	}
+
+	const nRequests = 10_000
+	nHosts := map[int]int{}
+	for i := 0; i < nRequests; i++ {
+		endpoints := proxy.heathlyEndpoints.filterHealthyEndpoints(route.LBEndpoints, route)
+		v := nHosts[len(endpoints)]
+		nHosts[len(endpoints)] = v + 1
+
+		if len(endpoints) == 2 {
+			assert.Equal(t, hosts[1], endpoints[0].Host)
+			assert.Equal(t, hosts[2], endpoints[1].Host)
+		}
+	}
+
+	assert.InDelta(t, 6*nRequests/10, nHosts[3], nRequests/10)
+	assert.InDelta(t, 4*nRequests/10, nHosts[2], nRequests/10)
+	assert.Equal(t, 0, nHosts[1])
+}