diff --git a/CHANGELOG.md b/CHANGELOG.md index 5253620469..47dbba996e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,7 @@ * [ENHANCEMENT] Update prometheus alertmanager version to v0.33.0. #7647 * [ENHANCEMENT] Querier/Ingester: Detach ingester series from gRPC buffers to reduce heap. #7670 * [ENHANCEMENT] Ingester: Add `cortex_ingester_tsdb_head_max_timestamp` metric that re-exports the TSDB head max timestamp (`prometheus_tsdb_head_max_time`) per user, to help investigate ingestion issues like out-of-bounds (too old sample) errors. #7694 +* [ENHANCEMENT] Ingester: Include the TSDB head max time in the `out of bounds` and `too old sample` error messages, so that users can see how far behind the accepted time range a rejected sample is. #7695 * [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370 * [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380 * [BUGFIX] Distributor: Return HTTP 401 Unauthorized when tenant ID resolution fails in the Prometheus Remote Write 2.0 path. #7389 diff --git a/pkg/ingester/ingester.go b/pkg/ingester/ingester.go index 2d813ff91c..d35c8b8dd9 100644 --- a/pkg/ingester/ingester.go +++ b/pkg/ingester/ingester.go @@ -81,6 +81,7 @@ const ( errTSDBCreateIncompatibleState = "cannot create a new TSDB while the ingester is not in active state (current state: %s)" errTSDBIngestWithTimestamp = "err: %v. series=%s" // Using error.Wrap puts the message before the error and if the series is too long, its truncated. errTSDBIngest = "err: %v. timestamp=%s, series=%s" // Using error.Wrap puts the message before the error and if the series is too long, its truncated. + errTSDBIngestWithHeadMaxTime = "err: %v. timestamp=%s, tsdbHeadMaxTimestamp=%s, series=%s" errTSDBIngestExemplar = "err: %v. timestamp=%s, series=%s, exemplar=%s" // Jitter applied to the idle timeout to prevent compaction in all ingesters concurrently. @@ -1417,6 +1418,13 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte perMetricSeriesLimitCount = 0 discardedNativeHistogramCount = 0 + // headMaxTime is snapshotted right after the appender is created. The appender + // derives its lower time bound checks (e.g. out of bounds) from the head max + // time at creation time. We cannot access the appender's internal bound + // directly, so we snapshot the head max time once here instead of fetching it + // again for every failed sample. + headMaxTime int64 + updateFirstPartial = func(errFn func() error) { if firstPartialErr == nil { firstPartialErr = errFn() @@ -1432,7 +1440,9 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte case errors.Is(cause, storage.ErrOutOfBounds): sampleOutOfBoundsCount++ i.validateMetrics.DiscardedSeriesTracker.Track(sampleOutOfBounds, userID, copiedLabels.Hash()) - updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(timestampMs), lbls) }) + updateFirstPartial(func() error { + return wrappedTSDBIngestErrWithHeadMaxTime(err, model.Time(timestampMs), model.Time(headMaxTime), lbls) + }) case errors.Is(cause, storage.ErrOutOfOrderSample): sampleOutOfOrderCount++ @@ -1447,7 +1457,9 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte case errors.Is(cause, storage.ErrTooOldSample): sampleTooOldCount++ i.validateMetrics.DiscardedSeriesTracker.Track(sampleTooOld, userID, copiedLabels.Hash()) - updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(timestampMs), lbls) }) + updateFirstPartial(func() error { + return wrappedTSDBIngestErrWithHeadMaxTime(err, model.Time(timestampMs), model.Time(headMaxTime), lbls) + }) case errors.Is(cause, errMaxSeriesPerUserLimitExceeded): perUserSeriesLimitCount++ @@ -1493,6 +1505,7 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte // Walk the samples, appending them to the users database app := db.Appender(ctx).(extendedAppender) + headMaxTime = db.Head().MaxTime() // Ensure the appender is always released so that we don't leak TSDB head // series references, mmap'd chunks and pending state on early returns. @@ -3947,6 +3960,23 @@ func wrappedTSDBIngestErr(ingestErr error, timestamp model.Time, labels []cortex } } +// wrappedTSDBIngestErrWithHeadMaxTime is like wrappedTSDBIngestErr, but it also includes the +// TSDB head max time in the error message. This helps users understand how far behind the +// accepted time range a rejected sample is, e.g. for out of bounds or too old sample errors. +func wrappedTSDBIngestErrWithHeadMaxTime(ingestErr error, timestamp, headMaxTime model.Time, labels []cortexpb.LabelAdapter) error { + if ingestErr == nil { + return nil + } + + // The TSDB head max time is unset when the head is empty (e.g. no sample ingested yet + // after startup). Fall back to the error message without the head max time in that case. + if int64(headMaxTime) == math.MinInt64 { + return wrappedTSDBIngestErr(ingestErr, timestamp, labels) + } + + return fmt.Errorf(errTSDBIngestWithHeadMaxTime, ingestErr, timestamp.Time().UTC().Format(time.RFC3339Nano), headMaxTime.Time().UTC().Format(time.RFC3339Nano), cortexpb.FromLabelAdaptersToLabels(labels).String()) +} + func wrappedTSDBIngestExemplarErr(ingestErr error, timestamp model.Time, seriesLabels, exemplarLabels []cortexpb.LabelAdapter) error { if ingestErr == nil { return nil diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index a886afedfe..cca8dbf165 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -1504,7 +1504,7 @@ func TestIngester_Push(t *testing.T) { }, cortexpb.API), }, - expectedErr: httpgrpc.Errorf(http.StatusBadRequest, "%s", wrapWithUser(wrappedTSDBIngestErr(storage.ErrOutOfBounds, model.Time(1575043969-(86400*1000)), cortexpb.FromLabelsToLabelAdapters(metricLabels)), userID).Error()), + expectedErr: httpgrpc.Errorf(http.StatusBadRequest, "%s", wrapWithUser(wrappedTSDBIngestErrWithHeadMaxTime(storage.ErrOutOfBounds, model.Time(1575043969-(86400*1000)), model.Time(1575043969), cortexpb.FromLabelsToLabelAdapters(metricLabels)), userID).Error()), expectedIngested: []cortexpb.TimeSeries{ {Labels: metricLabelAdapters, Samples: []cortexpb.Sample{{Value: 2, TimestampMs: 1575043969}}}, }, @@ -1561,7 +1561,7 @@ func TestIngester_Push(t *testing.T) { cortexpb.API), }, oooTimeWindow: 5 * time.Minute, - expectedErr: httpgrpc.Errorf(http.StatusBadRequest, "%s", wrapWithUser(wrappedTSDBIngestErr(storage.ErrTooOldSample, model.Time(1575043969-(600*1000)), cortexpb.FromLabelsToLabelAdapters(metricLabels)), userID).Error()), + expectedErr: httpgrpc.Errorf(http.StatusBadRequest, "%s", wrapWithUser(wrappedTSDBIngestErrWithHeadMaxTime(storage.ErrTooOldSample, model.Time(1575043969-(600*1000)), model.Time(1575043969), cortexpb.FromLabelsToLabelAdapters(metricLabels)), userID).Error()), expectedIngested: []cortexpb.TimeSeries{ {Labels: metricLabelAdapters, Samples: []cortexpb.Sample{{Value: 2, TimestampMs: 1575043969}}}, }, @@ -1663,7 +1663,7 @@ func TestIngester_Push(t *testing.T) { cortexpb.API), }, oooTimeWindow: 5 * time.Minute, - expectedErr: httpgrpc.Errorf(http.StatusBadRequest, "%s", wrapWithUser(wrappedTSDBIngestErr(storage.ErrTooOldSample, model.Time(1575043969-(600*1000)), cortexpb.FromLabelsToLabelAdapters(metricLabels)), userID).Error()), + expectedErr: httpgrpc.Errorf(http.StatusBadRequest, "%s", wrapWithUser(wrappedTSDBIngestErrWithHeadMaxTime(storage.ErrTooOldSample, model.Time(1575043969-(600*1000)), model.Time(1575043969), cortexpb.FromLabelsToLabelAdapters(metricLabels)), userID).Error()), expectedIngested: []cortexpb.TimeSeries{ {Labels: metricLabelAdapters, Histograms: []cortexpb.WrappedHistogram{cortexpb.WrapHistogram(cortexpb.HistogramToHistogramProto(1575043969, tsdbutil.GenerateTestHistogram(1)))}}, }, @@ -8359,3 +8359,25 @@ func TestIngester_DiscardOutOfOrderFlagIntegration(t *testing.T) { require.NoError(t, iter.Err()) require.Equal(t, 1, sampleCount, "Should have exactly one sample stored") } + +func TestWrappedTSDBIngestErrWithHeadMaxTime(t *testing.T) { + lbls := cortexpb.FromLabelsToLabelAdapters(labels.FromStrings(labels.MetricName, "test")) + + // The error message should include both the sample timestamp and the TSDB head max time. + err := wrappedTSDBIngestErrWithHeadMaxTime(storage.ErrOutOfBounds, model.Time(1575043969), model.Time(1575043969+60000), lbls) + require.Error(t, err) + require.Equal(t, `err: out of bounds. timestamp=1970-01-19T05:30:43.969Z, tsdbHeadMaxTimestamp=1970-01-19T05:31:43.969Z, series={__name__="test"}`, err.Error()) + + err = wrappedTSDBIngestErrWithHeadMaxTime(storage.ErrTooOldSample, model.Time(1575043969), model.Time(1575043969+60000), lbls) + require.Error(t, err) + require.Equal(t, `err: too old sample. timestamp=1970-01-19T05:30:43.969Z, tsdbHeadMaxTimestamp=1970-01-19T05:31:43.969Z, series={__name__="test"}`, err.Error()) + + // When the TSDB head max time is unset (empty head), fall back to the error message + // without the head max time. + err = wrappedTSDBIngestErrWithHeadMaxTime(storage.ErrOutOfBounds, model.Time(1575043969), model.Time(math.MinInt64), lbls) + require.Error(t, err) + require.Equal(t, `err: out of bounds. timestamp=1970-01-19T05:30:43.969Z, series={__name__="test"}`, err.Error()) + + // A nil ingest error returns nil. + require.NoError(t, wrappedTSDBIngestErrWithHeadMaxTime(nil, model.Time(1575043969), model.Time(1575043969), lbls)) +}