Repository: prometheus/prometheus (language: go)

Issue:
Error "out of bounds" may disable all jobs when invalid metrics are received by remote-write
### What did you do?

Prometheus 2.39 running with `--web.enable-remote-write-receiver` to receive remote-written metrics from some other Prometheus instances and OpenTelemetry collectors in the same network.

### What did you expect to see?

If a remote machine sends invalid metrics (perhaps with grossly incorrect timestamps?), these samples should just be dropped, in the same way out of order samples are (normally) dropped by the scraper. Other jobs and sources should be unaffected.

### What did you see instead? Under which circumstances?

In some circumstances it may happen that the logs are flooded with errors like:

```
ts=2023-02-20T08:53:21.719Z caller=scrape.go:1294 level=warn component="scrape manager" scrape_pool=blackbox target=<redacted> msg="Appending scrape report failed" err="out of bounds"
ts=2023-02-20T08:53:22.795Z caller=manager.go:683 level=warn component="rule manager" <redacted> msg="Rule evaluation result discarded" err="out of bounds" sample=<redacted> => 1 @[1676883202753]"
```

At the same time,
* the `/targets` page shows most targets across all jobs down with "out of bounds" error;
* a `count(up) by (job)` query will show that most targets have completely vanished.

This is different from https://github.com/prometheus/prometheus/issues/10900 in various ways:
1) there isn't any log line containing both "Out of order sample from remote write" and "out of bounds";
2) the local clock was in sync;
3) deleting the database didn't help;
4) the error vanishes the moment the `--web.enable-remote-write-receiver` option is removed.

### System information

```
$ uname -a
Linux <redacted> 4.18.0-425.10.1.el8_7.x86_64 #1 SMP Wed Dec 14 16:00:01 EST 2022 x86_64 x86_64 x86_64 GNU/Linux
$ cat /etc/*release* | head -n5
NAME="Red Hat Enterprise Linux"
VERSION="8.7 (Ootpa)"
ID="rhel"
ID_LIKE="fedora"
VERSION_ID="8.7"
```

### Prometheus version

```text
$ prometheus --version
prometheus, version 2.39.1 (branch: HEAD, revision: dcd6af9e0d56165c6f5c64ebbc1fae798d24933a)
  build user:       root@273d60c69592
  build date:       20221007-15:57:09
  go version:       go1.19.2
  platform:         linux/amd64
```


### Prometheus configuration file

`/etc/systemd/system/prometheus.service` includes

```
ExecStart=/usr/local/bin/prometheus \
<redacted>
  --web.enable-lifecycle \
  --web.enable-admin-api \
  --web.enable-remote-write-receiver
```




Discussion / hints:
We have also experienced this.

