Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 13, 01:45 UTC
reverseproxy: Experimental streaming timeouts (#5567)
* reverseproxy: WIP streaming timeouts * More verbose logging by using the child logger * reverseproxy: Implement streaming timeouts * reverseproxy: Refactor cleanup * reverseproxy: Avoid **time.Timer --------- Co-authored-by: Francis Lavoie <lavofr@gmail.com>
mmm444 committed
Jun 19, 2023 at 23:54 UTC
424ae0f420f478e1b38189fd6632d29e13df7eee
3 files changed
+133
-34
modules/caddyhttp/reverseproxy/caddyfile.go
+32
-2
@@ -83,10 +83,12 @@ func parseCaddyfile(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error)
83
// unhealthy_request_count <num>
84
//
85
// # streaming
86
-// flush_interval <duration>
86
+// flush_interval <duration>
87
// buffer_requests
88
// buffer_responses
89
-// max_buffer_size <size>
89
+// max_buffer_size <size>
90
+// stream_timeout <duration>
91
+// stream_close_delay <duration>
92
//
93
// # request manipulation
94
// trusted_proxies [private_ranges] <ranges...>
@@ -571,6 +573,34 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
573
caddy.Log().Named("config.adapter.caddyfile").Warn("DEPRECATED: max_buffer_size: use request_buffers and/or response_buffers instead (with maximum buffer sizes)")
574
h.DeprecatedMaxBufferSize = int64(size)
575
576
+ case "stream_timeout":
577
+ if !d.NextArg() {
578
+ return d.ArgErr()
579
+ }
580
+ if fi, err := strconv.Atoi(d.Val()); err == nil {
581
+ h.StreamTimeout = caddy.Duration(fi)
582
+ } else {
583
+ dur, err := caddy.ParseDuration(d.Val())
584
+ if err != nil {
585
+ return d.Errf("bad duration value '%s': %v", d.Val(), err)
586
+ }
587
+ h.StreamTimeout = caddy.Duration(dur)
588
+ }
589
+
590
+ case "stream_close_delay":
591
+ if !d.NextArg() {
592
+ return d.ArgErr()
593
+ }
594
+ if fi, err := strconv.Atoi(d.Val()); err == nil {
595
+ h.StreamCloseDelay = caddy.Duration(fi)
596
+ } else {
597
+ dur, err := caddy.ParseDuration(d.Val())
598
+ if err != nil {
599
+ return d.Errf("bad duration value '%s': %v", d.Val(), err)
600
+ }
601
+ h.StreamCloseDelay = caddy.Duration(dur)
602
+ }
603
+
604
case "trusted_proxies":
605
for d.NextArg() {
606
if d.Val() == "private_ranges" {
modules/caddyhttp/reverseproxy/reverseproxy.go
+20
-24
@@ -157,6 +157,19 @@ type Handler struct {
157
// could be useful if the backend has tighter memory constraints.
158
ResponseBuffers int64 `json:"response_buffers,omitempty"`
159
160
+ // If nonzero, streaming requests such as WebSockets will be
161
+ // forcibly closed at the end of the timeout. Default: no timeout.
162
+ StreamTimeout caddy.Duration `json:"stream_timeout,omitempty"`
163
+
164
+ // If nonzero, streaming requests such as WebSockets will not be
165
+ // closed when the proxy config is unloaded, and instead the stream
166
+ // will remain open until the delay is complete. In other words,
167
+ // enabling this prevents streams from closing when Caddy's config
168
+ // is reloaded. Enabling this may be a good idea to avoid a thundering
169
+ // herd of reconnecting clients which had their connections closed
170
+ // by the previous config closing. Default: no delay.
171
+ StreamCloseDelay caddy.Duration `json:"stream_close_delay,omitempty"`
172
+
173
// If configured, rewrites the copy of the upstream request.
174
// Allows changing the request method and URI (path and query).
175
// Since the rewrite is applied to the copy, it does not persist
@@ -198,8 +211,9 @@ type Handler struct {
211
handleResponseSegments []*caddyfile.Dispenser
212
213
// Stores upgraded requests (hijacked connections) for proper cleanup
201
- connections map[io.ReadWriteCloser]openConnection
202
- connectionsMu *sync.Mutex
214
+ connections map[io.ReadWriteCloser]openConnection
215
+ connectionsCloseTimer *time.Timer
216
+ connectionsMu *sync.Mutex
217
218
ctx caddy.Context
219
logger *zap.Logger
@@ -382,25 +396,7 @@ func (h *Handler) Provision(ctx caddy.Context) error {
396
397
// Cleanup cleans up the resources made by h.
398
func (h *Handler) Cleanup() error {
385
- // close hijacked connections (both to client and backend)
386
- var err error
387
- h.connectionsMu.Lock()
388
- for _, oc := range h.connections {
389
- if oc.gracefulClose != nil {
390
- // this is potentially blocking while we have the lock on the connections
391
- // map, but that should be OK since the server has in theory shut down
392
- // and we are no longer using the connections map
393
- gracefulErr := oc.gracefulClose()
394
- if gracefulErr != nil && err == nil {
395
- err = gracefulErr
396
- }
397
- }
398
- closeErr := oc.conn.Close()
399
- if closeErr != nil && err == nil {
400
- err = closeErr
401
- }
402
- }
403
- h.connectionsMu.Unlock()
399
+ err := h.cleanupConnections()
400
401
// remove hosts from our config from the pool
402
for _, upstream := range h.Upstreams {
@@ -872,7 +868,7 @@ func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, origRe
868
repl.Set("http.reverse_proxy.status_code", res.StatusCode)
869
repl.Set("http.reverse_proxy.status_text", res.Status)
870
875
- h.logger.Debug("handling response", zap.Int("handler", i))
871
+ logger.Debug("handling response", zap.Int("handler", i))
872
873
// we make some data available via request context to child routes
874
// so that they may inherit some options and functions from the
@@ -917,7 +913,7 @@ func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, origRe
913
}
914
915
// finalizeResponse prepares and copies the response.
920
-func (h Handler) finalizeResponse(
916
+func (h *Handler) finalizeResponse(
917
rw http.ResponseWriter,
918
req *http.Request,
919
res *http.Response,
@@ -967,7 +963,7 @@ func (h Handler) finalizeResponse(
963
// there's nothing an error handler can do to recover at this point;
964
// the standard lib's proxy panics at this point, but we'll just log
965
// the error and abort the stream here
970
- h.logger.Error("aborting with incomplete response", zap.Error(err))
966
+ logger.Error("aborting with incomplete response", zap.Error(err))
967
return nil
968
}
969
modules/caddyhttp/reverseproxy/streaming.go
+81
-8
@@ -33,19 +33,19 @@ import (
33
"golang.org/x/net/http/httpguts"
34
)
35
36
-func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWriter, req *http.Request, res *http.Response) {
36
+func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWriter, req *http.Request, res *http.Response) {
37
reqUpType := upgradeType(req.Header)
38
resUpType := upgradeType(res.Header)
39
40
// Taken from https://github.com/golang/go/commit/5c489514bc5e61ad9b5b07bd7d8ec65d66a0512a
41
// We know reqUpType is ASCII, it's checked by the caller.
42
if !asciiIsPrint(resUpType) {
43
- h.logger.Debug("backend tried to switch to invalid protocol",
43
+ logger.Debug("backend tried to switch to invalid protocol",
44
zap.String("backend_upgrade", resUpType))
45
return
46
}
47
if !asciiEqualFold(reqUpType, resUpType) {
48
- h.logger.Debug("backend tried to switch to unexpected protocol via Upgrade header",
48
+ logger.Debug("backend tried to switch to unexpected protocol via Upgrade header",
49
zap.String("backend_upgrade", resUpType),
50
zap.String("requested_upgrade", reqUpType))
51
return
@@ -53,12 +53,12 @@ func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrite
53
54
hj, ok := rw.(http.Hijacker)
55
if !ok {
56
- h.logger.Sugar().Errorf("can't switch protocols using non-Hijacker ResponseWriter type %T", rw)
56
+ logger.Error("can't switch protocols using non-Hijacker ResponseWriter", zap.String("type", fmt.Sprintf("%T", rw)))
57
return
58
}
59
backConn, ok := res.Body.(io.ReadWriteCloser)
60
if !ok {
61
- h.logger.Error("internal error: 101 switching protocols response with non-writable body")
61
+ logger.Error("internal error: 101 switching protocols response with non-writable body")
62
return
63
}
64
@@ -83,7 +83,7 @@ func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrite
83
logger.Debug("upgrading connection")
84
conn, brw, err := hj.Hijack()
85
if err != nil {
86
- h.logger.Error("hijack failed on protocol switch", zap.Error(err))
86
+ logger.Error("hijack failed on protocol switch", zap.Error(err))
87
return
88
}
89
@@ -94,7 +94,7 @@ func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrite
94
}()
95
96
if err := brw.Flush(); err != nil {
97
- h.logger.Debug("response flush", zap.Error(err))
97
+ logger.Debug("response flush", zap.Error(err))
98
return
99
}
100
@@ -120,10 +120,23 @@ func (h Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrite
120
121
spc := switchProtocolCopier{user: conn, backend: backConn}
122
123
+ // setup the timeout if requested
124
+ var timeoutc <-chan time.Time
125
+ if h.StreamTimeout > 0 {
126
+ timer := time.NewTimer(time.Duration(h.StreamTimeout))
127
+ defer timer.Stop()
128
+ timeoutc = timer.C
129
+ }
130
+
131
errc := make(chan error, 1)
132
go spc.copyToBackend(errc)
133
go spc.copyFromBackend(errc)
126
- <-errc
134
+ select {
135
+ case err := <-errc:
136
+ logger.Debug("streaming error", zap.Error(err))
137
+ case time := <-timeoutc:
138
+ logger.Debug("stream timed out", zap.Time("timeout", time))
139
+ }
140
}
141
142
// flushInterval returns the p.FlushInterval value, conditionally
@@ -243,10 +256,70 @@ func (h *Handler) registerConnection(conn io.ReadWriteCloser, gracefulClose func
256
return func() {
257
h.connectionsMu.Lock()
258
delete(h.connections, conn)
259
+ // if there is no connection left before the connections close timer fires
260
+ if len(h.connections) == 0 && h.connectionsCloseTimer != nil {
261
+ // we release the timer that holds the reference to Handler
262
+ if (*h.connectionsCloseTimer).Stop() {
263
+ h.logger.Debug("stopped streaming connections close timer - all connections are already closed")
264
+ }
265
+ h.connectionsCloseTimer = nil
266
+ }
267
h.connectionsMu.Unlock()
268
}
269
}
270
271
+// closeConnections immediately closes all hijacked connections (both to client and backend).
272
+func (h *Handler) closeConnections() error {
273
+ var err error
274
+ h.connectionsMu.Lock()
275
+ defer h.connectionsMu.Unlock()
276
+
277
+ for _, oc := range h.connections {
278
+ if oc.gracefulClose != nil {
279
+ // this is potentially blocking while we have the lock on the connections
280
+ // map, but that should be OK since the server has in theory shut down
281
+ // and we are no longer using the connections map
282
+ gracefulErr := oc.gracefulClose()
283
+ if gracefulErr != nil && err == nil {
284
+ err = gracefulErr
285
+ }
286
+ }
287
+ closeErr := oc.conn.Close()
288
+ if closeErr != nil && err == nil {
289
+ err = closeErr
290
+ }
291
+ }
292
+ return err
293
+}
294
+
295
+// cleanupConnections closes hijacked connections.
296
+// Depending on the value of StreamCloseDelay it does that either immediately
297
+// or sets up a timer that will do that later.
298
+func (h *Handler) cleanupConnections() error {
299
+ if h.StreamCloseDelay == 0 {
300
+ return h.closeConnections()
301
+ }
302
+
303
+ h.connectionsMu.Lock()
304
+ defer h.connectionsMu.Unlock()
305
+ // the handler is shut down, no new connection can appear,
306
+ // so we can skip setting up the timer when there are no connections
307
+ if len(h.connections) > 0 {
308
+ delay := time.Duration(h.StreamCloseDelay)
309
+ h.connectionsCloseTimer = time.AfterFunc(delay, func() {
310
+ h.logger.Debug("closing streaming connections after delay",
311
+ zap.Duration("delay", delay))
312
+ err := h.closeConnections()
313
+ if err != nil {
314
+ h.logger.Error("failed to closed connections after delay",
315
+ zap.Error(err),
316
+ zap.Duration("delay", delay))
317
+ }
318
+ })
319
+ }
320
+ return nil
321
+}
322
+
323
// writeCloseControl sends a best-effort Close control message to the given
324
// WebSocket connection. Thanks to @pascaldekloe who provided inspiration
325
// from his simple implementation of this I was able to learn from at: