@marcel91200 / caddy Mirror · read-only / commits / db298602
Fork 0

Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS

Mirrored from https://github.com/Slazee/caddy.git

Synced Nov 11, 14:45 UTC

reverseproxy: Track dynamic upstreams, enable passive healthchecking (#7539)

* reverseproxy: Track dynamic upstreams, enable passive healthchecking * Add tests for dynamic upstream tracking, admin endpoint, health checks

Francis Lavoie committed Mar 4, 2026 at 15:05 UTC db2986028fc573ae3add0a9a3381268dd7599267
6 files changed +1096 -13
modules/caddyhttp/reverseproxy/admin.go
+12 -1
@@ -75,7 +75,7 @@ func (adminUpstreams) handleUpstreams(w http.ResponseWriter, r *http.Request) er
75 results := []upstreamStatus{}
76 knownHosts := make(map[string]struct{})
77
78 - // Iterate over the upstream pool (needs to be fast)
78 + // Iterate over the static upstream pool (needs to be fast)
79 var rangeErr error
80 hosts.Range(func(key, val any) bool {
81 address, ok := key.(string)
@@ -121,6 +121,17 @@ func (adminUpstreams) handleUpstreams(w http.ResponseWriter, r *http.Request) er
121 return rangeErr
122 }
123
124 + // Also include dynamic upstreams
125 + dynamicHostsMu.RLock()
126 + for address, entry := range dynamicHosts {
127 + results = append(results, upstreamStatus{
128 + Address: address,
129 + NumRequests: entry.host.NumRequests(),
130 + Fails: entry.host.Fails(),
131 + })
132 + }
133 + dynamicHostsMu.RUnlock()
134 +
135 err := enc.Encode(results)
136 if err != nil {
137 return caddy.APIError{
modules/caddyhttp/reverseproxy/admin_test.go new
+275
@@ -0,0 +1,275 @@
1 +// Copyright 2015 Matthew Holt and The Caddy Authors
2 +//
3 +// Licensed under the Apache License, Version 2.0 (the "License");
4 +// you may not use this file except in compliance with the License.
5 +// You may obtain a copy of the License at
6 +//
7 +// http://www.apache.org/licenses/LICENSE-2.0
8 +//
9 +// Unless required by applicable law or agreed to in writing, software
10 +// distributed under the License is distributed on an "AS IS" BASIS,
11 +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 +// See the License for the specific language governing permissions and
13 +// limitations under the License.
14 +
15 +package reverseproxy
16 +
17 +import (
18 + "encoding/json"
19 + "net/http"
20 + "net/http/httptest"
21 + "testing"
22 + "time"
23 +)
24 +
25 +// adminHandlerFixture sets up the global host state for an admin endpoint test
26 +// and returns a cleanup function that must be deferred by the caller.
27 +//
28 +// staticAddrs are inserted into the UsagePool (as a static upstream would be).
29 +// dynamicAddrs are inserted into the dynamicHosts map (as a dynamic upstream would be).
30 +func adminHandlerFixture(t *testing.T, staticAddrs, dynamicAddrs []string) func() {
31 + t.Helper()
32 +
33 + for _, addr := range staticAddrs {
34 + u := &Upstream{Dial: addr}
35 + u.fillHost()
36 + }
37 +
38 + dynamicHostsMu.Lock()
39 + for _, addr := range dynamicAddrs {
40 + dynamicHosts[addr] = dynamicHostEntry{host: new(Host), lastSeen: time.Now()}
41 + }
42 + dynamicHostsMu.Unlock()
43 +
44 + return func() {
45 + // Remove static entries from the UsagePool.
46 + for _, addr := range staticAddrs {
47 + _, _ = hosts.Delete(addr)
48 + }
49 + // Remove dynamic entries.
50 + dynamicHostsMu.Lock()
51 + for _, addr := range dynamicAddrs {
52 + delete(dynamicHosts, addr)
53 + }
54 + dynamicHostsMu.Unlock()
55 + }
56 +}
57 +
58 +// callAdminUpstreams fires a GET against handleUpstreams and returns the
59 +// decoded response body.
60 +func callAdminUpstreams(t *testing.T) []upstreamStatus {
61 + t.Helper()
62 + req := httptest.NewRequest(http.MethodGet, "/reverse_proxy/upstreams", nil)
63 + w := httptest.NewRecorder()
64 +
65 + handler := adminUpstreams{}
66 + if err := handler.handleUpstreams(w, req); err != nil {
67 + t.Fatalf("handleUpstreams returned unexpected error: %v", err)
68 + }
69 + if w.Code != http.StatusOK {
70 + t.Fatalf("expected 200, got %d", w.Code)
71 + }
72 + if ct := w.Header().Get("Content-Type"); ct != "application/json" {
73 + t.Fatalf("expected Content-Type application/json, got %q", ct)
74 + }
75 +
76 + var results []upstreamStatus
77 + if err := json.NewDecoder(w.Body).Decode(&results); err != nil {
78 + t.Fatalf("failed to decode response: %v", err)
79 + }
80 + return results
81 +}
82 +
83 +// resultsByAddress indexes a slice of upstreamStatus by address for easier
84 +// lookup in assertions.
85 +func resultsByAddress(statuses []upstreamStatus) map[string]upstreamStatus {
86 + m := make(map[string]upstreamStatus, len(statuses))
87 + for _, s := range statuses {
88 + m[s.Address] = s
89 + }
90 + return m
91 +}
92 +
93 +// TestAdminUpstreamsMethodNotAllowed verifies that non-GET methods are rejected.
94 +func TestAdminUpstreamsMethodNotAllowed(t *testing.T) {
95 + for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} {
96 + req := httptest.NewRequest(method, "/reverse_proxy/upstreams", nil)
97 + w := httptest.NewRecorder()
98 + err := (adminUpstreams{}).handleUpstreams(w, req)
99 + if err == nil {
100 + t.Errorf("method %s: expected an error, got nil", method)
101 + continue
102 + }
103 + apiErr, ok := err.(interface{ HTTPStatus() int })
104 + if !ok {
105 + // caddy.APIError stores the code in HTTPStatus field, access via the
106 + // exported interface it satisfies indirectly; just check non-nil.
107 + continue
108 + }
109 + if code := apiErr.HTTPStatus(); code != http.StatusMethodNotAllowed {
110 + t.Errorf("method %s: expected 405, got %d", method, code)
111 + }
112 + }
113 +}
114 +
115 +// TestAdminUpstreamsEmpty verifies that an empty response is valid JSON when
116 +// no upstreams are registered.
117 +func TestAdminUpstreamsEmpty(t *testing.T) {
118 + resetDynamicHosts()
119 +
120 + results := callAdminUpstreams(t)
121 + if results == nil {
122 + t.Error("expected non-nil (empty) slice, got nil")
123 + }
124 + if len(results) != 0 {
125 + t.Errorf("expected 0 results with empty pools, got %d", len(results))
126 + }
127 +}
128 +
129 +// TestAdminUpstreamsStaticOnly verifies that static upstreams (from the
130 +// UsagePool) appear in the response with correct addresses.
131 +func TestAdminUpstreamsStaticOnly(t *testing.T) {
132 + resetDynamicHosts()
133 + cleanup := adminHandlerFixture(t,
134 + []string{"10.0.0.1:80", "10.0.0.2:80"},
135 + nil,
136 + )
137 + defer cleanup()
138 +
139 + results := callAdminUpstreams(t)
140 + byAddr := resultsByAddress(results)
141 +
142 + for _, addr := range []string{"10.0.0.1:80", "10.0.0.2:80"} {
143 + if _, ok := byAddr[addr]; !ok {
144 + t.Errorf("expected static upstream %q in response", addr)
145 + }
146 + }
147 + if len(results) != 2 {
148 + t.Errorf("expected exactly 2 results, got %d", len(results))
149 + }
150 +}
151 +
152 +// TestAdminUpstreamsDynamicOnly verifies that dynamic upstreams (from
153 +// dynamicHosts) appear in the response with correct addresses.
154 +func TestAdminUpstreamsDynamicOnly(t *testing.T) {
155 + resetDynamicHosts()
156 + cleanup := adminHandlerFixture(t,
157 + nil,
158 + []string{"10.0.1.1:80", "10.0.1.2:80"},
159 + )
160 + defer cleanup()
161 +
162 + results := callAdminUpstreams(t)
163 + byAddr := resultsByAddress(results)
164 +
165 + for _, addr := range []string{"10.0.1.1:80", "10.0.1.2:80"} {
166 + if _, ok := byAddr[addr]; !ok {
167 + t.Errorf("expected dynamic upstream %q in response", addr)
168 + }
169 + }
170 + if len(results) != 2 {
171 + t.Errorf("expected exactly 2 results, got %d", len(results))
172 + }
173 +}
174 +
175 +// TestAdminUpstreamsBothPools verifies that static and dynamic upstreams are
176 +// both present in the same response and that there is no overlap or omission.
177 +func TestAdminUpstreamsBothPools(t *testing.T) {
178 + resetDynamicHosts()
179 + cleanup := adminHandlerFixture(t,
180 + []string{"10.0.2.1:80"},
181 + []string{"10.0.2.2:80"},
182 + )
183 + defer cleanup()
184 +
185 + results := callAdminUpstreams(t)
186 + if len(results) != 2 {
187 + t.Fatalf("expected 2 results (1 static + 1 dynamic), got %d", len(results))
188 + }
189 +
190 + byAddr := resultsByAddress(results)
191 + if _, ok := byAddr["10.0.2.1:80"]; !ok {
192 + t.Error("static upstream missing from response")
193 + }
194 + if _, ok := byAddr["10.0.2.2:80"]; !ok {
195 + t.Error("dynamic upstream missing from response")
196 + }
197 +}
198 +
199 +// TestAdminUpstreamsNoOverlapBetweenPools verifies that an address registered
200 +// only as a static upstream does not also appear as a dynamic entry, and
201 +// vice-versa.
202 +func TestAdminUpstreamsNoOverlapBetweenPools(t *testing.T) {
203 + resetDynamicHosts()
204 + cleanup := adminHandlerFixture(t,
205 + []string{"10.0.3.1:80"},
206 + []string{"10.0.3.2:80"},
207 + )
208 + defer cleanup()
209 +
210 + results := callAdminUpstreams(t)
211 + seen := make(map[string]int)
212 + for _, r := range results {
213 + seen[r.Address]++
214 + }
215 + for addr, count := range seen {
216 + if count > 1 {
217 + t.Errorf("address %q appeared %d times; expected exactly once", addr, count)
218 + }
219 + }
220 +}
221 +
222 +// TestAdminUpstreamsReportsFailCounts verifies that fail counts accumulated on
223 +// a dynamic upstream's Host are reflected in the response.
224 +func TestAdminUpstreamsReportsFailCounts(t *testing.T) {
225 + resetDynamicHosts()
226 +
227 + const addr = "10.0.4.1:80"
228 + h := new(Host)
229 + _ = h.countFail(3)
230 +
231 + dynamicHostsMu.Lock()
232 + dynamicHosts[addr] = dynamicHostEntry{host: h, lastSeen: time.Now()}
233 + dynamicHostsMu.Unlock()
234 + defer func() {
235 + dynamicHostsMu.Lock()
236 + delete(dynamicHosts, addr)
237 + dynamicHostsMu.Unlock()
238 + }()
239 +
240 + results := callAdminUpstreams(t)
241 + byAddr := resultsByAddress(results)
242 +
243 + status, ok := byAddr[addr]
244 + if !ok {
245 + t.Fatalf("expected %q in response", addr)
246 + }
247 + if status.Fails != 3 {
248 + t.Errorf("expected Fails=3, got %d", status.Fails)
249 + }
250 +}
251 +
252 +// TestAdminUpstreamsReportsNumRequests verifies that the active request count
253 +// for a static upstream is reflected in the response.
254 +func TestAdminUpstreamsReportsNumRequests(t *testing.T) {
255 + resetDynamicHosts()
256 +
257 + const addr = "10.0.4.2:80"
258 + u := &Upstream{Dial: addr}
259 + u.fillHost()
260 + defer func() { _, _ = hosts.Delete(addr) }()
261 +
262 + _ = u.Host.countRequest(2)
263 + defer func() { _ = u.Host.countRequest(-2) }()
264 +
265 + results := callAdminUpstreams(t)
266 + byAddr := resultsByAddress(results)
267 +
268 + status, ok := byAddr[addr]
269 + if !ok {
270 + t.Fatalf("expected %q in response", addr)
271 + }
272 + if status.NumRequests != 2 {
273 + t.Errorf("expected NumRequests=2, got %d", status.NumRequests)
274 + }
275 +}
modules/caddyhttp/reverseproxy/dynamic_upstreams_test.go new
+345
@@ -0,0 +1,345 @@
1 +// Copyright 2015 Matthew Holt and The Caddy Authors
2 +//
3 +// Licensed under the Apache License, Version 2.0 (the "License");
4 +// you may not use this file except in compliance with the License.
5 +// You may obtain a copy of the License at
6 +//
7 +// http://www.apache.org/licenses/LICENSE-2.0
8 +//
9 +// Unless required by applicable law or agreed to in writing, software
10 +// distributed under the License is distributed on an "AS IS" BASIS,
11 +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 +// See the License for the specific language governing permissions and
13 +// limitations under the License.
14 +
15 +package reverseproxy
16 +
17 +import (
18 + "sync"
19 + "testing"
20 + "time"
21 +
22 + "github.com/caddyserver/caddy/v2"
23 +)
24 +
25 +// resetDynamicHosts clears global dynamic host state between tests.
26 +func resetDynamicHosts() {
27 + dynamicHostsMu.Lock()
28 + dynamicHosts = make(map[string]dynamicHostEntry)
29 + dynamicHostsMu.Unlock()
30 + // Reset the Once so cleanup goroutine tests can re-trigger if needed.
31 + dynamicHostsCleanerOnce = sync.Once{}
32 +}
33 +
34 +// TestFillDynamicHostCreatesEntry verifies that calling fillDynamicHost on a
35 +// new address inserts an entry into dynamicHosts and assigns a non-nil Host.
36 +func TestFillDynamicHostCreatesEntry(t *testing.T) {
37 + resetDynamicHosts()
38 +
39 + u := &Upstream{Dial: "192.0.2.1:80"}
40 + u.fillDynamicHost()
41 +
42 + if u.Host == nil {
43 + t.Fatal("expected Host to be set after fillDynamicHost")
44 + }
45 +
46 + dynamicHostsMu.RLock()
47 + entry, ok := dynamicHosts["192.0.2.1:80"]
48 + dynamicHostsMu.RUnlock()
49 +
50 + if !ok {
51 + t.Fatal("expected entry in dynamicHosts map")
52 + }
53 + if entry.host != u.Host {
54 + t.Error("dynamicHosts entry host should be the same pointer assigned to Upstream.Host")
55 + }
56 + if entry.lastSeen.IsZero() {
57 + t.Error("expected lastSeen to be set")
58 + }
59 +}
60 +
61 +// TestFillDynamicHostReusesSameHost verifies that two calls for the same address
62 +// return the exact same *Host pointer so that state (e.g. fail counts) is shared.
63 +func TestFillDynamicHostReusesSameHost(t *testing.T) {
64 + resetDynamicHosts()
65 +
66 + u1 := &Upstream{Dial: "192.0.2.2:80"}
67 + u1.fillDynamicHost()
68 +
69 + u2 := &Upstream{Dial: "192.0.2.2:80"}
70 + u2.fillDynamicHost()
71 +
72 + if u1.Host != u2.Host {
73 + t.Error("expected both upstreams to share the same *Host pointer")
74 + }
75 +}
76 +
77 +// TestFillDynamicHostUpdatesLastSeen verifies that a second call for the same
78 +// address advances the lastSeen timestamp.
79 +func TestFillDynamicHostUpdatesLastSeen(t *testing.T) {
80 + resetDynamicHosts()
81 +
82 + u := &Upstream{Dial: "192.0.2.3:80"}
83 + u.fillDynamicHost()
84 +
85 + dynamicHostsMu.RLock()
86 + first := dynamicHosts["192.0.2.3:80"].lastSeen
87 + dynamicHostsMu.RUnlock()
88 +
89 + // Ensure measurable time passes.
90 + time.Sleep(2 * time.Millisecond)
91 +
92 + u2 := &Upstream{Dial: "192.0.2.3:80"}
93 + u2.fillDynamicHost()
94 +
95 + dynamicHostsMu.RLock()
96 + second := dynamicHosts["192.0.2.3:80"].lastSeen
97 + dynamicHostsMu.RUnlock()
98 +
99 + if !second.After(first) {
100 + t.Error("expected lastSeen to be updated on second fillDynamicHost call")
101 + }
102 +}
103 +
104 +// TestFillDynamicHostIndependentAddresses verifies that different addresses get
105 +// independent Host entries.
106 +func TestFillDynamicHostIndependentAddresses(t *testing.T) {
107 + resetDynamicHosts()
108 +
109 + u1 := &Upstream{Dial: "192.0.2.4:80"}
110 + u1.fillDynamicHost()
111 +
112 + u2 := &Upstream{Dial: "192.0.2.5:80"}
113 + u2.fillDynamicHost()
114 +
115 + if u1.Host == u2.Host {
116 + t.Error("different addresses should have different *Host entries")
117 + }
118 +}
119 +
120 +// TestFillDynamicHostPreservesFailCount verifies that fail counts on a dynamic
121 +// host survive across multiple fillDynamicHost calls (simulating sequential
122 +// requests), which is the core behaviour fixed by this change.
123 +func TestFillDynamicHostPreservesFailCount(t *testing.T) {
124 + resetDynamicHosts()
125 +
126 + // First "request": provision and record a failure.
127 + u1 := &Upstream{Dial: "192.0.2.6:80"}
128 + u1.fillDynamicHost()
129 + _ = u1.Host.countFail(1)
130 +
131 + if u1.Host.Fails() != 1 {
132 + t.Fatalf("expected 1 fail, got %d", u1.Host.Fails())
133 + }
134 +
135 + // Second "request": provision the same address again (new *Upstream, same address).
136 + u2 := &Upstream{Dial: "192.0.2.6:80"}
137 + u2.fillDynamicHost()
138 +
139 + if u2.Host.Fails() != 1 {
140 + t.Errorf("expected fail count to persist across fillDynamicHost calls, got %d", u2.Host.Fails())
141 + }
142 +}
143 +
144 +// TestProvisionUpstreamDynamic verifies that provisionUpstream with dynamic=true
145 +// uses fillDynamicHost (not the UsagePool) and sets healthCheckPolicy /
146 +// MaxRequests correctly from handler config.
147 +func TestProvisionUpstreamDynamic(t *testing.T) {
148 + resetDynamicHosts()
149 +
150 + passive := &PassiveHealthChecks{
151 + FailDuration: caddy.Duration(10 * time.Second),
152 + MaxFails: 3,
153 + UnhealthyRequestCount: 5,
154 + }
155 + h := Handler{
156 + HealthChecks: &HealthChecks{
157 + Passive: passive,
158 + },
159 + }
160 +
161 + u := &Upstream{Dial: "192.0.2.7:80"}
162 + h.provisionUpstream(u, true)
163 +
164 + if u.Host == nil {
165 + t.Fatal("Host should be set after provisionUpstream")
166 + }
167 + if u.healthCheckPolicy != passive {
168 + t.Error("healthCheckPolicy should point to the handler's PassiveHealthChecks")
169 + }
170 + if u.MaxRequests != 5 {
171 + t.Errorf("expected MaxRequests=5 from UnhealthyRequestCount, got %d", u.MaxRequests)
172 + }
173 +
174 + // Must be in dynamicHosts, not in the static UsagePool.
175 + dynamicHostsMu.RLock()
176 + _, inDynamic := dynamicHosts["192.0.2.7:80"]
177 + dynamicHostsMu.RUnlock()
178 + if !inDynamic {
179 + t.Error("dynamic upstream should be stored in dynamicHosts")
180 + }
181 + _, inPool := hosts.References("192.0.2.7:80")
182 + if inPool {
183 + t.Error("dynamic upstream should NOT be stored in the static UsagePool")
184 + }
185 +}
186 +
187 +// TestProvisionUpstreamStatic verifies that provisionUpstream with dynamic=false
188 +// uses the UsagePool and does NOT insert into dynamicHosts.
189 +func TestProvisionUpstreamStatic(t *testing.T) {
190 + resetDynamicHosts()
191 +
192 + h := Handler{}
193 +
194 + u := &Upstream{Dial: "192.0.2.8:80"}
195 + h.provisionUpstream(u, false)
196 +
197 + if u.Host == nil {
198 + t.Fatal("Host should be set after provisionUpstream")
199 + }
200 +
201 + refs, inPool := hosts.References("192.0.2.8:80")
202 + if !inPool {
203 + t.Error("static upstream should be in the UsagePool")
204 + }
205 + if refs != 1 {
206 + t.Errorf("expected ref count 1, got %d", refs)
207 + }
208 +
209 + dynamicHostsMu.RLock()
210 + _, inDynamic := dynamicHosts["192.0.2.8:80"]
211 + dynamicHostsMu.RUnlock()
212 + if inDynamic {
213 + t.Error("static upstream should NOT be in dynamicHosts")
214 + }
215 +
216 + // Clean up the pool entry we just added.
217 + _, _ = hosts.Delete("192.0.2.8:80")
218 +}
219 +
220 +// TestDynamicHostHealthyConsultsFails verifies the end-to-end passive health
221 +// check path: after enough failures are recorded against a dynamic upstream's
222 +// shared *Host, Healthy() returns false for a newly provisioned *Upstream with
223 +// the same address.
224 +func TestDynamicHostHealthyConsultsFails(t *testing.T) {
225 + resetDynamicHosts()
226 +
227 + passive := &PassiveHealthChecks{
228 + FailDuration: caddy.Duration(time.Minute),
229 + MaxFails: 2,
230 + }
231 + h := Handler{
232 + HealthChecks: &HealthChecks{Passive: passive},
233 + }
234 +
235 + // First request: provision and record two failures.
236 + u1 := &Upstream{Dial: "192.0.2.9:80"}
237 + h.provisionUpstream(u1, true)
238 +
239 + _ = u1.Host.countFail(1)
240 + _ = u1.Host.countFail(1)
241 +
242 + // Second request: fresh *Upstream, same address.
243 + u2 := &Upstream{Dial: "192.0.2.9:80"}
244 + h.provisionUpstream(u2, true)
245 +
246 + if u2.Healthy() {
247 + t.Error("upstream should be unhealthy after MaxFails failures have been recorded against its shared Host")
248 + }
249 +}
250 +
251 +// TestDynamicHostCleanupEvictsStaleEntries verifies that the cleanup sweep
252 +// removes entries whose lastSeen is older than dynamicHostIdleExpiry.
253 +func TestDynamicHostCleanupEvictsStaleEntries(t *testing.T) {
254 + resetDynamicHosts()
255 +
256 + const addr = "192.0.2.10:80"
257 +
258 + // Insert an entry directly with a lastSeen far in the past.
259 + dynamicHostsMu.Lock()
260 + dynamicHosts[addr] = dynamicHostEntry{
261 + host: new(Host),
262 + lastSeen: time.Now().Add(-2 * dynamicHostIdleExpiry),
263 + }
264 + dynamicHostsMu.Unlock()
265 +
266 + // Run the cleanup logic inline (same logic as the goroutine).
267 + dynamicHostsMu.Lock()
268 + for a, entry := range dynamicHosts {
269 + if time.Since(entry.lastSeen) > dynamicHostIdleExpiry {
270 + delete(dynamicHosts, a)
271 + }
272 + }
273 + dynamicHostsMu.Unlock()
274 +
275 + dynamicHostsMu.RLock()
276 + _, stillPresent := dynamicHosts[addr]
277 + dynamicHostsMu.RUnlock()
278 +
279 + if stillPresent {
280 + t.Error("stale dynamic host entry should have been evicted by cleanup sweep")
281 + }
282 +}
283 +
284 +// TestDynamicHostCleanupRetainsFreshEntries verifies that the cleanup sweep
285 +// keeps entries whose lastSeen is within dynamicHostIdleExpiry.
286 +func TestDynamicHostCleanupRetainsFreshEntries(t *testing.T) {
287 + resetDynamicHosts()
288 +
289 + const addr = "192.0.2.11:80"
290 +
291 + dynamicHostsMu.Lock()
292 + dynamicHosts[addr] = dynamicHostEntry{
293 + host: new(Host),
294 + lastSeen: time.Now(),
295 + }
296 + dynamicHostsMu.Unlock()
297 +
298 + // Run the cleanup logic inline.
299 + dynamicHostsMu.Lock()
300 + for a, entry := range dynamicHosts {
301 + if time.Since(entry.lastSeen) > dynamicHostIdleExpiry {
302 + delete(dynamicHosts, a)
303 + }
304 + }
305 + dynamicHostsMu.Unlock()
306 +
307 + dynamicHostsMu.RLock()
308 + _, stillPresent := dynamicHosts[addr]
309 + dynamicHostsMu.RUnlock()
310 +
311 + if !stillPresent {
312 + t.Error("fresh dynamic host entry should be retained by cleanup sweep")
313 + }
314 +}
315 +
316 +// TestDynamicHostConcurrentFillHost verifies that concurrent calls to
317 +// fillDynamicHost for the same address all get the same *Host pointer and
318 +// don't race (run with -race).
319 +func TestDynamicHostConcurrentFillHost(t *testing.T) {
320 + resetDynamicHosts()
321 +
322 + const addr = "192.0.2.12:80"
323 + const goroutines = 50
324 +
325 + var wg sync.WaitGroup
326 + hosts := make([]*Host, goroutines)
327 +
328 + for i := range goroutines {
329 + wg.Add(1)
330 + go func(idx int) {
331 + defer wg.Done()
332 + u := &Upstream{Dial: addr}
333 + u.fillDynamicHost()
334 + hosts[idx] = u.Host
335 + }(i)
336 + }
337 + wg.Wait()
338 +
339 + first := hosts[0]
340 + for i, h := range hosts {
341 + if h != first {
342 + t.Errorf("goroutine %d got a different *Host pointer; expected all to share the same entry", i)
343 + }
344 + }
345 +}
modules/caddyhttp/reverseproxy/hosts.go
+61
@@ -19,7 +19,9 @@ import (
19 "fmt"
20 "net/netip"
21 "strconv"
22 + "sync"
23 "sync/atomic"
24 + "time"
25
26 "github.com/caddyserver/caddy/v2"
27 "github.com/caddyserver/caddy/v2/modules/caddyhttp"
@@ -132,6 +134,43 @@ func (u *Upstream) fillHost() {
134 u.Host = host
135 }
136
137 +// fillDynamicHost is like fillHost, but stores the host in the separate
138 +// dynamicHosts map rather than the reference-counted UsagePool. Dynamic
139 +// hosts are not reference-counted; instead, they are retained as long as
140 +// they are actively seen and are evicted by a background cleanup goroutine
141 +// after dynamicHostIdleExpiry of inactivity. This preserves health state
142 +// (e.g. passive fail counts) across sequential requests.
143 +func (u *Upstream) fillDynamicHost() {
144 + dynamicHostsMu.Lock()
145 + entry, ok := dynamicHosts[u.String()]
146 + if ok {
147 + entry.lastSeen = time.Now()
148 + dynamicHosts[u.String()] = entry
149 + u.Host = entry.host
150 + } else {
151 + h := new(Host)
152 + dynamicHosts[u.String()] = dynamicHostEntry{host: h, lastSeen: time.Now()}
153 + u.Host = h
154 + }
155 + dynamicHostsMu.Unlock()
156 +
157 + // ensure the cleanup goroutine is running
158 + dynamicHostsCleanerOnce.Do(func() {
159 + go func() {
160 + for {
161 + time.Sleep(dynamicHostCleanupInterval)
162 + dynamicHostsMu.Lock()
163 + for addr, entry := range dynamicHosts {
164 + if time.Since(entry.lastSeen) > dynamicHostIdleExpiry {
165 + delete(dynamicHosts, addr)
166 + }
167 + }
168 + dynamicHostsMu.Unlock()
169 + }
170 + }()
171 + })
172 +}
173 +
174 // Host is the basic, in-memory representation of the state of a remote host.
175 // Its fields are accessed atomically and Host values must not be copied.
176 type Host struct {
@@ -268,6 +307,28 @@ func GetDialInfo(ctx context.Context) (DialInfo, bool) {
307 // through config reloads.
308 var hosts = caddy.NewUsagePool()
309
310 +// dynamicHosts tracks hosts that were provisioned from dynamic upstream
311 +// sources. Unlike static upstreams which are reference-counted via the
312 +// UsagePool, dynamic upstream hosts are not reference-counted. Instead,
313 +// their last-seen time is updated on each request, and a background
314 +// goroutine evicts entries that have been idle for dynamicHostIdleExpiry.
315 +// This preserves health state (e.g. passive fail counts) across requests
316 +// to the same dynamic backend.
317 +var (
318 + dynamicHosts = make(map[string]dynamicHostEntry)
319 + dynamicHostsMu sync.RWMutex
320 + dynamicHostsCleanerOnce sync.Once
321 + dynamicHostCleanupInterval = 5 * time.Minute
322 + dynamicHostIdleExpiry = time.Hour
323 +)
324 +
325 +// dynamicHostEntry holds a Host and the last time it was seen
326 +// in a set of dynamic upstreams returned for a request.
327 +type dynamicHostEntry struct {
328 + host *Host
329 + lastSeen time.Time
330 +}
331 +
332 // dialInfoVarKey is the key used for the variable that holds
333 // the dial info for the upstream connection.
334 const dialInfoVarKey = "reverse_proxy.dial_info"
modules/caddyhttp/reverseproxy/passive_health_test.go new
+391
@@ -0,0 +1,391 @@
1 +// Copyright 2015 Matthew Holt and The Caddy Authors
2 +//
3 +// Licensed under the Apache License, Version 2.0 (the "License");
4 +// you may not use this file except in compliance with the License.
5 +// You may obtain a copy of the License at
6 +//
7 +// http://www.apache.org/licenses/LICENSE-2.0
8 +//
9 +// Unless required by applicable law or agreed to in writing, software
10 +// distributed under the License is distributed on an "AS IS" BASIS,
11 +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 +// See the License for the specific language governing permissions and
13 +// limitations under the License.
14 +
15 +package reverseproxy
16 +
17 +import (
18 + "context"
19 + "testing"
20 + "time"
21 +
22 + "github.com/caddyserver/caddy/v2"
23 +)
24 +
25 +// newPassiveHandler builds a minimal Handler with passive health checks
26 +// configured and a live caddy.Context so the fail-forgetter goroutine can
27 +// be cancelled cleanly. The caller must call cancel() when done.
28 +func newPassiveHandler(t *testing.T, maxFails int, failDuration time.Duration) (*Handler, context.CancelFunc) {
29 + t.Helper()
30 + caddyCtx, cancel := caddy.NewContext(caddy.Context{Context: context.Background()})
31 + h := &Handler{
32 + ctx: caddyCtx,
33 + HealthChecks: &HealthChecks{
34 + Passive: &PassiveHealthChecks{
35 + MaxFails: maxFails,
36 + FailDuration: caddy.Duration(failDuration),
37 + },
38 + },
39 + }
40 + return h, cancel
41 +}
42 +
43 +// provisionedStaticUpstream creates a static upstream, registers it in the
44 +// UsagePool, and returns a cleanup func that removes it from the pool.
45 +func provisionedStaticUpstream(t *testing.T, h *Handler, addr string) (*Upstream, func()) {
46 + t.Helper()
47 + u := &Upstream{Dial: addr}
48 + h.provisionUpstream(u, false)
49 + return u, func() { _, _ = hosts.Delete(addr) }
50 +}
51 +
52 +// provisionedDynamicUpstream creates a dynamic upstream, registers it in
53 +// dynamicHosts, and returns a cleanup func that removes it.
54 +func provisionedDynamicUpstream(t *testing.T, h *Handler, addr string) (*Upstream, func()) {
55 + t.Helper()
56 + u := &Upstream{Dial: addr}
57 + h.provisionUpstream(u, true)
58 + return u, func() {
59 + dynamicHostsMu.Lock()
60 + delete(dynamicHosts, addr)
61 + dynamicHostsMu.Unlock()
62 + }
63 +}
64 +
65 +// --- countFailure behaviour ---
66 +
67 +// TestCountFailureNoopWhenNoHealthChecks verifies that countFailure is a no-op
68 +// when HealthChecks is nil.
69 +func TestCountFailureNoopWhenNoHealthChecks(t *testing.T) {
70 + resetDynamicHosts()
71 + h := &Handler{}
72 + u := &Upstream{Dial: "10.1.0.1:80", Host: new(Host)}
73 +
74 + h.countFailure(u)
75 +
76 + if u.Host.Fails() != 0 {
77 + t.Errorf("expected 0 fails with no HealthChecks config, got %d", u.Host.Fails())
78 + }
79 +}
80 +
81 +// TestCountFailureNoopWhenZeroDuration verifies that countFailure is a no-op
82 +// when FailDuration is 0 (the zero value disables passive checks).
83 +func TestCountFailureNoopWhenZeroDuration(t *testing.T) {
84 + resetDynamicHosts()
85 + caddyCtx, cancel := caddy.NewContext(caddy.Context{Context: context.Background()})
86 + defer cancel()
87 + h := &Handler{
88 + ctx: caddyCtx,
89 + HealthChecks: &HealthChecks{
90 + Passive: &PassiveHealthChecks{MaxFails: 1, FailDuration: 0},
91 + },
92 + }
93 + u := &Upstream{Dial: "10.1.0.2:80", Host: new(Host)}
94 +
95 + h.countFailure(u)
96 +
97 + if u.Host.Fails() != 0 {
98 + t.Errorf("expected 0 fails with zero FailDuration, got %d", u.Host.Fails())
99 + }
100 +}
101 +
102 +// TestCountFailureIncrementsCount verifies that countFailure increments the
103 +// fail count on the upstream's Host.
104 +func TestCountFailureIncrementsCount(t *testing.T) {
105 + resetDynamicHosts()
106 + h, cancel := newPassiveHandler(t, 2, time.Minute)
107 + defer cancel()
108 + u := &Upstream{Dial: "10.1.0.3:80", Host: new(Host)}
109 +
110 + h.countFailure(u)
111 +
112 + if u.Host.Fails() != 1 {
113 + t.Errorf("expected 1 fail after countFailure, got %d", u.Host.Fails())
114 + }
115 +}
116 +
117 +// TestCountFailureDecrementsAfterDuration verifies that the fail count is
118 +// decremented back after FailDuration elapses.
119 +func TestCountFailureDecrementsAfterDuration(t *testing.T) {
120 + resetDynamicHosts()
121 + const failDuration = 50 * time.Millisecond
122 + h, cancel := newPassiveHandler(t, 2, failDuration)
123 + defer cancel()
124 + u := &Upstream{Dial: "10.1.0.4:80", Host: new(Host)}
125 +
126 + h.countFailure(u)
127 + if u.Host.Fails() != 1 {
128 + t.Fatalf("expected 1 fail immediately after countFailure, got %d", u.Host.Fails())
129 + }
130 +
131 + // Wait long enough for the forgetter goroutine to fire.
132 + time.Sleep(3 * failDuration)
133 +
134 + if u.Host.Fails() != 0 {
135 + t.Errorf("expected fail count to return to 0 after FailDuration, got %d", u.Host.Fails())
136 + }
137 +}
138 +
139 +// TestCountFailureCancelledContextForgets verifies that cancelling the handler
140 +// context (simulating a config unload) also triggers the forgetter to run,
141 +// decrementing the fail count.
142 +func TestCountFailureCancelledContextForgets(t *testing.T) {
143 + resetDynamicHosts()
144 + h, cancel := newPassiveHandler(t, 2, time.Hour) // very long duration
145 + u := &Upstream{Dial: "10.1.0.5:80", Host: new(Host)}
146 +
147 + h.countFailure(u)
148 + if u.Host.Fails() != 1 {
149 + t.Fatalf("expected 1 fail immediately after countFailure, got %d", u.Host.Fails())
150 + }
151 +
152 + // Cancelling the context should cause the forgetter goroutine to exit and
153 + // decrement the count.
154 + cancel()
155 + time.Sleep(50 * time.Millisecond)
156 +
157 + if u.Host.Fails() != 0 {
158 + t.Errorf("expected fail count to be decremented after context cancel, got %d", u.Host.Fails())
159 + }
160 +}
161 +
162 +// --- static upstream passive health check ---
163 +
164 +// TestStaticUpstreamHealthyWithNoFailures verifies that a static upstream with
165 +// no recorded failures is considered healthy.
166 +func TestStaticUpstreamHealthyWithNoFailures(t *testing.T) {
167 + resetDynamicHosts()
168 + h, cancel := newPassiveHandler(t, 2, time.Minute)
169 + defer cancel()
170 +
171 + u, cleanup := provisionedStaticUpstream(t, h, "10.2.0.1:80")
172 + defer cleanup()
173 +
174 + if !u.Healthy() {
175 + t.Error("upstream with no failures should be healthy")
176 + }
177 +}
178 +
179 +// TestStaticUpstreamUnhealthyAtMaxFails verifies that a static upstream is
180 +// marked unhealthy once its fail count reaches MaxFails.
181 +func TestStaticUpstreamUnhealthyAtMaxFails(t *testing.T) {
182 + resetDynamicHosts()
183 + h, cancel := newPassiveHandler(t, 2, time.Minute)
184 + defer cancel()
185 +
186 + u, cleanup := provisionedStaticUpstream(t, h, "10.2.0.2:80")
187 + defer cleanup()
188 +
189 + h.countFailure(u)
190 + if !u.Healthy() {
191 + t.Error("upstream should still be healthy after 1 of 2 allowed failures")
192 + }
193 +
194 + h.countFailure(u)
195 + if u.Healthy() {
196 + t.Error("upstream should be unhealthy after reaching MaxFails=2")
197 + }
198 +}
199 +
200 +// TestStaticUpstreamRecoversAfterFailDuration verifies that a static upstream
201 +// returns to healthy once its failures expire.
202 +func TestStaticUpstreamRecoversAfterFailDuration(t *testing.T) {
203 + resetDynamicHosts()
204 + const failDuration = 50 * time.Millisecond
205 + h, cancel := newPassiveHandler(t, 1, failDuration)
206 + defer cancel()
207 +
208 + u, cleanup := provisionedStaticUpstream(t, h, "10.2.0.3:80")
209 + defer cleanup()
210 +
211 + h.countFailure(u)
212 + if u.Healthy() {
213 + t.Fatal("upstream should be unhealthy immediately after MaxFails failure")
214 + }
215 +
216 + time.Sleep(3 * failDuration)
217 +
218 + if !u.Healthy() {
219 + t.Errorf("upstream should recover to healthy after FailDuration, Fails=%d", u.Host.Fails())
220 + }
221 +}
222 +
223 +// TestStaticUpstreamHealthPersistedAcrossReprovisioning verifies that static
224 +// upstreams share a Host via the UsagePool, so a second call to provisionUpstream
225 +// for the same address (as happens on config reload) sees the accumulated state.
226 +func TestStaticUpstreamHealthPersistedAcrossReprovisioning(t *testing.T) {
227 + resetDynamicHosts()
228 + h, cancel := newPassiveHandler(t, 2, time.Minute)
229 + defer cancel()
230 +
231 + u1, cleanup1 := provisionedStaticUpstream(t, h, "10.2.0.4:80")
232 + defer cleanup1()
233 +
234 + h.countFailure(u1)
235 + h.countFailure(u1)
236 +
237 + // Simulate a second handler instance referencing the same upstream
238 + // (e.g. after a config reload that keeps the same backend address).
239 + u2, cleanup2 := provisionedStaticUpstream(t, h, "10.2.0.4:80")
240 + defer cleanup2()
241 +
242 + if u1.Host != u2.Host {
243 + t.Fatal("expected both Upstream structs to share the same *Host via UsagePool")
244 + }
245 + if u2.Healthy() {
246 + t.Error("re-provisioned upstream should still see the prior fail count and be unhealthy")
247 + }
248 +}
249 +
250 +// --- dynamic upstream passive health check ---
251 +
252 +// TestDynamicUpstreamHealthyWithNoFailures verifies that a freshly provisioned
253 +// dynamic upstream is healthy.
254 +func TestDynamicUpstreamHealthyWithNoFailures(t *testing.T) {
255 + resetDynamicHosts()
256 + h, cancel := newPassiveHandler(t, 2, time.Minute)
257 + defer cancel()
258 +
259 + u, cleanup := provisionedDynamicUpstream(t, h, "10.3.0.1:80")
260 + defer cleanup()
261 +
262 + if !u.Healthy() {
263 + t.Error("dynamic upstream with no failures should be healthy")
264 + }
265 +}
266 +
267 +// TestDynamicUpstreamUnhealthyAtMaxFails verifies that a dynamic upstream is
268 +// marked unhealthy once its fail count reaches MaxFails.
269 +func TestDynamicUpstreamUnhealthyAtMaxFails(t *testing.T) {
270 + resetDynamicHosts()
271 + h, cancel := newPassiveHandler(t, 2, time.Minute)
272 + defer cancel()
273 +
274 + u, cleanup := provisionedDynamicUpstream(t, h, "10.3.0.2:80")
275 + defer cleanup()
276 +
277 + h.countFailure(u)
278 + if !u.Healthy() {
279 + t.Error("dynamic upstream should still be healthy after 1 of 2 allowed failures")
280 + }
281 +
282 + h.countFailure(u)
283 + if u.Healthy() {
284 + t.Error("dynamic upstream should be unhealthy after reaching MaxFails=2")
285 + }
286 +}
287 +
288 +// TestDynamicUpstreamFailCountPersistedBetweenRequests is the core regression
289 +// test: it simulates two sequential (non-concurrent) requests to the same
290 +// dynamic upstream. Before the fix, the UsagePool entry would be deleted
291 +// between requests, wiping the fail count. Now it should survive.
292 +func TestDynamicUpstreamFailCountPersistedBetweenRequests(t *testing.T) {
293 + resetDynamicHosts()
294 + h, cancel := newPassiveHandler(t, 2, time.Minute)
295 + defer cancel()
296 +
297 + // --- first request ---
298 + u1 := &Upstream{Dial: "10.3.0.3:80"}
299 + h.provisionUpstream(u1, true)
300 + h.countFailure(u1)
301 +
302 + if u1.Host.Fails() != 1 {
303 + t.Fatalf("expected 1 fail after first request, got %d", u1.Host.Fails())
304 + }
305 +
306 + // Simulate end of first request: no delete from any pool (key difference
307 + // vs. the old behaviour where hosts.Delete was deferred).
308 +
309 + // --- second request: brand-new *Upstream struct, same dial address ---
310 + u2 := &Upstream{Dial: "10.3.0.3:80"}
311 + h.provisionUpstream(u2, true)
312 +
313 + if u1.Host != u2.Host {
314 + t.Fatal("expected both requests to share the same *Host pointer from dynamicHosts")
315 + }
316 + if u2.Host.Fails() != 1 {
317 + t.Errorf("expected fail count to persist across requests, got %d", u2.Host.Fails())
318 + }
319 +
320 + // A second failure now tips it over MaxFails=2.
321 + h.countFailure(u2)
322 + if u2.Healthy() {
323 + t.Error("upstream should be unhealthy after accumulated failures across requests")
324 + }
325 +
326 + // Cleanup.
327 + dynamicHostsMu.Lock()
328 + delete(dynamicHosts, "10.3.0.3:80")
329 + dynamicHostsMu.Unlock()
330 +}
331 +
332 +// TestDynamicUpstreamRecoveryAfterFailDuration verifies that a dynamic
333 +// upstream's fail count expires and it returns to healthy.
334 +func TestDynamicUpstreamRecoveryAfterFailDuration(t *testing.T) {
335 + resetDynamicHosts()
336 + const failDuration = 50 * time.Millisecond
337 + h, cancel := newPassiveHandler(t, 1, failDuration)
338 + defer cancel()
339 +
340 + u, cleanup := provisionedDynamicUpstream(t, h, "10.3.0.4:80")
341 + defer cleanup()
342 +
343 + h.countFailure(u)
344 + if u.Healthy() {
345 + t.Fatal("upstream should be unhealthy immediately after MaxFails failure")
346 + }
347 +
348 + time.Sleep(3 * failDuration)
349 +
350 + // Re-provision (as a new request would) to get fresh *Upstream with policy set.
351 + u2 := &Upstream{Dial: "10.3.0.4:80"}
352 + h.provisionUpstream(u2, true)
353 +
354 + if !u2.Healthy() {
355 + t.Errorf("dynamic upstream should recover to healthy after FailDuration, Fails=%d", u2.Host.Fails())
356 + }
357 +}
358 +
359 +// TestDynamicUpstreamMaxRequestsFromUnhealthyRequestCount verifies that
360 +// UnhealthyRequestCount is copied into MaxRequests so Full() works correctly.
361 +func TestDynamicUpstreamMaxRequestsFromUnhealthyRequestCount(t *testing.T) {
362 + resetDynamicHosts()
363 + caddyCtx, cancel := caddy.NewContext(caddy.Context{Context: context.Background()})
364 + defer cancel()
365 + h := &Handler{
366 + ctx: caddyCtx,
367 + HealthChecks: &HealthChecks{
368 + Passive: &PassiveHealthChecks{
369 + UnhealthyRequestCount: 3,
370 + },
371 + },
372 + }
373 +
374 + u, cleanup := provisionedDynamicUpstream(t, h, "10.3.0.5:80")
375 + defer cleanup()
376 +
377 + if u.MaxRequests != 3 {
378 + t.Errorf("expected MaxRequests=3 from UnhealthyRequestCount, got %d", u.MaxRequests)
379 + }
380 +
381 + // Should not be full with fewer requests than the limit.
382 + _ = u.Host.countRequest(2)
383 + if u.Full() {
384 + t.Error("upstream should not be full with 2 of 3 allowed requests")
385 + }
386 +
387 + _ = u.Host.countRequest(1)
388 + if !u.Full() {
389 + t.Error("upstream should be full at UnhealthyRequestCount concurrent requests")
390 + }
391 +}
modules/caddyhttp/reverseproxy/reverseproxy.go
+12 -12
@@ -392,7 +392,7 @@ func (h *Handler) Provision(ctx caddy.Context) error {
392
393 // set up upstreams
394 for _, u := range h.Upstreams {
395 - h.provisionUpstream(u)
395 + h.provisionUpstream(u, false)
396 }
397
398 if h.HealthChecks != nil {
@@ -563,18 +563,11 @@ func (h *Handler) proxyLoopIteration(r *http.Request, origReq *http.Request, w h
563 } else {
564 upstreams = dUpstreams
565 for _, dUp := range dUpstreams {
566 - h.provisionUpstream(dUp)
566 + h.provisionUpstream(dUp, true)
567 }
568 if c := h.logger.Check(zapcore.DebugLevel, "provisioned dynamic upstreams"); c != nil {
569 c.Write(zap.Int("count", len(dUpstreams)))
570 }
571 - defer func() {
572 - // these upstreams are dynamic, so they are only used for this iteration
573 - // of the proxy loop; be sure to let them go away when we're done with them
574 - for _, upstream := range dUpstreams {
575 - _, _ = hosts.Delete(upstream.String())
576 - }
577 - }()
571 }
572 }
573
@@ -1324,9 +1317,16 @@ func (h *Handler) directRequest(req *http.Request, di DialInfo) {
1317 req.URL.Host = reqHost
1318 }
1319
1327 -func (h Handler) provisionUpstream(upstream *Upstream) {
1328 - // create or get the host representation for this upstream
1329 - upstream.fillHost()
1320 +func (h Handler) provisionUpstream(upstream *Upstream, dynamic bool) {
1321 + // create or get the host representation for this upstream;
1322 + // dynamic upstreams are tracked in a separate map with last-seen
1323 + // timestamps so their health state persists across requests without
1324 + // being reference-counted (and thus discarded between requests).
1325 + if dynamic {
1326 + upstream.fillDynamicHost()
1327 + } else {
1328 + upstream.fillHost()
1329 + }
1330
1331 // give it the circuit breaker, if any
1332 upstream.cb = h.CB