Log number of bytes read from HTTP request body

If Content-Length is not -1, report that in
http.request.body.bytes, otherwise report the
number of bytes read from the body.

changelogs/head.asciidoc

@@ -20,3 +20,4 @@ https://github.com/elastic/apm-server/compare/8.12\...main[View commits]
 ==== Added
 - map OTel's `span.Status: Unset` to `event.outcome: success` instead of `event.outcome: unknown`
 - Add support for Otel code.stacktrace {pull}12096[12096]
+- `http.request.body.bytes` now reports the bytes read from request body even if Content-Length is -1 {pull}TODO[TODO]

internal/beater/middleware/log_middleware.go

@@ -46,6 +46,7 @@ func LogMiddleware() Middleware {
 			}
 			h(c)
 			c.Logger = c.Logger.With("event.duration", time.Since(c.Timestamp))
+			c.Logger = c.Logger.With("http.request.body.bytes", c.RequestBodyBytes())
 			if c.MultipleWriteAttempts() {
 				c.Logger.Warn("multiple write attempts")
 			}
@@ -75,9 +76,6 @@ func loggerWithRequestContext(c *request.Context) *logp.Logger {
 	if c.ClientIP.IsValid() && c.ClientIP != c.SourceIP {
 		logger = logger.With("client.ip", c.ClientIP.String())
 	}
-	if c.Request.ContentLength != -1 {
-		logger = logger.With("http.request.body.bytes", c.Request.ContentLength)
-	}
 	return logger
 }
 

internal/beater/middleware/log_middleware_test.go

@@ -19,6 +19,8 @@ package middleware
 
 import (
 	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -136,3 +138,20 @@ func TestLogMiddleware(t *testing.T) {
 		})
 	}
 }
+
+func TestLogMiddlewareRequestBodyBytes(t *testing.T) {
+	configure.Logging("APM Server test", agentconfig.MustNewConfigFrom(`{"ecs":true}`))
+	require.NoError(t, logp.DevelopmentSetup(logp.ToObserverOutput()))
+
+	c := request.NewContext()
+	c.Reset(
+		httptest.NewRecorder(),
+		httptest.NewRequest(http.MethodGet, "/", strings.NewReader("content")),
+	)
+	Apply(LogMiddleware(), HandlerIdle)(c)
+
+	entries := logp.ObserverLogs().TakeAll()
+	require.Equal(t, 1, len(entries))
+	field := entries[0].ContextMap()["http.request.body.bytes"]
+	assert.Equal(t, int64(len("content")), field)
+}

internal/beater/request/context.go

@@ -54,6 +54,7 @@ type Context struct {
 	// compressedRequestReadCloser will be initialised for requests
 	// with a non-zero ContentLength and empty Content-Encoding.
 	compressedRequestReadCloser compressedRequestReadCloser
+	countingReadCloser          countingReadCloser
 	gzipReader                  *gzip.Reader
 	zlibReader                  zlibReadCloseResetter
 
@@ -157,6 +158,13 @@ func (c *Context) setRequest(r *http.Request) {
 		c.SourcePort, c.ClientPort = int(port), int(port)
 	}
 
+	if c.Request.ContentLength != -1 {
+		c.countingReadCloser.n = c.Request.ContentLength
+	} else {
+		c.countingReadCloser.ReadCloser = c.Request.Body
+		c.Request.Body = &c.countingReadCloser
+	}
+
 	if err := c.decodeRequestBody(); err != nil {
 		if c.Logger != nil {
 			c.Logger.Errorw("failed to decode request body", "error", err)
@@ -318,6 +326,29 @@ func (c *Context) resetGzip(r io.Reader) (io.ReadCloser, error) {
 	return c.gzipReader, err
 }
 
+// RequestBodyBytes returns the original c.Request.ContentLength if it
+// was not -1, otherwise it returns the number of bytes read from the
+// request body.
+//
+// RequestBodyBytes must not be called concurrently with
+// c.Request.Body.Read().
+func (c *Context) RequestBodyBytes() int64 {
+	return c.countingReadCloser.n
+}
+
+type countingReadCloser struct {
+	io.ReadCloser
+	n int64
+}
+
+func (r *countingReadCloser) Read(p []byte) (int, error) {
+	n, err := r.ReadCloser.Read(p)
+	if n > 0 {
+		r.n += int64(n)
+	}
+	return n, err
+}
+
 type compressedRequestReadCloser struct {
 	io.ReadCloser
 	magic     [2]byte

internal/beater/request/context_test.go

@@ -123,6 +123,8 @@ func TestContext_Reset(t *testing.T) {
 			assert.False(t, timestamp.Before(before))
 		case "compressedRequestReadCloser":
 			assert.Zero(t, c.compressedRequestReadCloser)
+		case "countingReadCloser":
+			assert.Zero(t, c.countingReadCloser)
 		case "zlibReader":
 			assert.Nil(t, c.zlibReader)
 		case "gzipReader":
@@ -251,6 +253,35 @@ func TestContextResetContentEncoding(t *testing.T) {
 	test("deflate_sniff", "", bytes.NewReader(deflateCompressed), "contents", "deflate")
 }
 
+func TestContextRequestBodyBytes(t *testing.T) {
+	t.Run("no_body", func(t *testing.T) {
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/", nil)
+		c := Context{Request: r, ResponseWriter: w}
+		c.Reset(w, r)
+		assert.Equal(t, int64(0), c.RequestBodyBytes())
+	})
+
+	t.Run("client_specified_content_length", func(t *testing.T) {
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/", strings.NewReader("contents"))
+		c := Context{Request: r, ResponseWriter: w}
+		c.Reset(w, r)
+		assert.Equal(t, int64(len("contents")), c.RequestBodyBytes())
+	})
+
+	t.Run("unspecified_content_length", func(t *testing.T) {
+		w := httptest.NewRecorder()
+		compressed := gzipCompressString("contents")
+		r := httptest.NewRequest(http.MethodGet, "/", bytes.NewReader(compressed))
+		r.ContentLength = -1
+		c := Context{Request: r, ResponseWriter: w}
+		c.Reset(w, r)
+		assertReaderContents(t, "contents", c.Request.Body)
+		assert.Equal(t, int64(len(compressed)), c.RequestBodyBytes())
+	})
+}
+
 func BenchmarkContextResetContentEncoding(b *testing.B) {
 	benchmark := func(name string, contentEncoding string, body io.ReadSeeker) {
 		w := httptest.NewRecorder()