Improve error message when max error message is hit (#6834)

before:  `entry with timestamp 2022-08-02 12:32:37 +0000 UTC ignored, reason: 'entry too far behind' for stream: {foo="bar2"}`
after: `entry with timestamp 2022-08-02 12:32:37 +0000 UTC ignored, reason: 'entry too far behind, oldest acceptable timestamp is: 2022-08-03T15:25:57Z' for stream: {foo="bar2"}`

pkg/chunkenc/interface.go

@@ -18,14 +18,30 @@ import (
 var (
 	ErrChunkFull       = errors.New("chunk full")
 	ErrOutOfOrder      = errors.New("entry out of order")
-	ErrTooFarBehind    = errors.New("entry too far behind")
 	ErrInvalidSize     = errors.New("invalid size")
 	ErrInvalidFlag     = errors.New("invalid flag")
 	ErrInvalidChecksum = errors.New("invalid chunk checksum")
 )
 
+type errTooFarBehind struct {
+	cutoff time.Time
+}
+
+func IsErrTooFarBehind(err error) bool {
+	_, ok := err.(*errTooFarBehind)
+	return ok
+}
+
+func ErrTooFarBehind(cutoff time.Time) error {
+	return &errTooFarBehind{cutoff: cutoff}
+}
+
+func (m *errTooFarBehind) Error() string {
+	return "entry too far behind, oldest acceptable timestamp is: " + m.cutoff.Format(time.RFC3339)
+}
+
 func IsOutOfOrderErr(err error) bool {
-	return err == ErrOutOfOrder || err == ErrTooFarBehind
+	return err == ErrOutOfOrder || IsErrTooFarBehind(err)
 }
 
 // Encoding is the identifier for a chunk encoding.

pkg/chunkenc/interface_test.go

@@ -2,6 +2,7 @@ package chunkenc
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -30,7 +31,7 @@ func TestParseEncoding(t *testing.T) {
 }
 
 func TestIsOutOfOrderErr(t *testing.T) {
-	for _, err := range []error{ErrOutOfOrder, ErrTooFarBehind} {
+	for _, err := range []error{ErrOutOfOrder, ErrTooFarBehind(time.Now())} {
 		require.Equal(t, true, IsOutOfOrderErr(err))
 	}
 }

pkg/ingester/stream.go

@@ -230,8 +230,9 @@ func (s *stream) Push(
 		}
 
 		// The validity window for unordered writes is the highest timestamp present minus 1/2 * max-chunk-age.
-		if !isReplay && s.unorderedWrites && !s.highestTs.IsZero() && s.highestTs.Add(-s.cfg.MaxChunkAge/2).After(entries[i].Timestamp) {
-			failedEntriesWithError = append(failedEntriesWithError, entryWithError{&entries[i], chunkenc.ErrTooFarBehind})
+		cutoff := s.highestTs.Add(-s.cfg.MaxChunkAge / 2)
+		if !isReplay && s.unorderedWrites && !s.highestTs.IsZero() && cutoff.After(entries[i].Timestamp) {
+			failedEntriesWithError = append(failedEntriesWithError, entryWithError{&entries[i], chunkenc.ErrTooFarBehind(cutoff)})
 			outOfOrderSamples++
 			outOfOrderBytes += len(entries[i].Line)
 		} else if err := chunk.chunk.Append(&entries[i]); err != nil {

pkg/ingester/stream_test.go

@@ -77,8 +77,9 @@ func TestMaxReturnedStreamsErrors(t *testing.T) {
 			var expected bytes.Buffer
 			for i := 0; i < tc.expectErrs; i++ {
 				fmt.Fprintf(&expected,
-					"entry with timestamp %s ignored, reason: 'entry too far behind' for stream: {foo=\"bar\"},\n",
+					"entry with timestamp %s ignored, reason: 'entry too far behind, oldest acceptable timestamp is: %s' for stream: {foo=\"bar\"},\n",
 					time.Unix(int64(i), 0).String(),
+					time.Unix(int64(numLogs), 0).Format(time.RFC3339),
 				)
 			}