reverseproxy: isolate active health-check state per distinct check config (#7916)
* reverseproxy: isolate active health-check state per distinct check config Multiple reverse_proxy handlers configured with different active health checks (health_uri, health_headers, ...) against the same upstream dial address currently share a single Host in the global pool, so one handler's failing probes mark the address unhealthy for every other handler. Key the pool by dial address plus a stable fingerprint of the active health-check config, so distinct checks get independent health state. The fingerprint is strictly internal to pool identity: the Prometheus upstreams_healthy label and the /reverse_proxy/upstreams admin endpoint continue to report the plain dial address, unchanged. Dynamic upstreams are intentionally out of scope here: they resolve through a separate per-lookup path (dynamicHosts) and collapsing there has different lifetime semantics; noted for a follow-up. Fixes #7870 * reverseproxy: use strings.Cut in hostKeyAddress Satisfies the modernize linter; behaviour is unchanged, since Cut returns the whole string when the separator is absent. * reverseproxy: expose the health-check fingerprint as a public discriminator Health state is now kept per (dial address, active health check config), but both user-visible surfaces still reported address alone: - caddy_reverse_proxy_upstreams_healthy was labeled only by upstream, so every handler sharing an address wrote the same series concurrently and the reported value was whichever updater ran last. The metric gains a health_check label carrying the config fingerprint ("" when no active checks), so each health target owns its series; aggregate across checks with sum/min by (upstream). - /reverse_proxy/upstreams reported one entry per pool key but with only the plain address, so consumers indexing by address silently discarded all but one entry. Entries now carry health_check (omitted when empty), and the endpoint documents that (address, health_check) is the entry's identity — one entry per health target, deliberately not aggregated, since any aggregation here would be lossy and undocumented. Tests: two handlers on one address with different checks must produce two metric series reflecting their own state (fails if the fingerprint is dropped from the label), and two admin entries distinguished by non-empty fingerprints. * reverseproxy: narrow the fix to per-Upstream active health counters Move the consecutive active pass/fail counters from Host onto Upstream, alongside the active unhealthy state that already lives there, instead of re-keying the global host pool. Host is keyed by dial address alone, but an active health check is configured per handler, so two handlers dialing the same address with different health_uri or health_headers share those counters and can push each other over their own thresholds. Upstream is already per-handler and already carries the active unhealthy flag, so the counters belong next to it and the pool keeps its plain dial-address keys. This drops the host key fingerprint and its exposure in the metric label and the admin upstreams endpoint; the metric series identity is left for separate consideration. --------- Co-authored-by: SillyZir <269283839+SillyZir@users.noreply.github.com> Co-authored-by: Zen Dodd <mail@steadytao.com>pull/7888/head^2
SillyZir
GitHub
parent
1fed032234
commit
6584c75999
|
|
@ -0,0 +1,151 @@
|
|||
// Copyright 2015 Matthew Holt and The Caddy Authors
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package reverseproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/caddyserver/caddy/v2"
|
||||
"github.com/caddyserver/caddy/v2/modules/caddyevents"
|
||||
)
|
||||
|
||||
// newActiveHandler builds a minimal Handler with active health checks
|
||||
// configured against addr, provisions its single upstream, and returns
|
||||
// the handler, its upstream, and a cancel func the caller must defer.
|
||||
func newActiveHandler(t *testing.T, addr, uri string, fails int) (*Handler, *Upstream, context.CancelFunc) {
|
||||
t.Helper()
|
||||
caddyCtx, cancel := caddy.NewContext(caddy.Context{Context: context.Background()})
|
||||
|
||||
eventsApp := new(caddyevents.App)
|
||||
if err := eventsApp.Provision(caddyCtx); err != nil {
|
||||
t.Fatalf("provisioning events app: %v", err)
|
||||
}
|
||||
|
||||
u := &Upstream{Dial: addr}
|
||||
h := &Handler{
|
||||
ctx: caddyCtx,
|
||||
logger: zap.NewNop(),
|
||||
events: eventsApp,
|
||||
Upstreams: UpstreamPool{u},
|
||||
HealthChecks: &HealthChecks{
|
||||
Active: &ActiveHealthChecks{
|
||||
URI: uri,
|
||||
Fails: fails,
|
||||
},
|
||||
},
|
||||
}
|
||||
h.provisionUpstream(u, false)
|
||||
if err := h.HealthChecks.Active.Provision(caddyCtx, h); err != nil {
|
||||
t.Fatalf("provisioning active health checks: %v", err)
|
||||
}
|
||||
return h, u, cancel
|
||||
}
|
||||
|
||||
// runActiveHealthCheck synchronously performs one active health check round
|
||||
// for the handler's upstream (what doActiveHealthCheckForAllHosts does per
|
||||
// tick, minus the goroutine).
|
||||
func runActiveHealthCheck(t *testing.T, h *Handler, u *Upstream) {
|
||||
t.Helper()
|
||||
dialInfo, err := u.fillDialInfo(caddy.NewReplacer())
|
||||
if err != nil {
|
||||
t.Fatalf("filling dial info: %v", err)
|
||||
}
|
||||
if err := h.doActiveHealthCheck(dialInfo, dialInfo.Address, u.Dial, u); err != nil {
|
||||
t.Fatalf("active health check: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// drainHostsPool removes every entry from the global static host pool,
|
||||
// deleting each exactly as many times as it was stored so the pool is
|
||||
// empty for subsequent tests.
|
||||
func drainHostsPool() {
|
||||
var keys []any
|
||||
hosts.Range(func(key, _ any) bool {
|
||||
keys = append(keys, key)
|
||||
return true
|
||||
})
|
||||
for _, key := range keys {
|
||||
if refs, ok := hosts.References(key); ok {
|
||||
for range refs {
|
||||
_, _ = hosts.Delete(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestActiveHealthChecksSameAddressDifferentChecksAreIndependent is a
|
||||
// regression test for https://github.com/caddyserver/caddy/issues/7870:
|
||||
// two handlers that dial the same upstream address but run different
|
||||
// active health checks (different health_uri) must keep independent
|
||||
// health state. Before the fix, the consecutive pass/fail counters lived
|
||||
// on the Host, which is shared by dial address, so both checkers mutated
|
||||
// the same counters: one vhost's failing probes could push the other
|
||||
// vhost's upstream over its own fails threshold and knock it out of that
|
||||
// vhost's data path.
|
||||
func TestActiveHealthChecksSameAddressDifferentChecksAreIndependent(t *testing.T) {
|
||||
resetDynamicHosts()
|
||||
defer drainHostsPool()
|
||||
|
||||
// one backend node serving two vhosts: vhost A's health endpoint is
|
||||
// down, vhost B's is up (except for one transient failure below)
|
||||
var vhostBUp atomic.Bool
|
||||
vhostBUp.Store(true)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/vhost-a/health":
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
case "/vhost-b/health":
|
||||
if vhostBUp.Load() {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
} else {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}
|
||||
default:
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
addr := strings.TrimPrefix(srv.URL, "http://")
|
||||
|
||||
hA, uA, cancelA := newActiveHandler(t, addr, "/vhost-a/health", 5)
|
||||
defer cancelA()
|
||||
hB, uB, cancelB := newActiveHandler(t, addr, "/vhost-b/health", 3)
|
||||
defer cancelB()
|
||||
|
||||
// vhost A's checker observes two consecutive failures — below its own
|
||||
// fails=5 threshold, so these must affect no one's health status
|
||||
runActiveHealthCheck(t, hA, uA)
|
||||
runActiveHealthCheck(t, hA, uA)
|
||||
|
||||
// vhost B's checker observes a single transient failure; B tolerates
|
||||
// up to fails=3 consecutive failures, so it must remain healthy
|
||||
vhostBUp.Store(false)
|
||||
runActiveHealthCheck(t, hB, uB)
|
||||
|
||||
if !uA.Healthy() {
|
||||
t.Errorf("vhost A's upstream should still be healthy after 2 of 5 tolerated failures")
|
||||
}
|
||||
if !uB.Healthy() {
|
||||
t.Errorf("vhost B's upstream was marked unhealthy after a single failed probe (fails=3); " +
|
||||
"its health state was polluted by vhost A's health check against the same address")
|
||||
}
|
||||
}
|
||||
|
|
@ -461,7 +461,7 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, networ
|
|||
|
||||
markUnhealthy := func() {
|
||||
// increment failures and then check if it has reached the threshold to mark unhealthy
|
||||
err := upstream.Host.countHealthFail(1)
|
||||
err := upstream.countHealthFail(1)
|
||||
if err != nil {
|
||||
if c := h.HealthChecks.Active.logger.Check(zapcore.ErrorLevel, "could not count active health failure"); c != nil {
|
||||
c.Write(
|
||||
|
|
@ -471,11 +471,11 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, networ
|
|||
}
|
||||
return
|
||||
}
|
||||
if upstream.Host.activeHealthFails() >= h.HealthChecks.Active.Fails {
|
||||
if upstream.activeHealthFails() >= h.HealthChecks.Active.Fails {
|
||||
// dispatch an event that the host newly became unhealthy
|
||||
if upstream.setHealthy(false) {
|
||||
h.events.Emit(h.ctx, "unhealthy", map[string]any{"host": hostAddr})
|
||||
upstream.Host.resetHealth()
|
||||
upstream.resetHealth()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -492,13 +492,13 @@ func (h *Handler) doActiveHealthCheck(dialInfo DialInfo, hostAddr string, networ
|
|||
}
|
||||
return
|
||||
}
|
||||
if upstream.Host.activeHealthPasses() >= h.HealthChecks.Active.Passes {
|
||||
if upstream.activeHealthPasses() >= h.HealthChecks.Active.Passes {
|
||||
if upstream.setHealthy(true) {
|
||||
if c := h.HealthChecks.Active.logger.Check(zapcore.InfoLevel, "host is up"); c != nil {
|
||||
c.Write(zap.String("host", hostAddr))
|
||||
}
|
||||
h.events.Emit(h.ctx, "healthy", map[string]any{"host": hostAddr})
|
||||
upstream.Host.resetHealth()
|
||||
upstream.resetHealth()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -61,7 +61,16 @@ type Upstream struct {
|
|||
activeHealthCheckUpstream string
|
||||
healthCheckPolicy *PassiveHealthChecks
|
||||
cb CircuitBreaker
|
||||
unhealthy atomic.Int32 // status from active health checker
|
||||
|
||||
// state from the active health checker. It lives here rather than on
|
||||
// the shared Host because the Host is keyed by dial address alone,
|
||||
// while an active health check is configured per handler: two handlers
|
||||
// dialing the same address with different health_uri or health_headers
|
||||
// are checking distinct health targets, and must not push each other
|
||||
// over their own consecutive pass/fail thresholds.
|
||||
unhealthy atomic.Int32
|
||||
activePasses atomic.Int64
|
||||
activeFails atomic.Int64
|
||||
}
|
||||
|
||||
// (pointer receiver necessary to avoid a race condition, since
|
||||
|
|
@ -173,10 +182,8 @@ func (u *Upstream) fillDynamicHost() {
|
|||
// Host is the basic, in-memory representation of the state of a remote host.
|
||||
// Its fields are accessed atomically and Host values must not be copied.
|
||||
type Host struct {
|
||||
numRequests atomic.Int64
|
||||
fails atomic.Int64
|
||||
activePasses atomic.Int64
|
||||
activeFails atomic.Int64
|
||||
numRequests atomic.Int64
|
||||
fails atomic.Int64
|
||||
}
|
||||
|
||||
// NumRequests returns the number of active requests to the upstream.
|
||||
|
|
@ -189,16 +196,6 @@ func (h *Host) Fails() int {
|
|||
return int(h.fails.Load())
|
||||
}
|
||||
|
||||
// activeHealthPasses returns the number of consecutive active health check passes with the upstream.
|
||||
func (h *Host) activeHealthPasses() int {
|
||||
return int(h.activePasses.Load())
|
||||
}
|
||||
|
||||
// activeHealthFails returns the number of consecutive active health check failures with the upstream.
|
||||
func (h *Host) activeHealthFails() int {
|
||||
return int(h.activeFails.Load())
|
||||
}
|
||||
|
||||
// countRequest mutates the active request count by
|
||||
// delta. It returns an error if the adjustment fails.
|
||||
func (h *Host) countRequest(delta int) error {
|
||||
|
|
@ -219,10 +216,22 @@ func (h *Host) countFail(delta int) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// activeHealthPasses returns the number of consecutive passing
|
||||
// active health checks observed by this upstream's checker.
|
||||
func (u *Upstream) activeHealthPasses() int {
|
||||
return int(u.activePasses.Load())
|
||||
}
|
||||
|
||||
// activeHealthFails returns the number of consecutive failing
|
||||
// active health checks observed by this upstream's checker.
|
||||
func (u *Upstream) activeHealthFails() int {
|
||||
return int(u.activeFails.Load())
|
||||
}
|
||||
|
||||
// countHealthPass mutates the recent passes count by
|
||||
// delta. It returns an error if the adjustment fails.
|
||||
func (h *Host) countHealthPass(delta int) error {
|
||||
result := h.activePasses.Add(int64(delta))
|
||||
func (u *Upstream) countHealthPass(delta int) error {
|
||||
result := u.activePasses.Add(int64(delta))
|
||||
if result < 0 {
|
||||
return fmt.Errorf("count below 0: %d", result)
|
||||
}
|
||||
|
|
@ -231,8 +240,8 @@ func (h *Host) countHealthPass(delta int) error {
|
|||
|
||||
// countHealthFail mutates the recent failures count by
|
||||
// delta. It returns an error if the adjustment fails.
|
||||
func (h *Host) countHealthFail(delta int) error {
|
||||
result := h.activeFails.Add(int64(delta))
|
||||
func (u *Upstream) countHealthFail(delta int) error {
|
||||
result := u.activeFails.Add(int64(delta))
|
||||
if result < 0 {
|
||||
return fmt.Errorf("count below 0: %d", result)
|
||||
}
|
||||
|
|
@ -240,9 +249,9 @@ func (h *Host) countHealthFail(delta int) error {
|
|||
}
|
||||
|
||||
// resetHealth resets the health check counters.
|
||||
func (h *Host) resetHealth() {
|
||||
h.activePasses.Store(0)
|
||||
h.activeFails.Store(0)
|
||||
func (u *Upstream) resetHealth() {
|
||||
u.activePasses.Store(0)
|
||||
u.activeFails.Store(0)
|
||||
}
|
||||
|
||||
// healthy returns true if the upstream is not actively marked as unhealthy.
|
||||
|
|
|
|||
Loading…
Reference in New Issue