redis: optional health probe that reflects whether Redis answers

README.md

@@ -954,6 +954,8 @@ Some components may be turned OFF via configurations so overall health is not ef
   - Turns unhealthy if receives sigterm signal
     All components needs to be healthy for overall health to be healthy.
 
+The overall health is served on `/healthcheck` and the gRPC health service. Readiness is served separately on `/ready`: it is healthy when the overall health is, and, when the [Redis health probe](#health-checking-for-redis-active-connection) is enabled, while Redis answers. Use `/healthcheck` for liveness probes and `/ready` for readiness probes.
+
 ### Health-check configurations
 
 Health check can be configured to check if rate-limit configurations are loaded using the following environment variable.
@@ -1282,6 +1284,7 @@ mappings: # Requires statsd exporter >= v0.6.0 since it uses the "drop" action.
 The ratelimit service listens to HTTP 1.1 (by default on port 8080) with two endpoints:
 
 1. /healthcheck → return a 200 if this service is healthy
+1. /ready → return a 200 if this service is ready to serve: healthy, and, when the [Redis health probe](#health-checking-for-redis-active-connection) is enabled, able to reach Redis
 1. /json → HTTP 1.1 endpoint for interacting with ratelimit service
 
 ## /json endpoint
@@ -1468,6 +1471,13 @@ To configure whether to return health check failure if there is no active redis
 
 1. `REDIS_HEALTH_CHECK_ACTIVE_CONNECTION` : (default is "false")
 
+`REDIS_HEALTH_CHECK_ACTIVE_CONNECTION` only fails once every connection to Redis has been closed. A Redis that stops answering without closing its connections keeps it healthy. To make the health check reflect whether Redis actually answers, enable the health probe:
+
+1. `REDIS_HEALTH_CHECK_PING_INTERVAL`: PING each Redis (main and per-second) at this interval through the same path as rate limit calls, with the I/O timeout capped to the interval. The `/ready` endpoint fails after `REDIS_HEALTH_CHECK_FAILURE_THRESHOLD` consecutive failed PINGs and recovers after one success. `0` disables it. Default: `0`
+1. `REDIS_HEALTH_CHECK_FAILURE_THRESHOLD`: consecutive failed PINGs before `/ready` fails. Default: `3`
+
+The probe only affects `/ready`, never `/healthcheck` or the gRPC health service. Point readiness probes at `/ready` and keep liveness probes on `/healthcheck`: a Redis outage affects every replica at once, and restarting them cannot fix it, only delay recovery.
+
 ## Recovering from a failover (READONLY errors)
 
 1. `REDIS_CLOSE_CONNECTION_ON_READONLY_ERROR` : (default is "false")

src/redis/cache_impl.go

@@ -31,6 +31,10 @@ func NewRateLimiterCacheImplFromSettings(ctx context.Context, s settings.Setting
 				maxInflight:    s.RedisPerSecondMaxInflight,
 				tcpKeepAlive:   s.RedisPerSecondTcpKeepAlive,
 				tcpUserTimeout: s.RedisPerSecondTcpUserTimeout,
+
+				healthCheckInterval:         s.RedisHealthCheckPingInterval,
+				healthCheckFailureThreshold: s.RedisHealthCheckFailureThreshold,
+				healthComponent:             server.RedisPerSecondPingHealthComponentName,
 			})
 		closer.Closers = append(closer.Closers, perSecondPool)
 	}
@@ -47,6 +51,10 @@ func NewRateLimiterCacheImplFromSettings(ctx context.Context, s settings.Setting
 			maxInflight:    s.RedisMaxInflight,
 			tcpKeepAlive:   s.RedisTcpKeepAlive,
 			tcpUserTimeout: s.RedisTcpUserTimeout,
+
+			healthCheckInterval:         s.RedisHealthCheckPingInterval,
+			healthCheckFailureThreshold: s.RedisHealthCheckFailureThreshold,
+			healthComponent:             server.RedisPingHealthComponentName,
 		})
 	closer.Closers = append(closer.Closers, otherPool)
 

