OTLP ingestion: oversize gzip/uncompressed bodies are truncated instead of rejected, and gzip bodies arriving over httpgrpc are ignored #7900

AI Tool Usage Notice
If you used an AI tool to help draft this issue,
please make sure you have reviewed and validated all content before submitting.
You are responsible for the accuracy and quality of everything in this report.
Low-quality or unreviewed AI-generated submissions may be closed without further investigation.
See our Generative AI Contribution Policy for details.

Describe the bug
util.ParseProtoReader doesn't reject a body larger than maxSize when it reads from a stream (an HTTP request body) with NoCompression or Gzip. decompressFromReader reads at most maxSize+1 bytes through an io.LimitReader and returns them with a nil error (http.go#L200-L231), and the size check in decompressRequest only runs when an error has already occurred (if err != nil && len(body) > maxSize, http.go#L182-L198). The truncated bytes are then unmarshalled, so:

• most of the time the client gets unexpected EOF instead of received message larger than max (N vs M), which doesn't tell it which limit it hit;
• when the cut lands on a message boundary, the request succeeds and everything after the cut is dropped silently;
• a body of exactly maxSize+1 bytes is accepted.

OTLP ingestion is the only production caller that uses these compression types (otlp.go#L140-L166). It rejects Content-Length above -distributor.otlp-max-recv-msg-size up front, so this affects gzip requests whose decompressed size is over the limit (the cap from #7515 bounds memory, but the overflow isn't reported) and uncompressed requests sent without a Content-Length. The OTLP JSON branch truncates the same way (otlp.go#L167-L190), although truncated JSON fails to parse instead of being partially accepted. Remote write and remote read use RawSnappy, which decompressFromBuffer checks correctly.

A second, related problem is on the buffered path. When the reader exposes BytesBuffer(), ParseProtoReader uses decompressFromBuffer, which only handles NoCompression and RawSnappy and otherwise returns nil, nil (http.go#L233-L259), so a gzip body is unmarshalled as an empty request and accepted. Request bodies served by the httpgrpc server implement BytesBuffer(), and the server package registers that service on every component's gRPC server with the full HTTP router (server.go#L523), so a gzip OTLP push that reaches the distributor over httpgrpc returns success with nothing ingested. The Gzip case was added to decompressFromReader in #6333 but never to decompressFromBuffer.

To Reproduce
Steps to reproduce the behavior:

1. Start Cortex (SHA or version): master at df49fe6.
2. Perform Operations(Read/Write/Others): Write. Send an OTLP protobuf request to /api/v1/otlp/v1/metrics with Content-Encoding: gzip whose compressed size is below -distributor.otlp-max-recv-msg-size and whose decompressed size is above it. Depending on where the cut lands, the request fails with unexpected EOF or is accepted with only its leading part.

The same at the unit level with the test below: a stream reader (so ParseProtoReader takes the decompressFromReader path), 100 series that are 5303 bytes raw and 329 bytes gzipped, and every maxSize from 329 to 5302:

NoCompression chunked: maxSize=370 accepted with err=nil and 7 of 100 series
NoCompression chunked: over 4974 oversize limits: silent partial=95, size error=0, other error=4879 (e.g. unexpected EOF)
Gzip: maxSize=370 accepted with err=nil and 7 of 100 series
Gzip: over 4974 oversize limits: silent partial=95, size error=0, other error=4879 (e.g. unexpected EOF)

For the buffered path, the same one-series request gzipped and passed through a reader with a BytesBuffer() method (like the httpgrpc server's body) gives:

gzip via BytesBuffer reader: err=<nil> series=0
gzip via stream reader:      err=<nil> series=1

package util_test

import (
	"bytes"
	"compress/gzip"
	"context"
	"fmt"
	"io"
	"testing"

	"github.com/cortexproject/cortex/pkg/cortexpb"
	"github.com/cortexproject/cortex/pkg/util"
)

// streamReader hides *bytes.Reader so ParseProtoReader takes the decompressFromReader path, like an HTTP body.
type streamReader struct{ r io.Reader }

func (s streamReader) Read(p []byte) (int, error) { return s.r.Read(p) }

func TestOversizeBodyIsTruncated(t *testing.T) {
	var req cortexpb.WriteRequest
	for i := 0; i < 100; i++ {
		req.Timeseries = append(req.Timeseries, cortexpb.PreallocTimeseries{TimeSeries: &cortexpb.TimeSeries{
			Labels:  []cortexpb.LabelAdapter{{Name: "__name__", Value: fmt.Sprintf("metric_%03d", i)}, {Name: "job", Value: "probe"}},
			Samples: []cortexpb.Sample{{Value: 1, TimestampMs: 1}},
		}})
	}
	body, _ := req.Marshal()
	var gz bytes.Buffer
	w := gzip.NewWriter(&gz)
	_, _ = w.Write(body)
	_ = w.Close()

	for _, c := range []struct {
		name string
		comp util.CompressionType
		data []byte
		exp  int
	}{{"NoCompression chunked", util.NoCompression, body, -1}, {"Gzip", util.Gzip, gz.Bytes(), gz.Len()}} {
		silent, sizeErr, otherErr, firstSilent := 0, 0, 0, -1
		var firstOther error
		for maxSize := gz.Len(); maxSize < len(body); maxSize++ {
			var got cortexpb.WriteRequest
			err := util.ParseProtoReader(context.Background(), streamReader{bytes.NewReader(c.data)}, c.exp, maxSize, &got, c.comp)
			switch {
			case err == nil:
				silent++
				if firstSilent < 0 {
					firstSilent = maxSize
					t.Logf("%s: maxSize=%d accepted with err=nil and %d of 100 series", c.name, maxSize, len(got.Timeseries))
				}
			case bytes.Contains([]byte(err.Error()), []byte("larger than max")):
				sizeErr++
			default:
				otherErr++
				if firstOther == nil {
					firstOther = err
				}
			}
		}
		t.Logf("%s: over %d oversize limits: silent partial=%d, size error=%d, other error=%d (e.g. %v)", c.name, len(body)-gz.Len(), silent, sizeErr, otherErr, firstOther)
	}
}

Expected behavior
A body whose (decompressed) size exceeds maxSize is rejected with received message larger than max (...) and nothing from it is ingested, regardless of the compression and of whether the reader is a buffer. Checking len(body) > maxSize unconditionally after reading (e.g. dropping the err != nil condition from the deferred check in decompressRequest, and adding the same check to the OTLP JSON branch) would do it. A gzip body on the buffered path should be decompressed (with the same cap) like on the stream path.

Environment:

• Infrastructure: not environment-specific (reproduced with a unit test on master)
• Deployment tool: n/a

Additional Context
The conditional size check dates from #3682, so uncompressed streamed bodies have been truncated this way since then; #7515 added the same maxSize+1 cap to the decompressed gzip output, and its regression test (TestParseProtoReader_GzipDecompressionBomb) expects an oversize body to end in a proto unmarshal error rather than a size error. #7581 and #7655 are about making that test stricter; neither covers the case where the cut lands on a message boundary and the request is accepted.