Reduce distributor memory usage when error volume is high (#6095)

Signed-off-by: Xiaochao Dong (@damnever) <the.xcdong@gmail.com>
Signed-off-by: Daniel Blando <daniel@blando.com.br>
Co-authored-by: Daniel Blando <daniel@blando.com.br>

Author: damnever

CHANGELOG.md

@@ -25,6 +25,7 @@
 * [ENHANCEMENT] Ingester: Add link to renew 10% of the ingesters tokens in the admin page. #6063
 * [ENHANCEMENT] Ruler: Add support for filtering by `state` and `health` field on Rules API. #6040
 * [ENHANCEMENT] Ruler: Add support for filtering by `match` field on Rules API. #6083
+* [ENHANCEMENT] Distributor: Reduce memory usage when error volume is high. #6095
 * [ENHANCEMENT] Compactor: Add unique execution ID for each compaction cycle in log for easy debugging. #6097
 * [BUGFIX] Configsdb: Fix endline issue in db password. #5920
 * [BUGFIX] Ingester: Fix `user` and `type` labels for the `cortex_ingester_tsdb_head_samples_appended_total` TSDB metric. #5952

pkg/ring/batch.go

@@ -152,8 +152,7 @@ func (b *batchTracker) record(instance instance, err error) {
 		if err != nil {
 			// Track the number of errors by error family, and if it exceeds maxFailures
 			// shortcut the waiting rpc.
-			wrappedErr := httpgrpcutil.WrapHTTPGrpcError(err, "addr=%s state=%s zone=%s", instance.desc.Addr, instance.desc.State, instance.desc.Zone)
-			errCount := instance.itemTrackers[i].recordError(wrappedErr)
+			errCount := instance.itemTrackers[i].recordError(err)
 			// We should return an error if we reach the maxFailure (quorum) on a given error family OR
 			// we dont have any remaining ingesters to try
 			// Ex: 2xx, 4xx, 5xx -> return 4xx
@@ -162,13 +161,17 @@ func (b *batchTracker) record(instance instance, err error) {
 			// Ex: 5xx, _, 5xx -> return 5xx
 			if errCount > int32(sampleTrackers[i].maxFailures) {
 				if b.rpcsFailed.Inc() == 1 {
-					b.err <- httpgrpcutil.WrapHTTPGrpcError(sampleTrackers[i].getError(), "maxFailure (quorum) on a given error family")
+					b.err <- httpgrpcutil.WrapHTTPGrpcError(
+						sampleTrackers[i].getError(), "maxFailure (quorum) on a given error family, addr=%s state=%s zone=%s",
+						instance.desc.Addr, instance.desc.State, instance.desc.Zone)
 				}
 				continue
 			}
 			if sampleTrackers[i].remaining.Dec() == 0 {
 				if b.rpcsFailed.Inc() == 1 {
-					b.err <- httpgrpcutil.WrapHTTPGrpcError(sampleTrackers[i].getError(), "not enough remaining instances to try")
+					b.err <- httpgrpcutil.WrapHTTPGrpcError(
+						sampleTrackers[i].getError(), "not enough remaining instances to try, addr=%s state=%s zone=%s",
+						instance.desc.Addr, instance.desc.State, instance.desc.Zone)
 				}
 				continue
 			}
@@ -187,7 +190,9 @@ func (b *batchTracker) record(instance instance, err error) {
 			// Ex: 4xx, 5xx, 2xx
 			if sampleTrackers[i].remaining.Dec() == 0 {
 				if b.rpcsFailed.Inc() == 1 {
-					b.err <- httpgrpcutil.WrapHTTPGrpcError(sampleTrackers[i].getError(), "not enough remaining instances to try")
+					b.err <- httpgrpcutil.WrapHTTPGrpcError(
+						sampleTrackers[i].getError(), "not enough remaining instances to try, addr=%s state=%s zone=%s",
+						instance.desc.Addr, instance.desc.State, instance.desc.Zone)
 				}
 			}
 		}