Fix streams limit error check so that metrics are correctly labeled as ReasonStreamLimited (#3466)

* fix: replace direct error string compare with isErrMaxStreamsLimitExceeded helper

* update CHANGELOG

* Make errMaxStreamsLimitExceeded an error type

Author: maratkhv

CHANGELOG.md

@@ -14,6 +14,8 @@ v1.8.3
 
 - Fix `mimir.rules.kubernetes` panic on non-leader debug info retrieval (@TheoBrigitte)
 
+- Fix detection of the “streams limit exceeded” error in the Loki client so that metrics are correctly labeled as `ReasonStreamLimited`. (@maratkhv)
+
 - Fix `loki.source.file` race condition that often lead to panic when using `decompression`. (@kalleep)
 
 - Fix deadlock in `loki.source.file` that can happen when targets are removed. (@kalleep)

internal/component/common/loki/client/batch.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -16,8 +17,8 @@ import (
 	"github.com/grafana/alloy/internal/component/common/loki"
 )
 
-const (
-	errMaxStreamsLimitExceeded = "streams limit exceeded, streams: %d exceeds limit: %d, stream: '%s'"
+var (
+	errMaxStreamsLimitExceeded = errors.New("streams limit exceeded")
 )
 
 // SentDataMarkerHandler is a slice of the MarkerHandler interface, that the batch interacts with to report the event that
@@ -73,7 +74,7 @@ func (b *batch) add(entry loki.Entry) error {
 
 	streams := len(b.streams)
 	if b.maxStreams > 0 && streams >= b.maxStreams {
-		return fmt.Errorf(errMaxStreamsLimitExceeded, streams, b.maxStreams, labels)
+		return fmt.Errorf("%w, streams: %d exceeds limit: %d, stream: '%s'", errMaxStreamsLimitExceeded, streams, b.maxStreams, labels)
 	}
 	// Add the entry as a new stream
 	b.streams[labels] = &logproto.Stream{
@@ -98,7 +99,7 @@ func (b *batch) addFromWAL(lbs model.LabelSet, entry logproto.Entry, segmentNum
 
 	streams := len(b.streams)
 	if b.maxStreams > 0 && streams >= b.maxStreams {
-		return fmt.Errorf(errMaxStreamsLimitExceeded, streams, b.maxStreams, labels)
+		return fmt.Errorf("%w, streams: %d exceeds limit: %d, stream: '%s'", errMaxStreamsLimitExceeded, streams, b.maxStreams, labels)
 	}
 
 	// Add the entry as a new stream

internal/component/common/loki/client/batch_test.go

@@ -31,7 +31,7 @@ func TestBatch_MaxStreams(t *testing.T) {
 		err := b.add(entry)
 		if err != nil {
 			errCount++
-			assert.EqualError(t, err, fmt.Errorf(errMaxStreamsLimitExceeded, len(b.streams), b.maxStreams, entry.Labels).Error())
+			assert.ErrorIs(t, err, errMaxStreamsLimitExceeded)
 		}
 	}
 	assert.Equal(t, errCount, 2)

internal/component/common/loki/client/client.go

@@ -310,7 +310,7 @@ func (c *client) run() {
 			if err != nil {
 				level.Error(c.logger).Log("msg", "batch add err", "tenant", tenantID, "error", err)
 				reason := ReasonGeneric
-				if err.Error() == errMaxStreamsLimitExceeded {
+				if errors.Is(err, errMaxStreamsLimitExceeded) {
 					reason = ReasonStreamLimited
 				}
 				c.metrics.droppedBytes.WithLabelValues(c.cfg.URL.Host, tenantID, reason).Add(float64(len(e.Line)))

internal/component/common/loki/client/queue_client.go

@@ -358,7 +358,7 @@ func (c *queueClient) appendSingleEntry(segmentNum int, lbs model.LabelSet, e lo
 	if err != nil {
 		level.Error(c.logger).Log("msg", "batch add err", "tenant", tenantID, "error", err)
 		reason := ReasonGeneric
-		if err.Error() == errMaxStreamsLimitExceeded {
+		if errors.Is(err, errMaxStreamsLimitExceeded) {
 			reason = ReasonStreamLimited
 		}
 		c.metrics.droppedBytes.WithLabelValues(c.cfg.URL.Host, tenantID, reason).Add(float64(len(e.Line)))