Repository: prometheus/prometheus (language: go)

Issue:
Corrupting data written to remote storage in case sample_age_limit is hit
### What did you do?

We've started experiencing data corruption since we have started using `remote_write.queue_config.sample_age_limit`. We want to drop old samples after our Disaster recovery tests, in order to make sure we see fresh data as soon as possible (and we do not care much about samples scraped during the test itself). After these tests (which is basically the only time when `sample_age_limit` applies as we try to make sure we do not hit it during normal operation) we have seen unexpected counter resets as well as new, unexpected timeseries.

### What did you expect to see?

Hitting `sample_age_limit` should only drop old data.

### What did you see instead? Under which circumstances?

Our production setup is Prometheus with remote write directed towards Mimir cluster. In case when remote write endpoint is not accessible for more than `sample_age_limit`, we have seen corruptions in data ingested to remote storage. In both cases, just 2-3 samples just after remote write endpoint resumes its operation, and Prometheus drops old data and starts ingesting again.

So far we have notices two cases:
1. unexpected samples gets appended to an existing timeseries. Timestamp does not align with our scrape_interval, and value also does not align the timeseries. It probably happens to any metric type, but we have noticed this only with counters as the "fake" values are interpreted as a counter reset, causing huge spikes in our graphs.
2. new timeseries is created, with just a few samples just after the remote_write resume to ingest data to remote storage just like in the previous case. We have noticed it for cases which are clearly nonsense. It seems that existing series get mixed, like in this case when we tested it with Prometheus holding only its own exposed metrics and node_exporter - `go_gc_duration_seconds_count{cluster="local-test", instance="localhost:9100", job="node-exporter", name="systemd-networkd.service", state="activating", type="notify-reload"}`

The above we have been able to reproduce in setup with localy running Prometheus binary with configured remote_write towards our staging Mimir cluster. The outage have been simulated by `iptables -I OUTPUT -d __IP__ -j DROP`.

We have also tried to reproduce it in setup with two Prometheis with one serving as a remote_write receiver (see the attached docker-compose). In this scenario, we haven't seen any corrupted data in the receiving Prometheus, but we are getting log complaining about corrupted data:


### System information

Linux 6.5.0-27-generic x86_64

### Prometheus version

```text
prometheus, version 2.50.1 (branch: HEAD, revision: 8c9b0285360a0b6288d76214a75ce3025bce4050)
  build user:       root@6213bb3ee580
  build date:       20240226-11:36:26
  go version:       go1.21.7
  platform:         linux/amd64
  tags:             netgo,builtinassets,stringlabels
```


### Prometheus configuration file

```yaml
global:
        scrape_interval: 30s
        scrape_timeout: 10s
        scrape_protocols:
        - OpenMetricsText1.0.0
        - OpenMetricsText0.0.1
        - PrometheusText0.0.4
        evaluation_interval: 30s
        external_labels:
          __replica__: prometheus-local-test-0
          cluster: local-test
      scrape_configs:
      - job_name: prometheus-scif-monitoring
        honor_timestamps: false
        track_timestamps_staleness: false
        scrape_interval: 30s
        scrape_timeout: 10s
        scrape_protocols:
        - OpenMetricsText1.0.0
        - OpenMetricsText0.0.1
        - PrometheusText0.0.4
        metrics_path: /metrics
        scheme: http
        enable_compression: false
        follow_redirects: false
        enable_http2: false
        static_configs:
        - targets:
          - localhost:9090
      - job_name: node-exporter
        honor_timestamps: false
        track_timestamps_staleness: false
        scrape_interval: 5s
        scrape_timeout: 1s
        scrape_protocols:
        - OpenMetricsText1.0.0
        - OpenMetricsText0.0.1
        - PrometheusText0.0.4
        scheme: http
        enable_compression: false
        follow_redirects: false
        enable_http2: false
        metrics_path: /metrics
        static_configs:
        - targets:
          - node-exporter:9100
      remote_write:
      - url: http://receive-prometheus:9090/api/v1/write
        remote_timeout: 12s
        follow_redirects: true
        enable_http2: true
        queue_config:
          capacity: 100
          max_shards: 3
          min_shards: 12
          batch_send_deadline: 5s
          min_backoff: 30ms
          max_backoff: 5s
          sample_age_limit: 1m
        metadata_config:
          send: true
          send_interval: 1m
          max_samples_per_send: 100
```


### Alertmanager version

_No response_

### Alertmanager configuration file

_No response_

### Logs

