diff --git a/modules/caddyhttp/reverseproxy/reverseproxy.go b/modules/caddyhttp/reverseproxy/reverseproxy.go index 55d3aa8c..40569761 100644 --- a/modules/caddyhttp/reverseproxy/reverseproxy.go +++ b/modules/caddyhttp/reverseproxy/reverseproxy.go @@ -977,9 +977,11 @@ func (h Handler) finalizeResponse( } rw.WriteHeader(res.StatusCode) + logger.Debug("wrote header") - err := h.copyResponse(rw, res.Body, h.flushInterval(req, res)) - res.Body.Close() // close now, instead of defer, to populate res.Trailer + err := h.copyResponse(rw, res.Body, h.flushInterval(req, res), logger) + errClose := res.Body.Close() // close now, instead of defer, to populate res.Trailer + logger.Debug("closed response body from upstream", zap.Error(errClose)) if err != nil { // we're streaming the response and we've already written headers, so // there's nothing an error handler can do to recover at this point; @@ -1014,6 +1016,8 @@ func (h Handler) finalizeResponse( } } + logger.Debug("response finalized") + return nil } diff --git a/modules/caddyhttp/reverseproxy/streaming.go b/modules/caddyhttp/reverseproxy/streaming.go index 01d865db..7805a76b 100644 --- a/modules/caddyhttp/reverseproxy/streaming.go +++ b/modules/caddyhttp/reverseproxy/streaming.go @@ -166,12 +166,13 @@ func (h Handler) isBidirectionalStream(req *http.Request, res *http.Response) bo (ae == "identity" || ae == "") } -func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration) error { +func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.Duration, logger *zap.Logger) error { if flushInterval != 0 { if wf, ok := dst.(writeFlusher); ok { mlw := &maxLatencyWriter{ dst: wf, latency: flushInterval, + logger: logger.Named("max_latency_writer"), } defer mlw.stop() @@ -185,19 +186,22 @@ func (h Handler) copyResponse(dst io.Writer, src io.Reader, flushInterval time.D buf := streamingBufPool.Get().(*[]byte) defer streamingBufPool.Put(buf) - _, err := h.copyBuffer(dst, src, *buf) + _, err := h.copyBuffer(dst, src, *buf, logger) return err } // copyBuffer returns any write errors or non-EOF read errors, and the amount // of bytes written. -func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, error) { +func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte, logger *zap.Logger) (int64, error) { if len(buf) == 0 { buf = make([]byte, defaultBufferSize) } var written int64 for { + logger.Debug("waiting to read from upstream") nr, rerr := src.Read(buf) + logger := logger.With(zap.Int("read", nr)) + logger.Debug("read from upstream", zap.Error(rerr)) if rerr != nil && rerr != io.EOF && rerr != context.Canceled { // TODO: this could be useful to know (indeed, it revealed an error in our // fastcgi PoC earlier; but it's this single error report here that necessitates @@ -209,10 +213,15 @@ func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte) (int64, er h.logger.Error("reading from backend", zap.Error(rerr)) } if nr > 0 { + logger.Debug("writing to downstream") nw, werr := dst.Write(buf[:nr]) if nw > 0 { written += int64(nw) } + logger.Debug("wrote to downstream", + zap.Int("written", nw), + zap.Int64("written_total", written), + zap.Error(werr)) if werr != nil { return written, werr } @@ -295,17 +304,21 @@ type maxLatencyWriter struct { mu sync.Mutex // protects t, flushPending, and dst.Flush t *time.Timer flushPending bool + logger *zap.Logger } func (m *maxLatencyWriter) Write(p []byte) (n int, err error) { m.mu.Lock() defer m.mu.Unlock() n, err = m.dst.Write(p) + m.logger.Debug("wrote bytes", zap.Int("n", n), zap.Error(err)) if m.latency < 0 { + m.logger.Debug("flushing immediately") m.dst.Flush() return } if m.flushPending { + m.logger.Debug("delayed flush already pending") return } if m.t == nil { @@ -313,6 +326,7 @@ func (m *maxLatencyWriter) Write(p []byte) (n int, err error) { } else { m.t.Reset(m.latency) } + m.logger.Debug("timer set for delayed flush", zap.Duration("duration", m.latency)) m.flushPending = true return } @@ -321,8 +335,10 @@ func (m *maxLatencyWriter) delayedFlush() { m.mu.Lock() defer m.mu.Unlock() if !m.flushPending { // if stop was called but AfterFunc already started this goroutine + m.logger.Debug("delayed flush is not pending") return } + m.logger.Debug("delayed flush") m.dst.Flush() m.flushPending = false } diff --git a/modules/caddyhttp/reverseproxy/streaming_test.go b/modules/caddyhttp/reverseproxy/streaming_test.go index 4ed1f1e4..c5d933ae 100644 --- a/modules/caddyhttp/reverseproxy/streaming_test.go +++ b/modules/caddyhttp/reverseproxy/streaming_test.go @@ -4,6 +4,8 @@ import ( "bytes" "strings" "testing" + + "github.com/caddyserver/caddy/v2" ) func TestHandlerCopyResponse(t *testing.T) { @@ -18,7 +20,7 @@ func TestHandlerCopyResponse(t *testing.T) { for _, d := range testdata { src := bytes.NewBuffer([]byte(d)) dst.Reset() - err := h.copyResponse(dst, src, 0) + err := h.copyResponse(dst, src, 0, caddy.Log()) if err != nil { t.Errorf("failed with error: %v", err) }