Method cleanUnhealthy

ring/client/pool.go:261–302

cleanUnhealthy loops through all servers and deletes any that fail a healthcheck. The health checks are executed concurrently with p.cfg.MaxConcurrentHealthChecks.

()

// cleanUnhealthy loops through all servers and deletes any that fail a healthcheck.
// The health checks are executed concurrently with p.cfg.MaxConcurrentHealthChecks.
func (p *Pool) cleanUnhealthy() {
	addresses := p.RegisteredAddresses()
	_ = concurrency.ForEachJob(context.Background(), len(addresses), p.cfg.MaxConcurrentHealthChecks, func(ctx context.Context, idx int) error {
		addr := addresses[idx]
		member, ok := p.fromCache(addr)
		// not ok means someone removed a client between the start of this loop and now
		if !ok {
			return nil
		}

		err := healthCheck(ctx, member.client, p.cfg.HealthCheckTimeout)
		if err == nil {
			member.firstFailedHealthCheck = time.Time{}
			return nil
		}

		if member.firstFailedHealthCheck.IsZero() {
			member.firstFailedHealthCheck = time.Now()
		}

		if time.Since(member.firstFailedHealthCheck) >= p.cfg.HealthCheckGracePeriod {
			level.Warn(p.logger).Log(
				"msg", fmt.Sprintf("removing %s failing healthcheck", p.clientName),
				"addr", addr,
				"reason", err,
				"first_failed_at", member.firstFailedHealthCheck,
			)
			p.RemoveClientFor(addr)
		} else {
			level.Debug(p.logger).Log(
				"msg", fmt.Sprintf("%s failed healthcheck within grace period, not removing", p.clientName),
				"addr", addr,
				"reason", err,
				"first_failed_at", member.firstFailedHealthCheck,
			)
		}

		// Never return an error, because otherwise the processing would stop and
		// remaining health checks would not be executed.
		return nil
	})
}

// healthCheck will check if the client is still healthy, returning an error if it is not
func healthCheck(ctx context.Context, client PoolClient, timeout time.Duration) error {