```text
prometheus          | ts=2024-04-23T10:06:18.038Z caller=dedupe.go:112 component=remote level=warn remote_name=dabaca url=http://receive-prometheus:9090/api/v1/write msg="Failed to send batch, retrying" err="Post \"http://receive-prometheus:9090/api/v1/write\": context deadline exceeded"
prometheus          | ts=2024-04-23T10:07:26.043Z caller=dedupe.go:112 component=remote level=warn remote_name=dabaca url=http://receive-prometheus:9090/api/v1/write msg="Failed to send batch, retrying" err="Post \"http://receive-prometheus:9090/api/v1/write\": context deadline exceeded"
prometheus          | ts=2024-04-23T10:08:34.050Z caller=dedupe.go:112 component=remote level=warn remote_name=dabaca url=http://receive-prometheus:9090/api/v1/write msg="Failed to send batch, retrying" err="Post \"http://receive-prometheus:9090/api/v1/write\": context deadline exceeded"
prometheus          | ts=2024-04-23T10:09:11.336Z caller=dedupe.go:112 component=remote level=info remote_name=dabaca url=http://receive-prometheus:9090/api/v1/write msg="Remote storage resharding" from=12 to=3
receive-prometheus  | ts=2024-04-23T10:09:11.998Z caller=write_handler.go:77 level=error component=web msg="Error appending remote write" err="label name \"job\" is not unique: invalid sample"
receive-prometheus  | ts=2024-04-23T10:09:11.998Z caller=write_handler.go:77 level=error component=web msg="Error appending remote write" err="label name \"job\" is not unique: invalid sample"
prometheus          | ts=2024-04-23T10:09:12.029Z caller=dedupe.go:112 component=remote level=warn remote_name=dabaca url=http://receive-prometheus:9090/api/v1/write msg="Failed to send batch, retrying" err="server returned HTTP status 500 Internal Server Error: label name \"job\" is not unique: invalid sample"
receive-prometheus  | ts=2024-04-23T10:09:12.035Z caller=write_handler.go:134 level=error component=web msg="Out of order sample from remote write" err="duplicate sample for timestamp" series="{__name__=\"go_gc_duration_seconds\", __replica__=\"prometheus-local-test-0\", cluster=\"local-test\", instance=\"node-exporter:9100\", job=\"node-exporter\"}" timestamp=1713866946725
receive-prometheus  | ts=2024-04-23T10:09:12.035Z caller=write_handler.go:77 level=error component=web msg="Error appending remote write" err="label name \"job\" is not unique: invalid sample"
prometheus          | ts=2024-04-23T10:09:12.035Z caller=dedupe.go:112 component=remote level=error remote_name=dabaca url=http://receive-prometheus:9090/api/v1/write msg="non-recoverable error" count=2000 exemplarCount=0 err="server returned HTTP status 400 Bad Request: duplicate sample for timestamp"
```



