optimize memory usage by recycling v2 requests (#7131)

* optimize memory usage by recycling v2 requests

Signed-off-by: SungJin1212 <[email protected]>

* fix benchmark

Signed-off-by: SungJin1212 <[email protected]>

---------

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

CHANGELOG.md

@@ -11,6 +11,7 @@
 * [ENHANCEMENT] Alertmanager/Ruler: Introduce a user scanner to reduce the number of list calls to object storage. #6999
 * [ENHANCEMENT] Ruler: Add DecodingConcurrency config flag for Thanos Engine. #7118
 * [ENHANCEMENT] Compactor: Avoid double compaction by cleaning partition files in 2 cycles. #7129
+* [ENHANCEMENT] Distributor: Optimize memory usage by recycling v2 requests. #7131
 * [BUGFIX] Ring: Change DynamoDB KV to retry indefinitely for WatchKey. #7088
 * [BUGFIX] Ruler: Add XFunctions validation support. #7111
 * [BUGFIX] Distributor: Fix panic on health check failure when using stream push. #7116

pkg/util/push/push.go

@@ -79,6 +79,12 @@ func Handler(remoteWrite2Enabled bool, maxRecvMsgSize int, sourceIPs *middleware
 
 		handlePRW2 := func() {
 			var req cortexpb.PreallocWriteRequestV2
+			// v1 request is put back into the pool by the Distributor.
+			defer func() {
+				cortexpb.ReuseWriteRequestV2(&req)
+				req.Free()
+			}()
+
 			err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, &req, util.RawSnappy)
 			if err != nil {
 				level.Error(logger).Log("err", err.Error())

pkg/util/push/push_test.go

@@ -141,7 +141,7 @@ func Benchmark_Handler(b *testing.B) {
 			for b.Loop() {
 				resp := httptest.NewRecorder()
 				handler.ServeHTTP(resp, req)
-				assert.Equal(b, http.StatusOK, resp.Code)
+				assert.Equal(b, http.StatusNoContent, resp.Code)
 				req.Body.(*resetReader).Reset()
 			}
 		})