Make errMaxStreamsLimitExceeded an error type

Author: maratkhv

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

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"strings"
@@ -16,18 +17,10 @@ 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")
 )
 
-func isErrMaxStreamsLimitExceeded(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	return strings.HasPrefix(err.Error(), "streams limit exceeded")
-}
-
 // SentDataMarkerHandler is a slice of the MarkerHandler interface, that the batch interacts with to report the event that
 // all data in the batch has been delivered or a client failed to do so.
 type SentDataMarkerHandler interface {
@@ -81,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{
@@ -106,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)
@@ -185,29 +185,3 @@ func BenchmarkLabelsMapToString(b *testing.B) {
 	}
 	result = r
 }
-
-func TestIsErrMaxStreamsLimitExceeded(t *testing.T) {
-	testCases := []struct {
-		name     string
-		err      error
-		expected bool
-	}{
-		{
-			name:     "error is max stream limit exceeded",
-			err:      fmt.Errorf(errMaxStreamsLimitExceeded, 1, 2, "foo bar"),
-			expected: true,
-		},
-		{
-			name:     "error is different",
-			err:      fmt.Errorf("some other error"),
-			expected: false,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			result := isErrMaxStreamsLimitExceeded(tc.err)
-			assert.Equal(t, tc.expected, result)
-		})
-	}
-}

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 isErrMaxStreamsLimitExceeded(err) {
+				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 isErrMaxStreamsLimitExceeded(err) {
+		if errors.Is(err, errMaxStreamsLimitExceeded) {
 			reason = ReasonStreamLimited
 		}
 		c.metrics.droppedBytes.WithLabelValues(c.cfg.URL.Host, tenantID, reason).Add(float64(len(e.Line)))