diff --git a/modules/caddyhttp/reverseproxy/active_health_test.go b/modules/caddyhttp/reverseproxy/active_health_test.go new file mode 100644 index 000000000..96b73000e --- /dev/null +++ b/modules/caddyhttp/reverseproxy/active_health_test.go @@ -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") + } +} diff --git a/modules/caddyhttp/reverseproxy/healthchecks.go b/modules/caddyhttp/reverseproxy/healthchecks.go index 21fa0dfbb..f3515a326 100644 --- a/modules/caddyhttp/reverseproxy/healthchecks.go +++ b/modules/caddyhttp/reverseproxy/healthchecks.go @@ -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() } } } diff --git a/modules/caddyhttp/reverseproxy/hosts.go b/modules/caddyhttp/reverseproxy/hosts.go index 5c56c29e4..6b4750491 100644 --- a/modules/caddyhttp/reverseproxy/hosts.go +++ b/modules/caddyhttp/reverseproxy/hosts.go @@ -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.