Discussion / hints:
Attached is the docker compose we use to replicate the issue.
[docker-compose.yml](https://github.com/prometheus/prometheus/files/15081538/docker-compose.yml.txt)

EDIT:
Steps to reproduce:
1. run the attached compose
2. simulate outage of remote write receiving endpoints `date; docker pause mimir receive-prometheus`
3. *wait for at least sample_age_limit*
4. resume remote write `date; docker unpause mimir receive-prometheus`

@marctc haven't you experienced similar issues? You might have the context as author of this feature 🙏
Attempt to fix this https://github.com/prometheus/prometheus/pull/14078
Here is a PR which reproduces the bug https://github.com/prometheus/prometheus/pull/14157

Fix patch:
diff --git a/storage/remote/queue_manager.go b/storage/remote/queue_manager.go
index 488485e385a..dde78d35e58 100644
--- a/storage/remote/queue_manager.go
+++ b/storage/remote/queue_manager.go
@@ -1783,9 +1783,11 @@ func buildTimeSeries(timeSeries []prompb.TimeSeries, filter func(prompb.TimeSeri
 		if len(ts.Histograms) > 0 && ts.Histograms[0].Timestamp < lowest {
 			lowest = ts.Histograms[0].Timestamp
 		}
-
-		// Move the current element to the write position and increment the write pointer
-		timeSeries[keepIdx] = timeSeries[i]
+		if i != keepIdx {
+			// We have to swap the kept timeseries with the one which should be dropped.
+			// Copying any elements within timeSeries could cause data corruptions when reusing the slice in a next batch (shards.populateTimeSeries).
+			timeSeries[keepIdx], timeSeries[i] = timeSeries[i], timeSeries[keepIdx]
+		}
 		keepIdx++
 	}
 


Test patch:
diff --git a/storage/remote/queue_manager_test.go b/storage/remote/queue_manager_test.go
index 06783167fb8..4d299994bd2 100644
--- a/storage/remote/queue_manager_test.go
+++ b/storage/remote/queue_manager_test.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"math/rand"
 	"os"
 	"runtime/pprof"
 	"sort"
@@ -29,6 +30,7 @@ import (
 	"github.com/go-kit/log"
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
+	"github.com/google/go-cmp/cmp"
 	"github.com/prometheus/client_golang/prometheus"
 	client_testutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/model"
@@ -611,6 +613,30 @@ func createTimeseries(numSamples, numSeries int, extraLabels ...labels.Label) ([
 	return samples, series
 }
 
+func createProtoTimeseriesWithOld(numSamples, baseTs int64, extraLabels ...labels.Label) []prompb.TimeSeries {
+	samples := make([]prompb.TimeSeries, numSamples)
+	// use a fixed rand source so tests are consistent
+	r := rand.New(rand.NewSource(99))
+	for j := int64(0); j < numSamples; j++ {
+		name := fmt.Sprintf("test_metric_%d", j)
+
+		samples[j] = prompb.TimeSeries{
+			Labels: []prompb.Label{{Name: "__name__", Value: name}},
+			Samples: []prompb.Sample{
+				{
+					Timestamp: baseTs + j,
+					Value:     float64(j),
+				},
+			},
+		}
+		// 10% of the time use a ts that is too old
+		if r.Intn(10) == 0 {
+			samples[j].Samples[0].Timestamp = baseTs - 5
+		}
+	}
+	return samples
+}
+
 func createExemplars(numExemplars, numSeries int) ([]record.RefExemplar, []record.RefSeries) {
 	exemplars := make([]record.RefExemplar, 0, numExemplars)
 	series := make([]record.RefSeries, 0, numSeries)
@@ -679,8 +705,8 @@ func createHistograms(numSamples, numSeries int, floatHistogram bool) ([]record.
 	return histograms, nil, series
 }
 
-func getSeriesNameFromRef(r record.RefSeries) string {
-	return r.Labels.Get("__name__")
+func getSeriesIDFromRef(r record.RefSeries) string {
+	return r.Labels.String()
 }
 
 type TestWriteClient struct {
@@ -698,6 +724,9 @@ type TestWriteClient struct {
 	wg                      sync.WaitGroup
 	mtx                     sync.Mutex
 	buf                     []byte
+
+	storeWait   time.Duration
+	returnError error
 }
 
 func NewTestWriteClient() *TestWriteClient {
@@ -706,6 +735,8 @@ func NewTestWriteClient() *TestWriteClient {
 		receivedSamples:  map[string][]prompb.Sample{},
 		expectedSamples:  map[string][]prompb.Sample{},
 		receivedMetadata: map[string][]prompb.MetricMetadata{},
+		storeWait:        0,
+		returnError:      nil,
 	}
 }
 
@@ -720,12 +751,15 @@ func (c *TestWriteClient) expectSamples(ss []record.RefSample, series []record.R
 	c.receivedSamples = map[string][]prompb.Sample{}
 
 	for _, s := range ss {
-		seriesName := getSeriesNameFromRef(series[s.Ref])
-		c.expectedSamples[seriesName] = append(c.expectedSamples[seriesName], prompb.Sample{
+		tsID := getSeriesIDFromRef(series[s.Ref])
+		c.expectedSamples[tsID] = append(c.expectedSamples[tsID], prompb.Sample{
 			Timestamp: s.T,
 			Value:     s.V,
 		})
 	}
+	if !c.withWaitGroup {
+		return
+	}
 	c.wg.Add(len(ss))
 }
 
@@ -740,13 +774,13 @@ func (c *TestWriteClient) expectExemplars(ss []record.RefExemplar, series []reco
 	c.receivedExemplars = map[string][]prompb.Exemplar{}
 
 	for _, s := range ss {
-		seriesName := getSeriesNameFromRef(series[s.Ref])
+		tsID := getSeriesIDFromRef(series[s.Ref])
 		e := prompb.Exemplar{
 			Labels:    LabelsToLabelsProto(s.Labels, nil),
 			Timestamp: s.T,
 			Value:     s.V,
 		}
-		c.expectedExemplars[seriesName] = append(c.expectedExemplars[seriesName], e)
+		c.expectedExemplars[tsID] = append(c.expectedExemplars[tsID], e)
 	}
 	c.wg.Add(len(ss))
 }
@@ -762,8 +796,8 @@ func (c *TestWriteClient) expectHistograms(hh []record.RefHistogramSample, serie
 	c.receivedHistograms = map[string][]prompb.Histogram{}
 
 	for _, h := range hh {
-		seriesName := getSeriesNameFromRef(series[h.Ref])
-		c.expectedHistograms[seriesName] = append(c.expectedHistograms[seriesName], HistogramToHistogramProto(h.T, h.H))
+		tsID := getSeriesIDFromRef(series[h.Ref])
+		c.expectedHistograms[tsID] = append(c.expectedHistograms[tsID], HistogramToHistogramProto(h.T, h.H))
 	}
 	c.wg.Add(len(hh))
 }
@@ -779,8 +813,8 @@ func (c *TestWriteClient) expectFloatHistograms(fhs []record.RefFloatHistogramSa
 	c.receivedFloatHistograms = map[string][]prompb.Histogram{}
 
 	for _, fh := range fhs {
-		seriesName := getSeriesNameFromRef(series[fh.Ref])
-		c.expectedFloatHistograms[seriesName] = append(c.expectedFloatHistograms[seriesName], FloatHistogramToHistogramProto(fh.T, fh.FH))
+		tsID := getSeriesIDFromRef(series[fh.Ref])
+		c.expectedFloatHistograms[tsID] = append(c.expectedFloatHistograms[tsID], FloatHistogramToHistogramProto(fh.T, fh.FH))
 	}
 	c.wg.Add(len(fhs))
 }
@@ -806,9 +840,27 @@ func (c *TestWriteClient) waitForExpectedData(tb testing.TB) {
 	}
 }
 
+func (c *TestWriteClient) SetStoreWait(w time.Duration) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	c.storeWait = w
+}
+
+func (c *TestWriteClient) SetReturnError(err error) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	c.returnError = err
+}
+
 func (c *TestWriteClient) Store(_ context.Context, req []byte, _ int) error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
+	if c.storeWait > 0 {
+		time.Sleep(c.storeWait)
+	}
+	if c.returnError != nil {
+		return c.returnError
+	}
 	// nil buffers are ok for snappy, ignore cast error.
 	if c.buf != nil {
 		c.buf = c.buf[:cap(c.buf)]
@@ -827,23 +879,23 @@ func (c *TestWriteClient) Store(_ context.Context, req []byte, _ int) error {
 	count := 0
 	for _, ts := range reqProto.Timeseries {
 		labels := LabelProtosToLabels(&builder, ts.Labels)
-		seriesName := labels.Get("__name__")
+		tsID := labels.String()
 		for _, sample := range ts.Samples {
 			count++
-			c.receivedSamples[seriesName] = append(c.receivedSamples[seriesName], sample)
+			c.receivedSamples[tsID] = append(c.receivedSamples[tsID], sample)
 		}
 
 		for _, ex := range ts.Exemplars {
 			count++
-			c.receivedExemplars[seriesName] = append(c.receivedExemplars[seriesName], ex)
+			c.receivedExemplars[tsID] = append(c.receivedExemplars[tsID], ex)
 		}
 
 		for _, histogram := range ts.Histograms {
 			count++
 			if histogram.IsFloatHistogram() {
-				c.receivedFloatHistograms[seriesName] = append(c.receivedFloatHistograms[seriesName], histogram)
+				c.receivedFloatHistograms[tsID] = append(c.receivedFloatHistograms[tsID], histogram)
 			} else {
-				c.receivedHistograms[seriesName] = append(c.receivedHistograms[seriesName], histogram)
+				c.receivedHistograms[tsID] = append(c.receivedHistograms[tsID], histogram)
 			}
 		}
 	}
@@ -1441,6 +1493,99 @@ func TestIsSampleOld(t *testing.T) {
 	require.False(t, isSampleOld(currentTime, 60*time.Second, timestamp.FromTime(currentTime.Add(-59*time.Second))))
 }
 
+// Simulates scenario in which remote write endpoint is down and a subset of samples is dropped due to age limit while backoffing.
+func TestSendSamplesWithBackoffWithSampleAgeLimit(t *testing.T) {
+	maxSamplesPerSend := 10
+	sampleAgeLimit := time.Second
+
+	cfg := config.DefaultQueueConfig
+	cfg.MaxShards = 1
+	cfg.SampleAgeLimit = model.Duration(sampleAgeLimit)
+	// Set the batch send deadline to 5 minutes to effectively disable it.
+	cfg.BatchSendDeadline = model.Duration(time.Minute * 5)
+	cfg.Capacity = 10 * maxSamplesPerSend // more than the amount of data we append in the test
+	cfg.MaxBackoff = model.Duration(time.Millisecond * 100)
+	cfg.MinBackoff = model.Duration(time.Millisecond * 100)
+	cfg.MaxSamplesPerSend = maxSamplesPerSend
+	metadataCfg := config.DefaultMetadataConfig
+	metadataCfg.Send = true
+	metadataCfg.SendInterval = model.Duration(time.Second * 60)
+	metadataCfg.MaxSamplesPerSend = maxSamplesPerSend
+	c := NewTestWriteClient()
+	c.withWaitGroup = false
+	m := newTestQueueManager(t, cfg, metadataCfg, time.Second, c)
+
+	m.Start()
+
+	batchID := 0
+	expectedSamples := map[string][]prompb.Sample{}
+
+	appendData := func(numberOfSeries int, timeAdd time.Duration, shouldBeDropped bool) {
+		t.Log(">>>>  Appending series ", numberOfSeries, " as batch ID ", batchID, " with timeAdd ", timeAdd, " and should be dropped ", shouldBeDropped)
+		samples, series := createTimeseriesWithRandomLabelCount(strconv.Itoa(batchID), numberOfSeries, timeAdd, 9)
+		m.StoreSeries(series, batchID)
+		sent := m.Append(samples)
+		require.True(t, sent, "samples not sent")
+		if !shouldBeDropped {
+			for _, s := range samples {
+				tsID := getSeriesIDFromRef(series[s.Ref])
+				expectedSamples[tsID] = append(c.expectedSamples[tsID], prompb.Sample{
+					Timestamp: s.T,
+					Value:     s.V,
+				})
+			}
+		}
+		batchID++
+	}
+	timeShift := -time.Millisecond * 5
+
+	c.SetReturnError(RecoverableError{context.DeadlineExceeded, defaultBackoff})
+
+	appendData(maxSamplesPerSend/2, timeShift, true)
+	time.Sleep(sampleAgeLimit)
+	appendData(maxSamplesPerSend/2, timeShift, true)
+	time.Sleep(sampleAgeLimit / 10)
+	appendData(maxSamplesPerSend/2, timeShift, true)
+	time.Sleep(2 * sampleAgeLimit)
+	appendData(2*maxSamplesPerSend, timeShift, false)
+	time.Sleep(sampleAgeLimit / 2)
+	c.SetReturnError(nil)
+	appendData(5, timeShift, false)
+	m.Stop()
+
+	if diff := cmp.Diff(expectedSamples, c.receivedSamples); diff != "" {
+		t.Errorf("mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func createTimeseriesWithRandomLabelCount(id string, seriesCount int, timeAdd time.Duration, maxLabels int) ([]record.RefSample, []record.RefSeries) {
+	samples := []record.RefSample{}
+	series := []record.RefSeries{}
+	// use a fixed rand source so tests are consistent
+	r := rand.New(rand.NewSource(99))
+	for i := 0; i < seriesCount; i++ {
+		s := record.RefSample{
+			Ref: chunks.HeadSeriesRef(i),
+			T:   time.Now().Add(timeAdd).UnixMilli(),
+			V:   r.Float64(),
+		}
+		samples = append(samples, s)
+		labelsCount := r.Intn(maxLabels)
+		lb := labels.NewScratchBuilder(1 + labelsCount)
+		lb.Add("__name__", "batch_"+id+"_id_"+strconv.Itoa(i))
+		for j := 1; j < labelsCount+1; j++ {
+			// same for both name and value
+			label := "batch_" + id + "_label_" + strconv.Itoa(j)
+			lb.Add(label, label)
+		}
+		series = append(series, record.RefSeries{
+			Ref:    chunks.HeadSeriesRef(i),
+			Labels: lb.Labels(),
+		})
+	}
+	return samples, series
+}
+
 func createTimeseriesWithOldSamples(numSamples, numSeries int, extraLabels ...labels.Label) ([]record.RefSample, []record.RefSample, []record.RefSeries) {
 	newSamples := make([]record.RefSample, 0, numSamples)
 	samples := make([]record.RefSample, 0, numSamples)
@@ -1668,3 +1813,14 @@ func TestBuildTimeSeries(t *testing.T) {
 		})
 	}
 }
+
+func BenchmarkBuildTimeSeries(b *testing.B) {
+	// Send one sample per series, which is the typical remote_write case
+	const numSamples = 10000
+	filter := func(ts prompb.TimeSeries) bool { return filterTsLimit(99, ts) }
+	for i := 0; i < b.N; i++ {
+		samples := createProtoTimeseriesWithOld(numSamples, 100, extraLabels...)
+		_, _, result, _, _, _ := buildTimeSeries(samples, filter)
+		require.NotNil(b, result)
+	}
+}