Ruler: stop retrying staleness markers that can never be accepted

CHANGELOG.md

@@ -11,6 +11,8 @@
 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869
 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880
 * [BUGFIX] Ingester: Staleness markers written by the Ruler are no longer reported as discarded samples for the tenant. #7887
+* [BUGFIX] Ruler: Stop retrying staleness markers that can never be accepted. A rule moved between rule groups left the original group re-appending staleness markers for its series on every evaluation until the ruler restarted. #7888
+
 
 ## 1.22.0 in progress
 * [CHANGE] Ruler: Remove the deprecated `-ruler.evaluation-delay-duration` flag and its `ruler_evaluation_delay_duration` per-tenant limit. Use `-ruler.query-offset` / `ruler_query_offset`, which no longer takes the higher of the two values. Cortex decodes the runtime config strictly, so a leftover `ruler_evaluation_delay_duration` override makes the runtime config fail to load: Cortex **exits at startup** (`module failed`, `module=runtime-config`), and on an already-running process every reload fails, pinning the last good overrides and dropping `cortex_runtime_config_last_reload_successful` to 0. Run `grep -r ruler_evaluation_delay_duration` over your runtime configs before upgrading. #7792

pkg/ruler/compat.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
 	"net/url"
 
 	"time"
@@ -16,6 +17,7 @@ import (
 	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/metadata"
+	"github.com/prometheus/prometheus/model/value"
 	"github.com/prometheus/prometheus/notifier"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/rules"
@@ -97,6 +99,9 @@ func (a *PusherAppender) AppendExemplar(_ storage.SeriesRef, _ labels.Labels, _
 func (a *PusherAppender) Commit() error {
 	a.totalWrites.Inc()
 
+	// Computed before ToWriteRequest takes ownership of the samples.
+	onlyStaleMarkers := a.onlyStaleMarkers()
+
 	req := cortexpb.ToWriteRequest(a.labels, a.samples, nil, nil, cortexpb.RULE)
 	req.AddHistogramTimeSeries(a.histogramLabels, a.histograms)
 
@@ -112,6 +117,13 @@ func (a *PusherAppender) Commit() error {
 		// Don't report errors that ended with 4xx HTTP status code (series limits, duplicate samples, out of order, etc.)
 		if resp, ok := httpgrpc.HTTPResponseFromError(err); !ok || resp.Code/100 == 5 {
 			a.failedWrites.Inc()
+		} else if onlyStaleMarkers && resp.Code != http.StatusTooManyRequests {
+			// cleanupStaleSeries keeps its staleSeries list until Commit succeeds, so a
+			// permanent rejection makes it re-append the whole list on every evaluation
+			// until the ruler restarts. Report success so it drops them instead. Markers
+			// only: with rule results the ruler surfaces the error as rule health. 429 is
+			// excluded, the rate limiter rejects the whole request and is transient.
+			err = nil
 		}
 	}
 
@@ -122,6 +134,20 @@ func (a *PusherAppender) Commit() error {
 	return err
 }
 
+// onlyStaleMarkers reports whether every sample in the transaction is a staleness
+// marker, as in a transaction built by rules.Group.cleanupStaleSeries.
+func (a *PusherAppender) onlyStaleMarkers() bool {
+	if len(a.samples) == 0 || len(a.histograms) > 0 {
+		return false
+	}
+	for _, s := range a.samples {
+		if !value.IsStaleNaN(s.Value) {
+			return false
+		}
+	}
+	return true
+}
+
 func (a *PusherAppender) UpdateMetadata(_ storage.SeriesRef, _ labels.Labels, _ metadata.Metadata) (storage.SeriesRef, error) {
 	return 0, errors.New("update metadata unsupported")
 }

pkg/ruler/compat_test.go

@@ -312,6 +312,98 @@ func TestPusherErrors(t *testing.T) {
 	}
 }
 
+func TestPusherAppender_Commit_StaleMarkers(t *testing.T) {
+	staleNaN := math.Float64frombits(value.StaleNaN)
+
+	for name, tc := range map[string]struct {
+		samples          []float64
+		withHistogram    bool
+		returnedError    error
+		expectedError    error
+		expectedFailures int
+	}{
+		"only staleness markers, 4xx: reported as success so the caller stops retrying": {
+			samples:       []float64{staleNaN, staleNaN},
+			returnedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+			expectedError: nil,
+		},
+		"only staleness markers, 429: still an error, the rate limiter rejected the whole request and the limit is transient": {
+			samples:       []float64{staleNaN},
+			returnedError: httpgrpc.Errorf(http.StatusTooManyRequests, "test error"),
+			expectedError: httpgrpc.Errorf(http.StatusTooManyRequests, "test error"),
+		},
+		"only staleness markers, 5xx: still an error, retrying is correct": {
+			samples:          []float64{staleNaN},
+			returnedError:    httpgrpc.Errorf(http.StatusInternalServerError, "test error"),
+			expectedError:    httpgrpc.Errorf(http.StatusInternalServerError, "test error"),
+			expectedFailures: 1,
+		},
+		"only staleness markers, non-HTTP error: still an error": {
+			samples:          []float64{staleNaN},
+			returnedError:    errors.New("test error"),
+			expectedError:    errors.New("test error"),
+			expectedFailures: 1,
+		},
+		"only staleness markers, no error": {
+			samples:       []float64{staleNaN},
+			returnedError: nil,
+			expectedError: nil,
+		},
+		"markers alongside a rule result, 4xx: still an error, the ruler reports rule health from it": {
+			samples:       []float64{staleNaN, 123456},
+			returnedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+			expectedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+		},
+		"rule results only, 4xx: still an error": {
+			samples:       []float64{123456},
+			returnedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+			expectedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+		},
+		"staleness marker with a histogram, 4xx: still an error, cleanupStaleSeries never appends histograms": {
+			samples:       []float64{staleNaN},
+			withHistogram: true,
+			returnedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+			expectedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+		},
+		"empty transaction, 4xx: still an error": {
+			returnedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+			expectedError: httpgrpc.Errorf(http.StatusBadRequest, "test error"),
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			ctx := context.Background()
+			pusher := &fakePusher{err: tc.returnedError, response: &cortexpb.WriteResponse{}}
+
+			writes := prometheus.NewCounter(prometheus.CounterOpts{})
+			failures := prometheus.NewCounter(prometheus.CounterOpts{})
+
+			pa := NewPusherAppendable(pusher, "user-1", &ruleLimits{}, writes, failures)
+
+			lbls, err := parser.ParseMetric("foo_bar")
+			require.NoError(t, err)
+
+			a := pa.Appender(ctx)
+			for _, v := range tc.samples {
+				_, err = a.Append(0, lbls, int64(model.Now()), v)
+				require.NoError(t, err)
+			}
+			if tc.withHistogram {
+				_, err = a.AppendHistogram(0, lbls, int64(model.Now()), tsdbutil.GenerateTestHistogram(1), nil)
+				require.NoError(t, err)
+			}
+
+			if tc.expectedError == nil {
+				require.NoError(t, a.Commit())
+			} else {
+				require.Equal(t, tc.expectedError, a.Commit())
+			}
+
+			require.Equal(t, 1, int(testutil.ToFloat64(writes)))
+			require.Equal(t, tc.expectedFailures, int(testutil.ToFloat64(failures)))
+		})
+	}
+}
+
 func TestMetricsQueryFuncErrors(t *testing.T) {
 	for name, tc := range map[string]struct {
 		returnedError          error