src/redis/driver_impl.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -128,6 +129,9 @@ type clientImpl struct {
 	failWhenNoConnection bool
 	// ioTimeout is the I/O timeout of the connections, see clientOptions.
 	ioTimeout time.Duration
+	// stopProbe and probeDone stop the health probe, if any.
+	stopProbe, probeDone chan struct{}
+	stopOnce             sync.Once
 }
 
 func checkError(err error) {
@@ -220,6 +224,12 @@ type clientOptions struct {
 	tcpKeepAlive time.Duration
 	// tcpUserTimeout sets TCP_USER_TIMEOUT (Linux only). 0 leaves the OS default.
 	tcpUserTimeout time.Duration
+	// healthCheckInterval enables a health probe that PINGs Redis at this
+	// interval and fails healthComponent after healthCheckFailureThreshold
+	// consecutive failures. 0 disables it.
+	healthCheckInterval         time.Duration
+	healthCheckFailureThreshold int
+	healthComponent             string
 }
 
 func newClientImpl(ctx context.Context, scope stats.Scope, useTls bool, auth, redisSocketType, redisType, url string, poolSize int,
@@ -427,6 +437,20 @@ func newClientImpl(ctx context.Context, scope stats.Scope, useTls bool, auth, re
 		logger.Warnf("Redis pool %s: at most %d commands in flight", maskedUrl, opts.maxInflight)
 		ret.inflight = make(chan struct{}, opts.maxInflight)
 	}
+	if opts.healthCheckInterval > 0 && srv != nil {
+		component := opts.healthComponent
+		if component == "" {
+			component = server.RedisPingHealthComponentName
+		}
+		// PINGs time out like any other command; startHealthProbe caps it to
+		// the interval.
+		probeTimeout := opts.ioTimeout
+		if probeTimeout <= 0 {
+			probeTimeout = timeout
+		}
+		logger.Warnf("Redis pool %s: health probe every %s, unhealthy after %d consecutive failures", maskedUrl, opts.healthCheckInterval, opts.healthCheckFailureThreshold)
+		ret.startHealthProbe(srv.HealthChecker(), component, opts.healthCheckInterval, probeTimeout, opts.healthCheckFailureThreshold)
+	}
 	return ret
 }
 
@@ -443,6 +467,10 @@ func (c *clientImpl) DoCmd(rcv interface{}, cmd, key string, args ...interface{}
 }
 
 func (c *clientImpl) Close() error {
+	if c.stopProbe != nil {
+		c.stopOnce.Do(func() { close(c.stopProbe) })
+		<-c.probeDone
+	}
 	return c.client.Close()
 }
 

src/redis/e2e_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"path/filepath"
 	"sync"
@@ -41,8 +42,9 @@ descriptors:
 `
 
 type e2eService struct {
-	client pb.RateLimitServiceClient
-	store  gostats.Store
+	settings settings.Settings
+	client   pb.RateLimitServiceClient
+	store    gostats.Store
 }
 
 func freePort(t *testing.T) int {
@@ -110,7 +112,7 @@ func startService(t *testing.T, redisSrv, perSecondSrv *redis.FakeRedisForTest,
 	require.NoError(t, err)
 	t.Cleanup(func() { conn.Close() })
 
-	svc := &e2eService{client: pb.NewRateLimitServiceClient(conn), store: r.GetStatsStore()}
+	svc := &e2eService{settings: s, client: pb.NewRateLimitServiceClient(conn), store: r.GetStatsStore()}
 	require.Eventually(t, func() bool { return svc.call("minute", time.Second) == nil },
 		10*time.Second, 50*time.Millisecond, "service did not start")
 	return svc
@@ -154,6 +156,19 @@ func (s *e2eService) counter(name string) uint64 {
 	return s.store.NewCounter(name).Value()
 }
 
+func (s *e2eService) health() int { return s.get("/healthcheck") }
+
+func (s *e2eService) ready() int { return s.get("/ready") }
+
+func (s *e2eService) get(path string) int {
+	resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%d%s", s.settings.Port, path))
+	if err != nil {
+		return 0
+	}
+	resp.Body.Close()
+	return resp.StatusCode
+}
+
 func TestServiceFailsFastWhenRedisStopsAnswering(t *testing.T) {
 	ioTimeout := 300 * time.Millisecond
 	// Generous client deadline: the server must fail on its own, well before it.
@@ -264,3 +279,48 @@ func TestServiceReportsRedisOutage(t *testing.T) {
 	assert.Eventually(t, func() bool { return svc.counter("ratelimit.redis_pool.cx_local_close_error") > 0 },
 		2*time.Second, 20*time.Millisecond)
 }
+
+func TestServiceHealthReflectsUnresponsiveRedis(t *testing.T) {
+	for _, perSecond := range []bool{false, true} {
+		t.Run(fmt.Sprintf("perSecond=%t", perSecond), func(t *testing.T) {
+			main := redis.NewFakeRedisForTest(t)
+			failing := main
+			var perSecondSrv *redis.FakeRedisForTest
+			if perSecond {
+				failing = redis.NewFakeRedisForTest(t)
+				perSecondSrv = &failing
+			}
+			svc := startService(t, &main, perSecondSrv, func(s *settings.Settings) {
+				s.RedisTimeout = 200 * time.Millisecond
+				s.RedisPerSecondTimeout = 200 * time.Millisecond
+				s.RedisHealthCheckPingInterval = 50 * time.Millisecond
+				s.RedisHealthCheckFailureThreshold = 2
+			})
+			require.Eventually(t, func() bool { return svc.ready() == http.StatusOK }, 5*time.Second, 20*time.Millisecond)
+
+			// Redis keeps its connections open but stops answering: the
+			// replica is not ready, but its health check, used for liveness,
+			// must not fail and get it restarted.
+			failing.SetBlackhole(true)
+			assert.Eventually(t, func() bool { return svc.ready() == http.StatusInternalServerError },
+				5*time.Second, 20*time.Millisecond)
+			assert.Equal(t, http.StatusOK, svc.health())
+
+			failing.SetBlackhole(false)
+			assert.Eventually(t, func() bool { return svc.ready() == http.StatusOK },
+				10*time.Second, 20*time.Millisecond)
+			assert.Equal(t, http.StatusOK, svc.health())
+		})
+	}
+}
+
+func TestServiceHealthProbeDisabledByDefault(t *testing.T) {
+	srv := redis.NewFakeRedisForTest(t)
+	svc := startService(t, &srv, nil, func(s *settings.Settings) { s.RedisTimeout = 200 * time.Millisecond })
+	require.Eventually(t, func() bool { return svc.ready() == http.StatusOK }, 5*time.Second, 20*time.Millisecond)
+
+	srv.SetBlackhole(true)
+	time.Sleep(time.Second)
+	assert.Equal(t, http.StatusOK, svc.ready())
+	assert.Equal(t, http.StatusOK, svc.health())
+}

src/redis/health_probe.go

@@ -0,0 +1,102 @@
+package redis
+
+import (
+	"context"
+	"time"
+
+	"github.com/mediocregopher/radix/v4"
+	logger "github.com/sirupsen/logrus"
+
+	"github.com/envoyproxy/ratelimit/src/server"
+)
+
+// healthProbe periodically PINGs Redis and marks a readiness component
+// unhealthy after failureThreshold consecutive failures, and healthy again
+// after one success. It only affects readiness (/ready), not the health check
+// (/healthcheck): restarting ratelimit cannot fix Redis, so it must not fail
+// liveness probes.
+//
+// Health otherwise only tracks whether the pool has connections at all
+// (REDIS_HEALTH_CHECK_ACTIVE_CONNECTION), which stays true while every
+// connection is stuck waiting on a server that stopped answering. The probe
+// goes through the same client path as rate limit calls, so it fails whenever
+// they would.
+type healthProbe struct {
+	client           *clientImpl
+	health           *server.HealthChecker
+	component        string
+	interval         time.Duration
+	timeout          time.Duration
+	failureThreshold int
+	failures         int
+}
+
+func (c *clientImpl) startHealthProbe(health *server.HealthChecker, component string, interval, timeout time.Duration, failureThreshold int) {
+	if failureThreshold < 1 {
+		failureThreshold = 1
+	}
+	if timeout <= 0 || timeout > interval {
+		timeout = interval
+	}
+	p := &healthProbe{
+		client:           c,
+		health:           health,
+		component:        component,
+		interval:         interval,
+		timeout:          timeout,
+		failureThreshold: failureThreshold,
+	}
+	// The client connected at startup, so start healthy.
+	health.AddReadinessComponent(component, true)
+
+	c.stopProbe = make(chan struct{})
+	c.probeDone = make(chan struct{})
+	go p.run(c.stopProbe, c.probeDone)
+}
+
+func (p *healthProbe) run(stop <-chan struct{}, done chan<- struct{}) {
+	defer close(done)
+	ticker := time.NewTicker(p.interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-stop:
+			return
+		case <-ticker.C:
+			p.probe()
+		}
+	}
+}
+
+func (p *healthProbe) probe() {
+	ctx, cancel := context.WithTimeout(context.Background(), p.timeout)
+	defer cancel()
+
+	var pong string
+	err := p.client.do(ctx, func(ctx context.Context) error {
+		return p.client.client.Do(ctx, radix.Cmd(&pong, "PING"))
+	}, nil)
+	if err == nil && pong != "PONG" {
+		err = RedisError("unexpected PING reply: " + pong)
+	}
+
+	if err == nil {
+		if p.failures >= p.failureThreshold {
+			logger.Warnf("Redis health probe %s: PING succeeded, marking healthy", p.component)
+		}
+		p.failures = 0
+		if err := p.health.Ok(p.component); err != nil {
+			logger.Errorf("Unable to update health status: %s", err)
+		}
+		return
+	}
+
+	p.failures++
+	logger.Warnf("Redis health probe %s: PING failed (%d consecutive): %v", p.component, p.failures, err)
+	if p.failures == p.failureThreshold {
+		logger.Errorf("Redis health probe %s: %d consecutive PING failures, marking unhealthy", p.component, p.failures)
+		if err := p.health.Fail(p.component); err != nil {
+			logger.Errorf("Unable to update health status: %s", err)
+		}
+	}
+}