Distributor: reject oversize decompressed request bodies instead of truncating them

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [ENHANCEMENT] Ruler: Add rule information to query parameters sent to the Query Frontend when `-ruler.frontend-address` is configured, so the Query Frontend query stats logs (`-frontend.enabled-ruler-query-stats`) can identify the evaluated rule. In this mode, the Ruler query stats log (`-ruler.query-stats-enabled`) omits the fetched statistics fields, which are always zero. #6539
 * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873
 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889
+* [BUGFIX] Distributor: Reject OTLP request bodies whose decompressed size exceeds `-distributor.otlp-max-recv-msg-size` instead of silently truncating them, and decode gzip-compressed OTLP protobuf bodies received over httpgrpc instead of treating them as empty. #7902
 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877
 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869
 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880

pkg/util/http.go

@@ -181,8 +181,12 @@ func ParseProtoReader(ctx context.Context, reader io.Reader, expectedSize, maxSi
 
 func decompressRequest(reader io.Reader, expectedSize, maxSize int, compression CompressionType, sp opentracing.Span) (body []byte, err error) {
 	defer func() {
-		if err != nil && len(body) > maxSize {
+		// The readers are capped at maxSize+1 bytes, so a body longer than
+		// maxSize was truncated and must be rejected even when reading it
+		// did not fail.
+		if len(body) > maxSize {
 			err = fmt.Errorf(messageSizeLargerErrFmt, len(body), maxSize)
+			body = nil
 		}
 	}()
 	if expectedSize > maxSize {
@@ -254,6 +258,14 @@ func decompressFromBuffer(buffer *bytes.Buffer, maxSize int, compression Compres
 			return nil, err
 		}
 		return body, nil
+	case Gzip:
+		gzReader, err := gzip.NewReader(bytes.NewReader(buffer.Bytes()))
+		if err != nil {
+			return nil, err
+		}
+		var buf bytes.Buffer
+		_, err = buf.ReadFrom(io.LimitReader(gzReader, int64(maxSize)+1))
+		return buf.Bytes(), err
 	}
 	return nil, nil
 }

pkg/util/http_test.go

@@ -237,8 +237,68 @@ func TestParseProtoReader_GzipDecompressionBomb(t *testing.T) {
 
 	var fromWire cortexpb.PreallocWriteRequest
 	err = util.ParseProtoReader(context.Background(), io.NopCloser(&compressed), 0, maxSize, &fromWire, util.Gzip)
-	// The decompressed output should be limited to maxSize+1 bytes, causing a
-	// proto unmarshal error (not an OOM). The key assertion is that we don't
-	// allocate 1 MB of memory.
-	assert.NotNil(t, err)
+	// The decompressed output should be limited to maxSize+1 bytes (not an
+	// OOM) and the request rejected as too large.
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "received message larger than max")
+}
+
+// streamReader hides the underlying reader's type, so ParseProtoReader reads it
+// like an HTTP request body instead of using the BytesBuffer() fast path.
+type streamReader struct{ r io.Reader }
+
+func (s streamReader) Read(p []byte) (int, error) { return s.r.Read(p) }
+
+func TestParseProtoReader_OversizeBodyIsRejected(t *testing.T) {
+	req := cortexpb.PreallocWriteRequest{}
+	for i := range 100 {
+		req.Timeseries = append(req.Timeseries, cortexpb.PreallocTimeseries{TimeSeries: &cortexpb.TimeSeries{
+			Labels:  []cortexpb.LabelAdapter{{Name: "__name__", Value: "metric_" + strconv.Itoa(i)}, {Name: "job", Value: "probe"}},
+			Samples: []cortexpb.Sample{{Value: 1, TimestampMs: 1}},
+		}})
+	}
+	body, err := req.Marshal()
+	require.NoError(t, err)
+
+	var gz bytes.Buffer
+	gzw := gzip.NewWriter(&gz)
+	_, err = gzw.Write(body)
+	require.NoError(t, err)
+	require.NoError(t, gzw.Close())
+	require.Less(t, gz.Len(), len(body))
+
+	for _, tc := range []struct {
+		name           string
+		compression    util.CompressionType
+		data           []byte
+		useBytesBuffer bool
+	}{
+		{"noCompression", util.NoCompression, body, false},
+		{"gzip", util.Gzip, gz.Bytes(), false},
+		{"bytesbuffer gzip", util.Gzip, gz.Bytes(), true},
+	} {
+		newReader := func() io.Reader {
+			if tc.useBytesBuffer {
+				return bytesBuffered{Buffer: bytes.NewBuffer(append([]byte(nil), tc.data...))}
+			}
+			return streamReader{bytes.NewReader(tc.data)}
+		}
+
+		t.Run(tc.name+" within limit", func(t *testing.T) {
+			var fromWire cortexpb.PreallocWriteRequest
+			require.NoError(t, util.ParseProtoReader(context.Background(), newReader(), 0, len(body), &fromWire, tc.compression))
+			require.Len(t, fromWire.Timeseries, 100)
+		})
+
+		t.Run(tc.name+" over limit", func(t *testing.T) {
+			// Every limit below the decoded size must be reported as a size error,
+			// whether or not the cut lands on a message boundary.
+			for maxSize := gz.Len(); maxSize < len(body); maxSize++ {
+				var fromWire cortexpb.PreallocWriteRequest
+				err := util.ParseProtoReader(context.Background(), newReader(), 0, maxSize, &fromWire, tc.compression)
+				require.Error(t, err, "maxSize=%d", maxSize)
+				require.Contains(t, err.Error(), "received message larger than max", "maxSize=%d", maxSize)
+			}
+		})
+	}
 }

pkg/util/push/otlp.go

@@ -186,6 +186,10 @@ func decodeOTLPWriteRequest(ctx context.Context, r *http.Request, maxSize int) (
 			if err != nil {
 				return req, err
 			}
+			// The readers are capped at maxSize+1 bytes, so a longer body was truncated.
+			if buf.Len() > maxSize {
+				return req, fmt.Errorf("received message larger than max (%d vs %d)", buf.Len(), maxSize)
+			}
 
 			return req, req.UnmarshalJSON(buf.Bytes())
 		}

pkg/util/push/otlp_test.go

@@ -1188,6 +1188,47 @@ func TestOTLPWriteHandler(t *testing.T) {
 	}
 }
 