In our case the issue was caused by [this bug](https://github.com/grafana/loki/pull/8370) from loki, and after pushing those series prometheus stopped working (failed to scrape all targets after this).
I think we should indeed review our policy for ingesting samples in the future

> I think we should indeed review our policy for ingesting samples in the future

Yes!
I experienced a similar issue, where i have 2 prometheus instances pushing via remote write to another instance.
One of the downstream instances started up on a host with a system clock set to 10 years in the future. It scraped a future sample, pushed to the upstream prometheus, which pushed the current time to 10 years in the future, breaking all metric ingestion. I had to delete blocks fix it =(
I think we should add a `timeLimitAppender` to writer_handler.go like scrape.go to prevent invalid timestamp
https://github.com/prometheus/prometheus/blob/1ea477f4bcd04c580e51711dca346a06421b311c/scrape/scrape.go#L752-L755
I encounter the very same problem. A date command mistake on a prometheus agent change the date a week or so in the future. This prometheus agent pushed remote write data to several prom servers and those are broken with ` msg="Appending scrape report failed" err="out of bounds"`.

Date and Time have since been synchronized back on the faulty prometheus agent but the problem remains.

how would it be possible to fix the issue ? By deleting some blocks but which ?
> I encounter the very same problem. A date command mistake on a prometheus agent change the date a week or so in the future. This prometheus agent pushed remote write data to several prom servers and those are broken with ` msg="Appending scrape report failed" err="out of bounds"`.
> 
> Date and Time have since been synchronized back on the faulty prometheus agent but the problem remains.
> 
> how would it be possible to fix the issue ? By deleting some blocks but which ?

I manage to recover prometheus by:
1- stop prometheus
2- backup prometheus data directory
3- remove wal and chunks_head directories
4- start prom
Hello.

Same issue here, a badly configured date in a device pushing metrics with OpenTelemetry collector via remote-write causes the out of bounds error and moves the head block times to the future. We solved it by temporarily disabling that device and deleting wal and chunks_head.

Is there an available fix for this case? I'm currently looking for a way to avoid this happening again in the future, but it would be nice if "too far ahead" samples were discarded under remote-write the same way they are in pull mode.

**UPDATE**
A possible solution for anyone in my same situation is to use a transform processor to check the metrics timestamp in the OpenTelemetry collector:

```yaml
processors:
  transform:
    error_mode: ignore
    metric_statements:
    - context: datapoint
      statements:
      - set(time, Now()) where time > Now()
```
But I still think it would be better if some kind of check was implemented in Prometheus as suggested by @chzhuo above.

Best regards.
Ref #13573
## Indicate invalid metric

After I saw the logs on my instances, I saw that under Prometheus Web UI Status -> TSBD Status that the date is fate in the future (today, we have 10 June)

![image](https://github.com/prometheus/prometheus/assets/1560587/bd2632b4-ecf5-4d3a-aba1-8d7d718eefe3)

Then I do a hand-crafted Query

http://prometheus/api/v1/query_range?query=%7B__name__%3D~%22.%2B%22%7D&start=1726735890.207&end=1726739490.207&step=1

where `start` and `end` parameter match the numbers from TSDB Status page. The Current Time Timestamp need to be divided by 1000.

## recovery

I followed the step (https://github.com/prometheus/prometheus/issues/12006#issuecomment-1611410650) to recover from state, but it didn't help.

I head to remove blocks (one-by-one) coming from newest. **DATA WILL BE LOST HERE**, but it's necessary to identity and remove the invalid metric.

Fix patch:
diff --git a/storage/remote/write_handler.go b/storage/remote/write_handler.go
index e7515a42b88..0832c65abe1 100644
--- a/storage/remote/write_handler.go
+++ b/storage/remote/write_handler.go
@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -25,7 +26,9 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/prometheus/prometheus/model/exemplar"
+	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/timestamp"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/prometheus/prometheus/storage"
 	otlptranslator "github.com/prometheus/prometheus/storage/remote/otlptranslator/prometheusremotewrite"
@@ -38,6 +41,8 @@ type writeHandler struct {
 	samplesWithInvalidLabelsTotal prometheus.Counter
 }
 
+const maxAheadTime = 10 * time.Minute
+
 // NewWriteHandler creates a http.Handler that accepts remote write requests and
 // writes them to the provided appendable.
 func NewWriteHandler(logger log.Logger, reg prometheus.Registerer, appendable storage.Appendable) http.Handler {
@@ -104,17 +109,22 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err
 	outOfOrderExemplarErrs := 0
 	samplesWithInvalidLabels := 0
 
-	app := h.appendable.Appender(ctx)
+	timeLimitApp := &timeLimitAppender{
+		Appender: h.appendable.Appender(ctx),
+		maxTime:  timestamp.FromTime(time.Now().Add(maxAheadTime)),
+	}
+
 	defer func() {
 		if err != nil {
-			_ = app.Rollback()
+			_ = timeLimitApp.Rollback()
 			return
 		}
-		err = app.Commit()
+		err = timeLimitApp.Commit()
 	}()
 
 	b := labels.NewScratchBuilder(0)
 	var exemplarErr error
+
 	for _, ts := range req.Timeseries {
 		labels := LabelProtosToLabels(&b, ts.Labels)
 		if !labels.IsValid() {
@@ -124,7 +134,7 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err
 		}
 		var ref storage.SeriesRef
 		for _, s := range ts.Samples {
-			ref, err = app.Append(ref, labels, s.Timestamp, s.Value)
+			ref, err = timeLimitApp.Append(ref, labels, s.Timestamp, s.Value)
 			if err != nil {
 				unwrappedErr := errors.Unwrap(err)
 				if unwrappedErr == nil {
@@ -140,7 +150,7 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err
 		for _, ep := range ts.Exemplars {
 			e := exemplarProtoToExemplar(&b, ep)
 
-			_, exemplarErr = app.AppendExemplar(0, labels, e)
+			_, exemplarErr = timeLimitApp.AppendExemplar(0, labels, e)
 			exemplarErr = h.checkAppendExemplarError(exemplarErr, e, &outOfOrderExemplarErrs)
 			if exemplarErr != nil {
 				// Since exemplar storage is still experimental, we don't fail the request on ingestion errors.
@@ -151,11 +161,12 @@ func (h *writeHandler) write(ctx context.Context, req *prompb.WriteRequest) (err
 		for _, hp := range ts.Histograms {
 			if hp.IsFloatHistogram() {
 				fhs := FloatHistogramProtoToFloatHistogram(hp)
-				_, err = app.AppendHistogram(0, labels, hp.Timestamp, nil, fhs)
+				_, err = timeLimitApp.AppendHistogram(0, labels, hp.Timestamp, nil, fhs)
 			} else {
 				hs := HistogramProtoToHistogram(hp)
-				_, err = app.AppendHistogram(0, labels, hp.Timestamp, hs, nil)
+				_, err = timeLimitApp.AppendHistogram(0, labels, hp.Timestamp, hs, nil)
 			}
+
 			if err != nil {
 				unwrappedErr := errors.Unwrap(err)
 				if unwrappedErr == nil {
@@ -233,3 +244,45 @@ func (h *otlpWriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	w.WriteHeader(http.StatusOK)
 }
+
+type timeLimitAppender struct {
+	storage.Appender
+
+	maxTime int64
+}
+
+func (app *timeLimitAppender) Append(ref storage.SeriesRef, lset labels.Labels, t int64, v float64) (storage.SeriesRef, error) {
+	if t > app.maxTime {
+		return 0, fmt.Errorf("%w: timestamp is too far in the future", storage.ErrOutOfBounds)
+	}
+
+	ref, err := app.Appender.Append(ref, lset, t, v)
+	if err != nil {
+		return 0, err
+	}
+	return ref, nil
+}
+
+func (app *timeLimitAppender) AppendHistogram(ref storage.SeriesRef, l labels.Labels, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram) (storage.SeriesRef, error) {
+	if t > app.maxTime {
+		return 0, fmt.Errorf("%w: timestamp is too far in the future", storage.ErrOutOfBounds)
+	}
+
+	ref, err := app.Appender.AppendHistogram(ref, l, t, h, fh)
+	if err != nil {
+		return 0, err
+	}
+	return ref, nil
+}
+
+func (app *timeLimitAppender) AppendExemplar(ref storage.SeriesRef, l labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) {
+	if e.Ts > app.maxTime {
+		return 0, fmt.Errorf("%w: timestamp is too far in the future", storage.ErrOutOfBounds)
+	}
+
+	ref, err := app.Appender.AppendExemplar(ref, l, e)
+	if err != nil {
+		return 0, err
+	}
+	return ref, nil
+}


Test patch:
diff --git a/storage/remote/write_handler_test.go b/storage/remote/write_handler_test.go
index 1715e92c27f..30dc1b3d693 100644
--- a/storage/remote/write_handler_test.go
+++ b/storage/remote/write_handler_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net/http"
 	"net/http/httptest"
 	"strconv"
@@ -87,73 +88,127 @@ func TestRemoteWriteHandler(t *testing.T) {
 }
 
 func TestOutOfOrderSample(t *testing.T) {
-	buf, _, _, err := buildWriteRequest(nil, []prompb.TimeSeries{{
-		Labels:  []prompb.Label{{Name: "__name__", Value: "test_metric"}},
-		Samples: []prompb.Sample{{Value: 1, Timestamp: 0}},
-	}}, nil, nil, nil, nil)
-	require.NoError(t, err)
+	tests := []struct {
+		Name      string
+		Timestamp int64
+	}{
+		{
+			Name:      "historic",
+			Timestamp: 0,
+		},
+		{
+			Name:      "future",
+			Timestamp: math.MaxInt64,
+		},
+	}
 
-	req, err := http.NewRequest("", "", bytes.NewReader(buf))
-	require.NoError(t, err)
+	for _, tc := range tests {
+		t.Run(tc.Name, func(t *testing.T) {
+			buf, _, _, err := buildWriteRequest(nil, []prompb.TimeSeries{{
+				Labels:  []prompb.Label{{Name: "__name__", Value: "test_metric"}},
+				Samples: []prompb.Sample{{Value: 1, Timestamp: tc.Timestamp}},
+			}}, nil, nil, nil, nil)
+			require.NoError(t, err)
 
-	appendable := &mockAppendable{
-		latestSample: 100,
-	}
-	handler := NewWriteHandler(log.NewNopLogger(), nil, appendable)
+			req, err := http.NewRequest("", "", bytes.NewReader(buf))
+			require.NoError(t, err)
 
-	recorder := httptest.NewRecorder()
-	handler.ServeHTTP(recorder, req)
+			appendable := &mockAppendable{
+				latestSample: 100,
+			}
+			handler := NewWriteHandler(log.NewNopLogger(), nil, appendable)
 
-	resp := recorder.Result()
-	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+			recorder := httptest.NewRecorder()
+			handler.ServeHTTP(recorder, req)
+
+			resp := recorder.Result()
+			require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+		})
+	}
 }
 
 // This test case currently aims to verify that the WriteHandler endpoint
 // don't fail on ingestion errors since the exemplar storage is
 // still experimental.
 func TestOutOfOrderExemplar(t *testing.T) {
-	buf, _, _, err := buildWriteRequest(nil, []prompb.TimeSeries{{
-		Labels:    []prompb.Label{{Name: "__name__", Value: "test_metric"}},
-		Exemplars: []prompb.Exemplar{{Labels: []prompb.Label{{Name: "foo", Value: "bar"}}, Value: 1, Timestamp: 0}},
-	}}, nil, nil, nil, nil)
-	require.NoError(t, err)
+	tests := []struct {
+		Name      string
+		Timestamp int64
+	}{
+		{
+			Name:      "historic",
+			Timestamp: 0,
+		},
+		{
+			Name:      "future",
+			Timestamp: math.MaxInt64,
+		},
+	}
 
-	req, err := http.NewRequest("", "", bytes.NewReader(buf))
-	require.NoError(t, err)
+	for _, tc := range tests {
+		t.Run(tc.Name, func(t *testing.T) {
+			buf, _, _, err := buildWriteRequest(nil, []prompb.TimeSeries{{
+				Labels:    []prompb.Label{{Name: "__name__", Value: "test_metric"}},
+				Exemplars: []prompb.Exemplar{{Labels: []prompb.Label{{Name: "foo", Value: "bar"}}, Value: 1, Timestamp: tc.Timestamp}},
+			}}, nil, nil, nil, nil)
+			require.NoError(t, err)
 
-	appendable := &mockAppendable{
-		latestExemplar: 100,
-	}
-	handler := NewWriteHandler(log.NewNopLogger(), nil, appendable)
+			req, err := http.NewRequest("", "", bytes.NewReader(buf))
+			require.NoError(t, err)
 
-	recorder := httptest.NewRecorder()
-	handler.ServeHTTP(recorder, req)
+			appendable := &mockAppendable{
+				latestExemplar: 100,
+			}
+			handler := NewWriteHandler(log.NewNopLogger(), nil, appendable)
 
-	resp := recorder.Result()
-	// TODO: update to require.Equal(t, http.StatusConflict, resp.StatusCode) once exemplar storage is not experimental.
-	require.Equal(t, http.StatusNoContent, resp.StatusCode)
+			recorder := httptest.NewRecorder()
+			handler.ServeHTTP(recorder, req)
+
+			resp := recorder.Result()
+			// TODO: update to require.Equal(t, http.StatusConflict, resp.StatusCode) once exemplar storage is not experimental.
+			require.Equal(t, http.StatusNoContent, resp.StatusCode)
+		})
+	}
 }
 
 func TestOutOfOrderHistogram(t *testing.T) {
-	buf, _, _, err := buildWriteRequest(nil, []prompb.TimeSeries{{
-		Labels:     []prompb.Label{{Name: "__name__", Value: "test_metric"}},
-		Histograms: []prompb.Histogram{HistogramToHistogramProto(0, &testHistogram), FloatHistogramToHistogramProto(1, testHistogram.ToFloat(nil))},
-	}}, nil, nil, nil, nil)
-	require.NoError(t, err)
+	tests := []struct {
+		Name      string
+		Timestamp int64
+	}{
+		{
+			Name:      "historic",
+			Timestamp: 0,
+		},
+		{
+			Name:      "future",
+			Timestamp: math.MaxInt64,
+		},
+	}
 
-	req, err := http.NewRequest("", "", bytes.NewReader(buf))
-	require.NoError(t, err)
+	for _, tc := range tests {
+		t.Run(tc.Name, func(t *testing.T) {
+			buf, _, _, err := buildWriteRequest(nil, []prompb.TimeSeries{{
+				Labels:     []prompb.Label{{Name: "__name__", Value: "test_metric"}},
+				Histograms: []prompb.Histogram{HistogramToHistogramProto(tc.Timestamp, &testHistogram), FloatHistogramToHistogramProto(1, testHistogram.ToFloat(nil))},
+			}}, nil, nil, nil, nil)
+			require.NoError(t, err)
 
-	appendable := &mockAppendable{
-		latestHistogram: 100,
-	}
-	handler := NewWriteHandler(log.NewNopLogger(), nil, appendable)
+			req, err := http.NewRequest("", "", bytes.NewReader(buf))
+			require.NoError(t, err)
 
-	recorder := httptest.NewRecorder()
-	handler.ServeHTTP(recorder, req)
+			appendable := &mockAppendable{
+				latestHistogram: 100,
+			}
+			handler := NewWriteHandler(log.NewNopLogger(), nil, appendable)
 
-	resp := recorder.Result()
-	require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+			recorder := httptest.NewRecorder()
+			handler.ServeHTTP(recorder, req)
+
+			resp := recorder.Result()
+			require.Equal(t, http.StatusBadRequest, resp.StatusCode)
+		})
+	}
 }
 
 func BenchmarkRemoteWritehandler(b *testing.B) {