optimize logBody to not buffer and pass x-flow-id to the log, such th…

…at it can be searched in log analysis

Signed-off-by: Sandor Szücs <[email protected]>

filters/diag/logbody.go

@@ -2,9 +2,11 @@ package diag
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"github.com/zalando/skipper/filters"
+	"github.com/zalando/skipper/filters/flowid"
 	"github.com/zalando/skipper/net"
 )
 
@@ -59,13 +61,13 @@ func (lb logBody) Request(ctx filters.FilterContext) {
 
 	req := ctx.Request()
 	if req.Body != nil {
-		req.Body = net.WrapBody(
-			req.Context(),
-			func(p []byte) (int, error) {
-				ctx.Logger().Infof(`logBody("request"): %q`, p)
-				return len(p), nil
-			},
-			req.Body)
+		req.Body = net.LogBody(
+			context.Background(),
+			fmt.Sprintf(`logBody("request") %s: `, req.Header.Get(flowid.HeaderName)),
+			ctx.Logger().Infof,
+			req.Body,
+		)
+
 	}
 }
 
@@ -78,14 +80,14 @@ func (lb logBody) Response(ctx filters.FilterContext) {
 	if rsp.Body != nil {
 		// if this is not set we get from curl
 		//    Error while processing content unencoding: invalid stored block lengths
-		rsp.Header.Del("Content-Length")
-		rsp.ContentLength = -1
-
-		rsp.Body = net.WrapBody(
-			context.Background(), // not sure if it makes sense to be cancellable here
-			func(p []byte) (int, error) {
-				ctx.Logger().Infof(`logBody("response"): %q`, p)
-				return len(p), nil
-			}, rsp.Body)
+		// rsp.Header.Del("Content-Length")
+		// rsp.ContentLength = -1
+
+		rsp.Body = net.LogBody(
+			context.Background(),
+			fmt.Sprintf(`logBody("response") %s: `, ctx.Request().Header.Get(flowid.HeaderName)),
+			ctx.Logger().Infof,
+			rsp.Body,
+		)
 	}
 }

net/httpbody.go

@@ -26,6 +26,38 @@ const (
 	MaxBufferAbort
 )
 
+type logBody struct {
+	ctx    context.Context
+	fmtstr string
+	log    func(format string, args ...interface{})
+	input  io.ReadCloser
+}
+
+func newLogBody(ctx context.Context, fmtstr string, log func(format string, args ...interface{}), rc io.ReadCloser) io.ReadCloser {
+	return &logBody{
+		ctx:    ctx,
+		fmtstr: fmtstr,
+		input:  rc,
+		log:    log,
+	}
+}
+
+func (lb *logBody) Read(p []byte) (int, error) {
+	n, err := lb.input.Read(p)
+	if n > 0 {
+		lb.log("%s%s", lb.fmtstr, p)
+	}
+	return n, err
+}
+
+func (lb *logBody) Close() error {
+	return lb.input.Close()
+}
+
+func LogBody(ctx context.Context, fmtstr string, log func(format string, args ...interface{}), rc io.ReadCloser) io.ReadCloser {
+	return newLogBody(ctx, fmtstr, log, rc)
+}
+
 type matcher struct {
 	ctx               context.Context
 	once              sync.Once