+func TestOTLPWriteHandler_OversizeBodyIsRejected(t *testing.T) {
+	cfg := distributor.OTLPConfig{}
+	exportRequest := generateOTLPWriteRequest()
+
+	for _, contentType := range []string{pbContentType, jsonContentType} {
+		for _, encodingType := range []string{"", "gzip"} {
+			t.Run(contentType+" "+encodingType, func(t *testing.T) {
+				plain, err := getOTLPHttpRequest(&exportRequest, contentType, "")
+				require.NoError(t, err)
+				decodedSize := int(plain.ContentLength)
+
+				// The body is either sent without a Content-Length (uncompressed)
+				// or compressed below the limit, so only its decoded size is too large.
+				maxRecvMsgSize := decodedSize - 1
+				req, err := getOTLPHttpRequest(&exportRequest, contentType, encodingType)
+				require.NoError(t, err)
+				if encodingType == "" {
+					req.ContentLength = -1
+				}
+				require.Less(t, int(req.ContentLength), maxRecvMsgSize)
+
+				push := func(context.Context, *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) {
+					t.Error("an oversize request must not be pushed")
+					return &cortexpb.WriteResponse{}, nil
+				}
+				overrides := validation.NewOverrides(querier.DefaultLimitsConfig(), nil)
+				handler := OTLPHandler(maxRecvMsgSize, overrides, cfg, nil, push, nil)
+
+				recorder := httptest.NewRecorder()
+				handler.ServeHTTP(recorder, req)
+
+				resp := recorder.Result()
+				require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+				b, err := io.ReadAll(resp.Body)
+				require.NoError(t, err)
+				require.Contains(t, string(b), "received message larger than max")
+			})
+		}
+	}
+}
+
 func generateOTLPWriteRequestWithSeries(numSeries, samplesPerSeries, numHistogram int) pmetricotlp.ExportRequest {
 	d := pmetric.NewMetrics()