polish documentation

Signed-off-by: JAYICE <1185430411@qq.com>

app/vmagent/opentelemetry/request_handler.go

@@ -63,7 +63,6 @@ func insertRows(at *auth.Token, tss []prompb.TimeSeries, mms []prompb.MetricMeta
 
 	rowsTotal := 0
 	tssDst := ctx.WriteRequest.Timeseries[:0]
-	mmsDst := ctx.WriteRequest.Metadata[:0]
 	labels := ctx.Labels[:0]
 	samples := ctx.Samples[:0]
 	for i := range tss {
@@ -83,19 +82,7 @@ func insertRows(at *auth.Token, tss []prompb.TimeSeries, mms []prompb.MetricMeta
 
 	var metadataTotal int
 	if prommetadata.IsEnabled() {
-		for i := range mms {
-			mm := &mms[i]
-			mmsDst = append(mmsDst, prompb.MetricMetadata{
-				MetricFamilyName: mm.MetricFamilyName,
-				Help:             mm.Help,
-				Type:             mm.Type,
-				Unit:             mm.Unit,
-
-				AccountID: mm.AccountID,
-				ProjectID: mm.ProjectID,
-			})
-		}
-		ctx.WriteRequest.Metadata = mmsDst
+		ctx.WriteRequest.Metadata = mms
 		metadataTotal = len(mms)
 	}
 

app/vmagent/remotewrite/obfuscation.go

@@ -0,0 +1,88 @@
+package remotewrite
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"strings"
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
+)
+
+type obfuscationCtx struct {
+	labels                []prompb.Label
+	cacheObfuscatedResult map[string]string
+}
+
+func (ctx *obfuscationCtx) Reset() {
+	promrelabel.CleanLabels(ctx.labels)
+	ctx.labels = ctx.labels[:0]
+	clear(ctx.cacheObfuscatedResult)
+}
+
+var obfuscationCtxPool = &sync.Pool{
+	New: func() any {
+		return &obfuscationCtx{
+			cacheObfuscatedResult: make(map[string]string),
+		}
+	},
+}
+
+func (rwctx *remoteWriteCtx) initObfuscationConfig() {
+	if len(*obfuscationLabels) == 0 {
+		return
+	}
+	idx := rwctx.idx
+	rwctx.obfuscationLabels = make(map[string]struct{})
+	rwObfuscationLabels := obfuscationLabels.GetOptionalArg(idx)
+	rwObfuscationLabelsList := strings.Split(rwObfuscationLabels, "^^")
+
+	for _, label := range rwObfuscationLabelsList {
+		rwctx.obfuscationLabels[label] = struct{}{}
+	}
+}
+
+func (rwctx *remoteWriteCtx) applyObfuscation(tss []prompb.TimeSeries, ctx *obfuscationCtx) []prompb.TimeSeries {
+	if len(rwctx.obfuscationLabels) == 0 || len(tss) == 0 {
+		return tss
+	}
+	poolLabels := ctx.labels[:0]
+	for i := range tss {
+		ts := &tss[i]
+		labels := ts.Labels
+		j := 0
+		needToObfuscate := false
+		for ; j < len(labels); j++ {
+			label := &labels[j]
+			if _, ok := rwctx.obfuscationLabels[label.Name]; !ok {
+				continue
+			}
+			needToObfuscate = true
+			break
+		}
+		if !needToObfuscate {
+			continue
+		}
+		// Copy the label array to apply obfuscation
+		poolLabelsLen := len(poolLabels)
+		poolLabels = append(poolLabels, labels...)
+		ctx.labels = poolLabels
+		ts.Labels = poolLabels[poolLabelsLen:]
+		for ; j < len(ts.Labels); j++ {
+			label := &ts.Labels[j]
+			if _, ok := rwctx.obfuscationLabels[label.Name]; !ok {
+				continue
+			}
+			if obfuscatedValue, ok := ctx.cacheObfuscatedResult[label.Value]; ok {
+				// fast path: the obfuscated result was calculated before
+				label.Value = obfuscatedValue
+			} else {
+				obfuscatedResult := sha256.Sum256([]byte(label.Value))
+				ctx.cacheObfuscatedResult[label.Value] = hex.EncodeToString(obfuscatedResult[:])
+				label.Value = ctx.cacheObfuscatedResult[label.Value]
+			}
+		}
+	}
+	return tss
+}

app/vmagent/remotewrite/remotewrite.go

@@ -108,6 +108,8 @@ var (
 
 	enableMdx = flagutil.NewArrayBool("remoteWrite.mdx.enable", "Whether to only retain metrics from VictoriaMetrics services before sending them to the corresponding -remoteWrite.url. "+
 		"Please see https://docs.victoriametrics.com/victoriametrics/vmagent/#monitoring-data-exchange")
+	obfuscationLabels = flagutil.NewArrayString("remoteWrite.obfuscationLabels", "List of label names whose values will be obfuscated before being sent to the corresponding -remoteWrite.url. "+
+		"Multiple label names should be separated by `^^`, e.g. \"job^^instance,ip\".")
 )
 
 var (
@@ -881,6 +883,8 @@ type remoteWriteCtx struct {
 	pss        []*pendingSeries
 	pssNextIdx atomic.Uint64
 
+	obfuscationLabels map[string]struct{}
+
 	rowsPushedAfterRelabel *metrics.Counter
 	rowsDroppedByRelabel   *metrics.Counter
 	mdxRowsPreserved       *metrics.Counter
@@ -995,6 +999,7 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, sanitizedURL string)
 		rowsDroppedOnPushFailure:     metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_samples_dropped_total{path=%q,url=%q}`, queuePath, sanitizedURL)),
 	}
 	rwctx.initStreamAggrConfig()
+	rwctx.initObfuscationConfig()
 
 	if enableMdx.GetOptionalArg(argIdx) {
 		mdxFilter := mdx.NewFilter()
@@ -1198,6 +1203,7 @@ func (rwctx *remoteWriteCtx) tryPushMetadataInternal(mms []prompb.MetricMetadata
 func (rwctx *remoteWriteCtx) tryPushTimeSeriesInternal(tss []prompb.TimeSeries) bool {
 	var rctx *relabelCtx
 	var v *[]prompb.TimeSeries
+	var octx *obfuscationCtx
 	defer func() {
 		if rctx == nil {
 			return
@@ -1216,6 +1222,24 @@ func (rwctx *remoteWriteCtx) tryPushTimeSeriesInternal(tss []prompb.TimeSeries)
 		rctx.appendExtraLabels(tss, labelsGlobal)
 	}
 
+	if len(rwctx.obfuscationLabels) != 0 {
+		if rctx == nil {
+			shadowTss := tssPool.Get().(*[]prompb.TimeSeries)
+			tss = append(*shadowTss, tss...)
+			defer func() {
+				*shadowTss = prompb.ResetTimeSeries(tss)
+				tssPool.Put(shadowTss)
+			}()
+		}
+		octx = obfuscationCtxPool.Get().(*obfuscationCtx)
+		defer func() {
+			octx.Reset()
+			obfuscationCtxPool.Put(octx)
+		}()
+
+		tss = rwctx.applyObfuscation(tss, octx)
+	}
+
 	pss := rwctx.pss
 	idx := rwctx.pssNextIdx.Add(1) % uint64(len(pss))
 

app/vmagent/remotewrite/remotewrite_test.go

@@ -1,6 +1,8 @@
 package remotewrite
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"fmt"
 	"math"
 	"reflect"
@@ -376,3 +378,126 @@ func TestCalculateHealthyRwctxIdx(t *testing.T) {
 	f(1, []int{0}, nil)
 	f(1, []int{}, []int{0})
 }
+
+func TestRemoteWriteObfuscation(t *testing.T) {
+	f := func(obfuscationLabelList string, inputTss []prompb.TimeSeries, expectedTss []prompb.TimeSeries) {
+		t.Helper()
+		rwctx := &remoteWriteCtx{
+			idx: 0,
+		}
+		defer metrics.UnregisterAllMetrics()
+		originValue := *obfuscationLabels
+		defer func() {
+			*obfuscationLabels = originValue
+		}()
+		*obfuscationLabels = []string{obfuscationLabelList}
+		rwctx.initObfuscationConfig()
+		octx := obfuscationCtx{
+			cacheObfuscatedResult: make(map[string]string),
+		}
+		outputTss := rwctx.applyObfuscation(inputTss, &octx)
+
+		if !reflect.DeepEqual(expectedTss, outputTss) {
+			t.Fatalf("unexpected samples;\ngot\n%v\nwant\n%v", outputTss, expectedTss)
+		}
+	}
+
+	sha256Result := func(str string) string {
+		sha256Result := sha256.Sum256([]byte(str))
+		return hex.EncodeToString(sha256Result[:])
+	}
+
+	// 1. obfuscation is not set.
+	f("",
+		[]prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "ip", Value: "123"},
+					{Name: "instance", Value: "1234"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+		},
+		[]prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "ip", Value: "123"},
+					{Name: "instance", Value: "1234"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+		},
+	)
+
+	// 2. obfuscate the value of "ip" label
+	f("ip",
+		[]prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "ip", Value: "123"},
+					{Name: "instance", Value: "1234"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+		},
+		[]prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "ip", Value: sha256Result("123")},
+					{Name: "instance", Value: "1234"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+		},
+	)
+
+	// 3. obfuscate the values of "ip" and "instance"
+	f("ip^^instance",
+		[]prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "ip", Value: "123"},
+					{Name: "instance", Value: "1234"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+			{
+				Labels: []prompb.Label{
+					{Name: "job", Value: "123"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+		},
+		[]prompb.TimeSeries{
+			{
+				Labels: []prompb.Label{
+					{Name: "ip", Value: sha256Result("123")},
+					{Name: "instance", Value: sha256Result("1234")},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+			{
+				Labels: []prompb.Label{
+					{Name: "job", Value: "123"},
+				},
+				Samples: []prompb.Sample{
+					{Value: 1, Timestamp: 0},
+				},
+			},
+		},
+	)
+}

app/vmstorage/main.go

@@ -30,11 +30,6 @@ var (
 		"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#retention. See also -retentionFilter")
 	futureRetention = flagutil.NewRetentionDuration("futureRetention", "2d", "Data with timestamps bigger than now+futureRetention is automatically deleted. "+
 		"The minimum futureRetention is 2 days. See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#retention")
-	maxBackfillAge = flagutil.NewRetentionDuration("maxBackfillAge", "0", "The maximum allowed age for the ingested samples with historical timestamps. "+
-		"Samples with timestamps older than now-maxBackfillAge are rejected during data ingestion. "+
-		"By default, or when set to 0, -maxBackfillAge equals to -retentionPeriod, e.g. it is unlimited within the configured retention. "+
-		"This can be useful for limiting ingestion of historical samples, for example, when older data has been moved to another storage tier. "+
-		"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#retention")
 	vmselectAddr                  = flag.String("vmselectAddr", "", "TCP address to accept connections from vmselect services")
 	vmselectDisableRPCCompression = flag.Bool("rpc.disableCompression", false, "Whether to disable compression of the data sent from vmstorage to vmselect. "+
 		"This reduces CPU usage at the cost of higher network bandwidth usage")
@@ -151,7 +146,6 @@ func Init(vmselectMaxConcurrentRequests int, vmselectMaxQueueDuration time.Durat
 	opts := storage.OpenOptions{
 		Retention:                   retentionPeriod.Duration(),
 		FutureRetention:             futureRetention.Duration(),
-		MaxBackfillAge:              maxBackfillAge.Duration(),
 		DenyQueriesOutsideRetention: *denyQueriesOutsideRetention,
 		MaxHourlySeries:             getMaxHourlySeries(),
 		MaxDailySeries:              getMaxDailySeries(),
@@ -473,10 +467,8 @@ func (vms *VMStorage) writeStorageMetrics(w io.Writer) {
 	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="storage/inmemory"}`, tm.InmemorySizeBytes)
 	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="storage/small"}`, tm.SmallSizeBytes)
 	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="storage/big"}`, tm.BigSizeBytes)
-	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="storage/metaindex"}`, tm.MetaindexSizeBytes)
 	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="indexdb/inmemory"}`, idbm.InmemorySizeBytes)
 	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="indexdb/file"}`, idbm.FileSizeBytes)
-	metrics.WriteGaugeUint64(w, `vm_data_size_bytes{type="indexdb/metaindex"}`, idbm.MetaindexSizeBytes)
 
 	metrics.WriteCounterUint64(w, `vm_rows_received_by_storage_total`, m.RowsReceivedTotal)
 	metrics.WriteCounterUint64(w, `vm_rows_added_to_storage_total`, m.RowsAddedTotal)

apptest/tests/max_backfill_age_test.go

@@ -1,222 +0,0 @@
-package tests
-
-import (
-	"fmt"
-	"path/filepath"
-	"testing"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/apptest"
-)
-
-func TestSingleMaxBackfillAge(t *testing.T) {
-	tc := apptest.NewTestCase(t)
-	defer tc.Stop()
-
-	opts := maxBackfillAgeOpts{
-		start: func(retentionPeriod, maxBackfillAge string) apptest.PrometheusWriteQuerier {
-			return tc.MustStartVmsingle("vmsingle", []string{
-				"-storageDataPath=" + filepath.Join(tc.Dir(), "vmsingle"),
-				"-retentionPeriod=" + retentionPeriod,
-				"-maxBackfillAge=" + maxBackfillAge,
-			})
-		},
-		stop: func() {
-			tc.StopApp("vmsingle")
-		},
-	}
-
-	testMaxBackfillAge(tc, opts)
-}
-
-func TestClusterMaxBackfillAge(t *testing.T) {
-	tc := apptest.NewTestCase(t)
-	defer tc.Stop()
-
-	opts := maxBackfillAgeOpts{
-		start: func(retentionPeriod, maxBackfillAge string) apptest.PrometheusWriteQuerier {
-			return tc.MustStartCluster(&apptest.ClusterOptions{
-				Vmstorage1Instance: "vmstorage1",
-				Vmstorage1Flags: []string{
-					"-storageDataPath=" + filepath.Join(tc.Dir(), "vmstorage1"),
-					"-retentionPeriod=" + retentionPeriod,
-					"-maxBackfillAge=" + maxBackfillAge,
-				},
-				Vmstorage2Instance: "vmstorage2",
-				Vmstorage2Flags: []string{
-					"-storageDataPath=" + filepath.Join(tc.Dir(), "vmstorage2"),
-					"-retentionPeriod=" + retentionPeriod,
-					"-maxBackfillAge=" + maxBackfillAge,
-				},
-				VminsertInstance: "vminsert",
-				VminsertFlags:    []string{},
-				VmselectInstance: "vmselect",
-				VmselectFlags:    []string{},
-			})
-		},
-		stop: func() {
-			tc.StopApp("vminsert")
-			tc.StopApp("vmselect")
-			tc.StopApp("vmstorage1")
-			tc.StopApp("vmstorage2")
-		},
-	}
-
-	testMaxBackfillAge(tc, opts)
-}
-
-type maxBackfillAgeOpts struct {
-	start func(retentionPeriod, maxBackfillAge string) apptest.PrometheusWriteQuerier
-	stop  func()
-}
-
-func testMaxBackfillAge(tc *apptest.TestCase, opts maxBackfillAgeOpts) {
-	t := tc.T()
-
-	assertSeries := func(app apptest.PrometheusQuerier, prefix string, start, end int64, want []map[string]string) {
-		t.Helper()
-
-		query := fmt.Sprintf(`{__name__=~"metric_%s.*"}`, prefix)
-		tc.Assert(&apptest.AssertOptions{
-			Msg: "unexpected /api/v1/series response",
-			Got: func() any {
-				return app.PrometheusAPIV1Series(t, query, apptest.QueryOpts{
-					Start: fmt.Sprintf("%d", start),
-					End:   fmt.Sprintf("%d", end),
-				}).Sort()
-			},
-			Want: &apptest.PrometheusAPIV1SeriesResponse{
-				Status: "success",
-				Data:   want,
-			},
-			FailNow: true,
-		})
-	}
-
-	assertQueryResults := func(app apptest.PrometheusQuerier, prefix string, start, end, step int64, want []*apptest.QueryResult) {
-		t.Helper()
-
-		query := fmt.Sprintf(`{__name__=~"metric_%s.*"}`, prefix)
-		tc.Assert(&apptest.AssertOptions{
-			Msg: "unexpected /api/v1/query_range response",
-			Got: func() any {
-				return app.PrometheusAPIV1QueryRange(t, query, apptest.QueryOpts{
-					Start:       fmt.Sprintf("%d", start),
-					End:         fmt.Sprintf("%d", end),
-					Step:        fmt.Sprintf("%dms", step),
-					MaxLookback: fmt.Sprintf("%dms", step-1),
-					NoCache:     "1",
-				})
-			},
-			Want: &apptest.PrometheusAPIV1QueryResponse{
-				Status: "success",
-				Data: &apptest.QueryData{
-					ResultType: "matrix",
-					Result:     want,
-				},
-			},
-			FailNow: true,
-		})
-	}
-
-	const numMetrics = 1000
-	now := time.Now().UTC()
-	var start, end, step int64
-	emptySeries := []map[string]string{}
-	emptyQueryResults := []*apptest.QueryResult{}
-
-	// Start sut with the same -retentionPeriod and -maxBackfillAge.
-	sut := opts.start("1y", "1y")
-
-	// Verify that samples older than the retention period are rejected.
-	start = now.Add(-365 * 24 * time.Hour).Add(-time.Hour).UnixMilli()
-	end = now.Add(-365 * 24 * time.Hour).UnixMilli()
-	step = (end - start) / numMetrics
-	outsideRetention := genMaxBackfillAgeData("outside_retention", numMetrics, start, step)
-	sut.PrometheusAPIV1ImportPrometheus(t, outsideRetention.samples, apptest.QueryOpts{})
-	sut.ForceFlush(t)
-	assertSeries(sut, "outside_retention", start, end, emptySeries)
-	assertQueryResults(sut, "outside_retention", start, end, step, emptyQueryResults)
-
-	// Verify that samples within the retention period are accepted and
-	// searcheable.
-	start = now.Add(-365 * 24 * time.Hour).Add(time.Hour).UnixMilli()
-	end = now.Add(-365 * 24 * time.Hour).Add(2 * time.Hour).UnixMilli()
-	step = (end - start) / numMetrics
-	insideRetention := genMaxBackfillAgeData("inside_retention", numMetrics, start, step)
-	sut.PrometheusAPIV1ImportPrometheus(t, insideRetention.samples, apptest.QueryOpts{})
-	sut.ForceFlush(t)
-	assertSeries(sut, "inside_retention", start, end, insideRetention.wantSeries)
-	assertQueryResults(sut, "inside_retention", start, end, step, insideRetention.wantQueryResults)
-
-	// Restart sut with -maxBackfillAge shorter than the -retentionPeriod.
-	opts.stop()
-	sut = opts.start("1y", "6M")
-
-	// Verify that new samples older than max backfill age but still within the
-	// retention period are rejected but existing samples are still searcheable.
-	start = now.Add(-365 * 24 * time.Hour).Add(time.Hour).UnixMilli()
-	end = now.Add(-365 * 24 * time.Hour).Add(2 * time.Hour).UnixMilli()
-	step = (end - start) / numMetrics
-	insideRetention2 := genMaxBackfillAgeData("inside_retention2", numMetrics, start, step)
-	sut.PrometheusAPIV1ImportPrometheus(t, insideRetention2.samples, apptest.QueryOpts{})
-	sut.ForceFlush(t)
-	assertSeries(sut, "inside_retention2", start, end, emptySeries)
-	assertQueryResults(sut, "inside_retention2", start, end, step, emptyQueryResults)
-	assertSeries(sut, "inside_retention", start, end, insideRetention.wantSeries)
-	assertQueryResults(sut, "inside_retention", start, end, step, insideRetention.wantQueryResults)
-
-	// Verify that the metrics that are outside the backfill window can still
-	// be deleted.
-	sut.PrometheusAPIV1AdminTSDBDeleteSeries(t, `{__name__=~".*inside_retention.*"}`, apptest.QueryOpts{})
-	sut.ForceFlush(t)
-	assertSeries(sut, "inside_retention", start, end, emptySeries)
-	assertQueryResults(sut, "inside_retention", start, end, step, emptyQueryResults)
-
-	// Verify that the samples that are within the backfill window are accepted
-	// and searchable.
-	start = now.Add(-180 * 24 * time.Hour).UnixMilli()
-	end = now.Add(-180 * 24 * time.Hour).Add(1 * time.Hour).UnixMilli()
-	step = (end - start) / numMetrics
-	insideMaxBackfillAge := genMaxBackfillAgeData("inside_max_backfill_age", numMetrics, start, step)
-	sut.PrometheusAPIV1ImportPrometheus(t, insideMaxBackfillAge.samples, apptest.QueryOpts{})
-	sut.ForceFlush(t)
-	assertSeries(sut, "inside_max_backfill_age", start, end, insideMaxBackfillAge.wantSeries)
-	assertQueryResults(sut, "inside_max_backfill_age", start, end, step, insideMaxBackfillAge.wantQueryResults)
-
-	opts.stop()
-}
-
-type maxBackfillAgeData struct {
-	samples          []string
-	wantSeries       []map[string]string
-	wantQueryResults []*apptest.QueryResult
-}
-
-func genMaxBackfillAgeData(prefix string, numMetrics, start, step int64) maxBackfillAgeData {
-	samples := make([]string, numMetrics)
-	wantSeries := make([]map[string]string, numMetrics)
-	wantQueryResults := make([]*apptest.QueryResult, numMetrics)
-	for i := range numMetrics {
-		metricName := fmt.Sprintf("metric_%s_%04d", prefix, i)
-		labelName := fmt.Sprintf("label_%s_%04d", prefix, i)
-		labelValue := fmt.Sprintf("value_%s_%04d", prefix, i)
-		value := i
-		timestamp := start + i*step
-		samples[i] = fmt.Sprintf(`%s{%s="value", label="%s"} %d %d`, metricName, labelName, labelValue, value, timestamp)
-		wantSeries[i] = map[string]string{
-			"__name__": metricName,
-			labelName:  "value",
-			"label":    labelValue,
-		}
-		wantQueryResults[i] = &apptest.QueryResult{
-			Metric: map[string]string{
-				"__name__": metricName,
-				labelName:  "value",
-				"label":    labelValue,
-			},
-			Samples: []*apptest.Sample{{Timestamp: timestamp, Value: float64(value)}},
-		}
-	}
-	return maxBackfillAgeData{samples, wantSeries, wantQueryResults}
-}

docs/victoriametrics/README.md

@@ -403,9 +403,6 @@ Resources:
 * [Cardinality explorer blog post](https://victoriametrics.com/blog/cardinality-explorer/).
 * [skills/victoriametrics-cardinality-analysis](https://github.com/VictoriaMetrics/skills/blob/main/plugins/diagnostics/skills/victoriametrics-cardinality-analysis/SKILL.md) for [agent-assisted](https://docs.victoriametrics.com/ai-tools/#agent-skills) analysis.
 
-For monitoring or alerting on cardinality, use [vmestimator](https://docs.victoriametrics.com/victoriametrics/vmestimator/).
-vmestimator measures metrics cardinality across [arbitrary label dimensions](https://docs.victoriametrics.com/victoriametrics/vmestimator/#basic) in real time and exposes the [results as metrics](https://docs.victoriametrics.com/victoriametrics/vmestimator/#cardinality-metrics).
-
 ### Cardinality explorer statistic inaccuracy
 
 In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) each vmstorage tracks the stored time series individually.
@@ -1551,18 +1548,6 @@ For example, the following command starts VictoriaMetrics, which accepts samples
 /path/to/victoria-metrics -futureRetention=1y
 ```
 
-By default, VictoriaMetrics accepts samples with timestamps as old as the configured `-retentionPeriod` allows, e.g. it accepts backfilled
-historical data as long as it fits into the retention. If you need rejecting samples with historical timestamps older than the specified
-duration, then specify the desired duration via the `-maxBackfillAge` command-line flag. This can be useful for limiting ingestion of
-historical samples, for example, when older data has been moved to another storage tier (nvme/hdd, hot/cold).
-`-maxBackfillAge` cannot exceed the configured `-retentionPeriod` - bigger values are automatically clamped to `-retentionPeriod`.
-
-For example, the following command starts VictoriaMetrics, which rejects ingested samples with timestamps older than 2 days:
-
-```sh
-/path/to/victoria-metrics -maxBackfillAge=2d
-```
-
 ### Multiple retentions
 
 Distinct retentions for distinct time series can be configured via [retention filters](#retention-filters)

docs/victoriametrics/changelog/CHANGELOG.md

@@ -25,18 +25,15 @@ The sandbox cluster installation runs under the constant load generated by
 See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-releases/).
 
 ## tip
-* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): flush pending persistent queue data to chunk file before updating the metadata. This prevents the metadata writer offset from getting ahead of the chunk file size and avoids losing the persistent queue after an unclean shutdown. See [#11192](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11192).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): introduce obfuscation functionality for remote write. By setting `-remoteWrite.obfuscationLabels`, the values of the specific labels will be anonymized before they're sent to corresponding `-remoteWrite.url`. See [#10599](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10599).
 
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/victoriametrics/metricsql/): support `fill` modifiers to allow missing series on either side of a binary operation to be filled with a provided default value. See [#10598](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10598).
-* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): Improve background discovery performance for [http_sd](https://docs.victoriametrics.com/victoriametrics/sd_configs/#http_sd_configs) discovery. See [#8838](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8838).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): allow overriding `max_scrape_size` on a per-target basis via the `__max_scrape_size__` label during target relabeling. See [#11188](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11188).
-* FEATURE: [vmstorage](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): add `-maxBackfillAge` command-line flag for limiting ingestion of samples with historical timestamps, for example, when older data has been moved between storage tiers (nvme/hdd, hot/cold). See [#11199](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11199). Thanks to @AshwinRamaniPsg for contribution.
 
 * BUGFIX: `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): Now drops metadata blocks when communicating with vmstorage nodes over the legacy RPC protocol. To avoid this limitation, upgrade `vmstorage` to a version that supports the new RPC protocol (>= [v1.137.0](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/victoriametrics/changelog/CHANGELOG.md#v11370)). See [#11146](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11146).
 * BUGFIX: [vmbackup](https://docs.victoriametrics.com/victoriametrics/vmbackup/) and [vmbackupmanager](https://docs.victoriametrics.com/victoriametrics/vmbackupmanager/): retry S3 requests failing with `HTTP 429` status code or `TooManyRequests` error code. Previously such requests were not retried, so a short burst of rate limiting would fail the whole backup. See [#11218](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11218). Thanks to @gautamrizwani for contribution.
 * BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): properly apply limit to metrics metadata response. See [#11139](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11139).
 * BUGFIX: [vmui](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#vmui): keep only one header navigation dropdown (`Explore`, `Tools`) open at a time. Previously, hovering across two dropdowns could briefly leave both open due to the close delay. See [#11224](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11224). Thanks to @antedotee for contribution.
-* BUGFIX: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): fix a possible data race when processing OpenTelemetry metadata. See [#11238](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11238). Thanks to @nevgeny for contribution.
 
 ## [v1.147.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.147.0)
 
@@ -48,7 +45,6 @@ Released at 2026-07-06
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/): reduces CPU usage by 10% at [sharding among remote storages](https://docs.victoriametrics.com/victoriametrics/vmagent/#sharding-among-remote-storages). See [#11113](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11113). Thanks to @bennf for contribution.
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/), `vminsert` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): introduce `64KiB` size limit for `metric metadata` fields - `Unit`, `Help` and `MetricFamilyName`. See [#11128](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11128).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): reduce CPU usage for storing scrape target labels. See [#10919](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10919).
-* FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): expose `vm_data_size_bytes{type="storage/metaindex"}` and `vm_data_size_bytes{type="indexdb/metaindex"}` metrics for tracking memory occupied by metaindex data. See [#11204](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/11204). Thanks to @SamarthBagga for contribution.
 * FEATURE: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): add `optimize_repeated_binary_op_subexprs=1` query arg to [/api/v1/query_range](https://docs.victoriametrics.com/victoriametrics/keyconcepts/#range-query) for executing binary operator sides sequentially when they share the same optimized aggregate rollup result expression. This allows the second side to reuse rollup result cache populated by the first side. See [#10575](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10575). Thanks to @xhebox for the contribution.
 * FEATURE: [vmauth](https://docs.victoriametrics.com/victoriametrics/vmauth/): prevent possible password brute-force attacks with an artificial 2-3 second delay as recommended by [OWASP](https://owasp.org/Top10/2025/A07_2025-Authentication_Failures). See [#11180](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11180).
 * FEATURE: [alerts](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules): add `InvalidAuthTokenRequestErrors` alerting rule to [vmauth alerts](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/rules/alerts-vmauth.yml). The new rule notifies when vmauth receives requests with invalid or missing auth tokens, which may indicate a client misconfiguration, expired token use, or brute-force attack. See [#11180](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/11180).
@@ -215,7 +211,6 @@ Released at 2026-04-24
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): fix incorrect evaluation of binary operations caused by an ordering bug (e.g. `10 - (3 + 3 + 4)` being evaluated as `10 - 3 + 3 + 4`). The issue was introduced in v1.140.0, v1.136.4, and v1.122.19. See [#10856](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10856).
 
 ## [v1.140.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.140.0)
-
 Released at 2026-04-10
 
 **Update Note 1:** [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/): [CSV export](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-export-csv-data) (`/api/v1/export/csv`) now adds a header row as the first line of the response, so existing CSV-processing scripts may need to skip this header. See [#10666](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10666).

docs/victoriametrics/vmagent.md

@@ -323,7 +323,8 @@ flowchart TB
       H2 --> H3[per-url <a href="https://docs.victoriametrics.com/victoriametrics/stream-aggregation">aggregation</a><br><b>-remoteWrite.streamAggr.config</b><br><b>-remoteWrite.streamAggr.dedupInterval</b>]
       H3 --> H4["per-url <a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#calculating-disk-space-for-persistence-queue">queue</a> (default: enabled)<br><b>-remoteWrite.disableOnDiskQueue</b>"]
       H4 --> H5[<a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#adding-labels-to-metrics">add extra labels</a><br><b>-remoteWrite.label</b>]
-      H5 --> H6[[push to <b>-remoteWrite.url</b>]]
+      H5 --> H6[<a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#obfuscating-label-values">obfuscate labels</a><br><b>-remoteWrite.obfuscationLabels</b>]
+      H6 --> H7[[push to <b>-remoteWrite.url</b>]]
 
       %% Right branch
       G --> R1[per-url <a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#monitoring-data-exchange">mdx filter</a><br><b>-remoteWrite.mdx.enable</b>]
@@ -331,7 +332,8 @@ flowchart TB
       R2 --> R3[per-url <a href="https://docs.victoriametrics.com/victoriametrics/stream-aggregation">aggregation</a><br><b>-remoteWrite.streamAggr.config</b><br><b>-remoteWrite.streamAggr.dedupInterval</b>]
       R3 --> R4["per-url <a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#calculating-disk-space-for-persistence-queue">queue</a> (default: enabled)<br><b>-remoteWrite.disableOnDiskQueue</b>"]
       R4 --> R5[<a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#adding-labels-to-metrics">add extra labels</a><br><b>-remoteWrite.label</b>]
-      R5 --> R6[[push to <b>-remoteWrite.url</b>]]
+      R5 --> R6[<a href="https://docs.victoriametrics.com/victoriametrics/vmagent/#obfuscating-label-values">obfuscate labels</a><br><b>-remoteWrite.obfuscationLabels</b>]
+      R6 --> R7[[push to <b>-remoteWrite.url</b>]]
 ```
 
 Scraping has additional settings that can be applied before samples are pushed to the processing pipeline above:
@@ -646,6 +648,25 @@ Extra labels can be added to metrics collected by `vmagent` via the following me
   /path/to/vmagent -remoteWrite.url=http://127.0.0.1:8428/api/v1/write?extra_label="env=prod"
   ```
 
+## Obfuscating label values
+
+Before sending metrics to `-remoteWrite.url`, `vmagent` can obfuscate the values of specific labels by using `-remoteWrite.obfuscationLabels`.
+
+This is useful when one or more `-remoteWrite.url` endpoints point to external services, such as monitoring vendors outside the department or company. 
+To meet security and compliance requirements, sensitive label values such as `ip`, `host`, `instance`, or `datacenter` can be obfuscated before metrics are sent to these external systems.
+
+Use `-remoteWrite.obfuscationLabels` to specify which labels should have their values obfuscated for the corresponding `-remoteWrite.url`. Multiple label names must be separated with `^^`.
+
+```sh
+./vmagent \
+  -remoteWrite.url=http://<external-service1> \
+  -remoteWrite.obfuscationLabels='instance^^datacenter' \
+  -remoteWrite.url=http://<external-service2> \
+  -remoteWrite.obfuscationLabels='instance' \
+  -remoteWrite.url=http://<internal-service> \
+  -remoteWrite.obfuscationLabels='' 
+```
+
 ## Automatically generated metrics
 
 `vmagent` automatically generates the following metrics for each scrape of every [Prometheus-compatible target](#how-to-collect-metrics-in-prometheus-format)

lib/mergeset/part.go

@@ -78,8 +78,7 @@ type part struct {
 
 	size uint64
 
-	mrs                []metaindexRow
-	metaindexSizeBytes uint64
+	mrs []metaindexRow
 
 	indexFile fs.MustReadAtCloser
 	itemsFile fs.MustReadAtCloser
@@ -132,7 +131,6 @@ func newPart(ph *partHeader, path string, size uint64, metaindexReader filestrea
 	p.path = path
 	p.size = size
 	p.mrs = mrs
-	p.metaindexSizeBytes = metaindexSizeBytes(mrs)
 
 	p.indexFile = indexFile
 	p.itemsFile = itemsFile
@@ -157,14 +155,6 @@ func (p *part) MustClose() {
 	ibSparseCache.RemoveBlocksForPart(p)
 }
 
-func metaindexSizeBytes(mrs []metaindexRow) uint64 {
-	n := uint64(cap(mrs)) * uint64(unsafe.Sizeof(metaindexRow{}))
-	for i := range mrs {
-		n += uint64(cap(mrs[i].firstItem))
-	}
-	return n
-}
-
 type indexBlock struct {
 	bhs []blockHeader
 

lib/mergeset/table.go

@@ -584,8 +584,6 @@ type TableMetrics struct {
 	PartsRefCount uint64
 
 	TooLongItemsDroppedTotal uint64
-
-	MetaindexSizeBytes uint64
 }
 
 // TotalItemsCount returns the total number of items in the table.
@@ -619,7 +617,6 @@ func (tb *Table) UpdateMetrics(m *TableMetrics) {
 		m.InmemoryBlocksCount += p.ph.blocksCount
 		m.InmemoryItemsCount += p.ph.itemsCount
 		m.InmemorySizeBytes += p.size
-		m.MetaindexSizeBytes += p.metaindexSizeBytes
 		m.PartsRefCount += uint64(pw.refCount.Load())
 	}
 
@@ -629,7 +626,6 @@ func (tb *Table) UpdateMetrics(m *TableMetrics) {
 		m.FileBlocksCount += p.ph.blocksCount
 		m.FileItemsCount += p.ph.itemsCount
 		m.FileSizeBytes += p.size
-		m.MetaindexSizeBytes += p.metaindexSizeBytes
 		m.PartsRefCount += uint64(pw.refCount.Load())
 	}
 	tb.partsLock.Unlock()

lib/persistentqueue/persistentqueue.go

@@ -52,7 +52,6 @@ type queue struct {
 	writerFlushedOffset uint64
 
 	lastMetainfoFlushTime uint64
-	hasDataToFlush        bool
 
 	blocksDropped *metrics.Counter
 	bytesDropped  *metrics.Counter
@@ -85,7 +84,6 @@ func (q *queue) mustResetFiles() {
 	}
 	q.reader.MustClose()
 	q.writer.MustClose()
-	q.hasDataToFlush = false
 	fs.MustRemovePath(q.readerPath)
 
 	q.writerOffset = 0
@@ -320,7 +318,6 @@ func tryOpeningQueue(path, name string, chunkFileSize, maxBlockSize, maxPendingB
 func (q *queue) MustClose() {
 	// Close writer.
 	q.writer.MustClose()
-	q.hasDataToFlush = false
 	q.writer = nil
 
 	// Close reader.
@@ -417,7 +414,7 @@ func (q *queue) writeBlock(block []byte) error {
 	}
 	q.blocksWritten.Inc()
 	q.bytesWritten.Add(len(block))
-	return q.flushBufAndMetainfoIfNeeded()
+	return q.flushWriterMetainfoIfNeeded()
 }
 
 var writeDurationSeconds = metrics.NewFloatCounter(`vm_persistentqueue_write_duration_seconds_total`)
@@ -425,7 +422,6 @@ var writeDurationSeconds = metrics.NewFloatCounter(`vm_persistentqueue_write_dur
 func (q *queue) nextChunkFileForWrite() error {
 	// Finalize the current chunk and start new one.
 	q.writer.MustClose()
-	q.hasDataToFlush = false
 	// There is no need to do fs.MustSyncPath(q.writerPath) here,
 	// since MustClose already does this.
 	if n := q.writerOffset % q.chunkFileSize; n > 0 {
@@ -517,7 +513,7 @@ again:
 	}
 	q.blocksRead.Inc()
 	q.bytesRead.Add(int(blockLen))
-	if err := q.flushBufAndMetainfoIfNeeded(); err != nil {
+	if err := q.flushReaderMetainfoIfNeeded(); err != nil {
 		return dst, err
 	}
 	return dst, nil
@@ -570,7 +566,6 @@ func (q *queue) write(buf []byte) error {
 	}
 	q.writerLocalOffset += bufLen
 	q.writerOffset += bufLen
-	q.hasDataToFlush = true
 	return nil
 }
 
@@ -600,16 +595,24 @@ func (q *queue) checkReaderWriterOffsets() error {
 	return nil
 }
 
-func (q *queue) flushBufAndMetainfoIfNeeded() error {
+func (q *queue) flushReaderMetainfoIfNeeded() error {
 	t := fasttime.UnixTimestamp()
 	if t == q.lastMetainfoFlushTime {
 		return nil
 	}
-	if q.hasDataToFlush {
-		q.writer.MustFlush(true)
-		q.writerFlushedOffset = q.writerOffset
-		q.hasDataToFlush = false
+	if err := q.flushMetainfo(); err != nil {
+		return fmt.Errorf("cannot flush metainfo: %w", err)
 	}
+	q.lastMetainfoFlushTime = t
+	return nil
+}
+
+func (q *queue) flushWriterMetainfoIfNeeded() error {
+	t := fasttime.UnixTimestamp()
+	if t == q.lastMetainfoFlushTime {
+		return nil
+	}
+	q.writer.MustFlush(true)
 	if err := q.flushMetainfo(); err != nil {
 		return fmt.Errorf("cannot flush metainfo: %w", err)
 	}

lib/persistentqueue/persistentqueue_synctest_test.go

@@ -1,60 +0,0 @@
-//go:build synctest
-
-package persistentqueue
-
-import (
-	"testing"
-	"testing/synctest"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
-)
-
-func TestFlushReaderMetainfoFlushesPendingWriterData(t *testing.T) {
-	synctest.Test(t, func(t *testing.T) {
-		path := "queue-flush-reader-metainfo"
-		fs.MustRemoveDir(path)
-		q := mustOpen(path, "foobar", 0)
-		defer func() {
-			q.MustClose()
-			fs.MustRemoveDir(path)
-		}()
-
-		block := []byte("foobar")
-		data := encoding.MarshalUint64(nil, uint64(len(block)))
-		data = append(data, block...)
-		// it will call `flushBufAndMetainfoIfNeeded` internally to flush the data and metadata.
-		err := q.writeBlock(data)
-		if err != nil {
-			t.Fatalf("unexpected error when writing data to queue: %s", err)
-		}
-		// the second call will update the writeOffset in memory without flushing the data and metadata,
-		// because the last flush was performed less than 1 second ago.
-		err = q.writeBlock(data)
-		if err != nil {
-			t.Fatalf("unexpected error when writing data to queue: %s", err)
-		}
-		time.Sleep(2 * time.Second)
-
-		// it will call `flushBufAndMetainfoIfNeeded` internally to flush the data and metadata.
-		if _, err = q.readBlock(nil); err != nil {
-			t.Fatalf("unexpected error when flushing reader metainfo: %s", err)
-		}
-
-		if fileSize := fs.MustFileSize(q.writerPath); fileSize != q.writerOffset {
-			t.Fatalf("unexpected writer file size after flushing reader metainfo; got %d bytes; want %d bytes", fileSize, q.writerOffset)
-		}
-		var mi metainfo
-		if err := mi.ReadFromFile(q.metainfoPath()); err != nil {
-			t.Fatalf("cannot read metainfo: %s", err)
-		}
-		if mi.ReaderOffset != q.readerOffset {
-			t.Fatalf("unexpected ReaderOffset in metainfo; got %d; want %d", mi.ReaderOffset, q.readerOffset)
-		}
-		if mi.WriterOffset != q.writerOffset {
-			t.Fatalf("unexpected WriterOffset in metainfo; got %d; want %d", mi.WriterOffset, q.writerOffset)
-		}
-
-	})
-}

lib/promscrape/config.go

@@ -367,9 +367,6 @@ func (sc *ScrapeConfig) mustStart(baseDir string) {
 	for i := range sc.KubernetesSDConfigs {
 		sc.KubernetesSDConfigs[i].MustStart(baseDir, swosFunc)
 	}
-	for i := range sc.HTTPSDConfigs {
-		sc.HTTPSDConfigs[i].MustStart(baseDir)
-	}
 }
 
 func (sc *ScrapeConfig) mustStop() {

lib/promscrape/discovery/http/api.go

@@ -1,41 +1,25 @@
 package http
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/url"
 	"strconv"
-	"sync"
-	"sync/atomic"
-	"time"
-
-	"github.com/VictoriaMetrics/metrics"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
+	"github.com/VictoriaMetrics/metrics"
 )
 
+var configMap = discoveryutil.NewConfigMap()
+
 type apiConfig struct {
-	client        *discoveryutil.Client
-	path          string
-	sourceURL     string
-	checkInterval time.Duration
+	client *discoveryutil.Client
+	path   string
 
 	fetchErrors *metrics.Counter
 	parseErrors *metrics.Counter
-
-	initOnce        sync.Once
-	prevAPIResponse atomic.Pointer[[]byte]
-	targetLabels    atomic.Pointer[targetLabelsResult]
-
-	wg sync.WaitGroup
-}
-
-type targetLabelsResult struct {
-	labels []*promutil.Labels
-	err    error
 }
 
 // httpGroupTarget represent prometheus GroupTarget
@@ -65,89 +49,37 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
 	}
 	cfg := &apiConfig{
-		client:        client,
-		path:          parsedURL.RequestURI(),
-		sourceURL:     sdc.URL,
-		checkInterval: max(*SDCheckInterval/2, time.Second),
-		fetchErrors:   metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="fetch",url=%q}`, sdc.URL)),
-		parseErrors:   metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="parse",url=%q}`, sdc.URL)),
+		client:      client,
+		path:        parsedURL.RequestURI(),
+		fetchErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="fetch",url=%q}`, sdc.URL)),
+		parseErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="parse",url=%q}`, sdc.URL)),
 	}
-	cfg.wg.Go(func() {
-		cfg.run()
-	})
 	return cfg, nil
 }
 
-func (cfg *apiConfig) init() {
-	cfg.initOnce.Do(func() {
-		cfg.refreshTargetsIfNeeded()
-	})
-}
-
-func (cfg *apiConfig) run() {
-	cfg.init()
-
-	ticker := time.NewTicker(cfg.checkInterval)
-	defer ticker.Stop()
-	stopCh := cfg.client.Context().Done()
-	for {
-		select {
-		case <-ticker.C:
-			cfg.refreshTargetsIfNeeded()
-		case <-stopCh:
-			return
-		}
-	}
-}
-
-func (cfg *apiConfig) refreshTargetsIfNeeded() {
-	apiResponse, err := cfg.getAPIResponseData()
+func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	v, err := configMap.Get(sdc, func() (any, error) { return newAPIConfig(sdc, baseDir) })
 	if err != nil {
-		cfg.targetLabels.Store(&targetLabelsResult{err: err})
-		cfg.prevAPIResponse.Store(nil)
-		return
+		return nil, err
 	}
-	prevAPIResponse := cfg.prevAPIResponse.Load()
-	if prevAPIResponse != nil && bytes.Equal(apiResponse, *prevAPIResponse) {
-		return
-	}
-	hts, err := parseAPIResponse(apiResponse, cfg.path)
-	if err != nil {
-		cfg.prevAPIResponse.Store(nil)
-		cfg.parseErrors.Inc()
-		cfg.targetLabels.Store(&targetLabelsResult{err: err})
-		return
-	}
-	newTargets := addHTTPTargetLabels(hts, cfg.sourceURL)
-	cfg.targetLabels.Store(&targetLabelsResult{labels: newTargets})
-	cfg.prevAPIResponse.Store(&apiResponse)
+	return v.(*apiConfig), nil
 }
 
-func (cfg *apiConfig) getAPIResponseData() ([]byte, error) {
+func getHTTPTargets(cfg *apiConfig) ([]httpGroupTarget, error) {
 	data, err := cfg.client.GetAPIResponseWithReqParams(cfg.path, func(request *http.Request) {
-		request.Header.Set("X-Prometheus-Refresh-Interval-Seconds", strconv.FormatFloat(cfg.checkInterval.Seconds(), 'f', 0, 64))
+		request.Header.Set("X-Prometheus-Refresh-Interval-Seconds", strconv.FormatFloat(SDCheckInterval.Seconds(), 'f', 0, 64))
 		request.Header.Set("Accept", "application/json")
 	})
 	if err != nil {
 		cfg.fetchErrors.Inc()
 		return nil, fmt.Errorf("cannot read http_sd api response: %w", err)
 	}
-	return data, nil
-}
-
-func (cfg *apiConfig) getLabels() ([]*promutil.Labels, error) {
-	cfg.init()
-
-	tlr := cfg.targetLabels.Load()
-	if tlr.err != nil {
-		return nil, tlr.err
+	tg, err := parseAPIResponse(data, cfg.path)
+	if err != nil {
+		cfg.parseErrors.Inc()
+		return nil, err
 	}
-	return tlr.labels, nil
-}
-
-func (cfg *apiConfig) mustStop() {
-	cfg.client.Stop()
-	cfg.wg.Wait()
+	return tg, nil
 }
 
 func parseAPIResponse(data []byte, path string) ([]httpGroupTarget, error) {

lib/promscrape/discovery/http/http.go

@@ -23,35 +23,19 @@ type SDConfig struct {
 	HTTPClientConfig  promauth.HTTPClientConfig  `yaml:",inline"`
 	ProxyURL          *proxy.URL                 `yaml:"proxy_url,omitempty"`
 	ProxyClientConfig promauth.ProxyClientConfig `yaml:",inline"`
-
-	cfg      *apiConfig
-	startErr error
-}
-
-// MustStart initializes sdc before its usage.
-func (sdc *SDConfig) MustStart(baseDir string) {
-	cfg, err := newAPIConfig(sdc, baseDir)
-	if err != nil {
-		sdc.startErr = fmt.Errorf("cannot create API config for http_sd: %w", err)
-		return
-	}
-	sdc.cfg = cfg
 }
 
 // GetLabels returns http service discovery labels according to sdc.
 func (sdc *SDConfig) GetLabels(baseDir string) ([]*promutil.Labels, error) {
-	if sdc.cfg == nil {
-		return nil, sdc.startErr
+	cfg, err := getAPIConfig(sdc, baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot get API config: %w", err)
 	}
-	return sdc.cfg.getLabels()
-}
-
-// MustStop stops further usage for sdc.
-func (sdc *SDConfig) MustStop() {
-	if sdc.cfg == nil {
-		return
+	hts, err := getHTTPTargets(cfg)
+	if err != nil {
+		return nil, err
 	}
-	sdc.cfg.mustStop()
+	return addHTTPTargetLabels(hts, sdc.URL), nil
 }
 
 func addHTTPTargetLabels(src []httpGroupTarget, sourceURL string) []*promutil.Labels {
@@ -70,3 +54,12 @@ func addHTTPTargetLabels(src []httpGroupTarget, sourceURL string) []*promutil.La
 	}
 	return ms
 }
+
+// MustStop stops further usage for sdc.
+func (sdc *SDConfig) MustStop() {
+	v := configMap.Delete(sdc)
+	if v != nil {
+		cfg := v.(*apiConfig)
+		cfg.client.Stop()
+	}
+}

lib/promscrape/discovery/http/http_test.go

@@ -1,9 +1,6 @@
 package http
 
 import (
-	"net/http"
-	"net/http/httptest"
-	"sync/atomic"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutil"
@@ -41,121 +38,3 @@ func TestAddHTTPTargetLabels(t *testing.T) {
 	}
 	f(src, labelssExpected)
 }
-
-func TestSDConfigGetLabels(t *testing.T) {
-
-	type apiResponse struct {
-		statusCode int
-		body       string
-	}
-	var currentResponse atomic.Pointer[apiResponse]
-
-	// add initial non-empty response
-	currentResponse.Store(&apiResponse{
-		body:       `[{"targets":["10.0.0.2:9100"],"labels":{"job":"node"}}]`,
-		statusCode: http.StatusOK,
-	})
-
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		resp := currentResponse.Load()
-		w.WriteHeader(resp.statusCode)
-		_, _ = w.Write([]byte(resp.body))
-	}))
-	defer srv.Close()
-
-	sdc := &SDConfig{
-		URL: srv.URL,
-	}
-	sdc.MustStart(".")
-	defer sdc.MustStop()
-
-	assertLabelss := func(expectedLabelss []*promutil.Labels) {
-		t.Helper()
-		got, err := sdc.GetLabels(".")
-		if err != nil {
-			t.Fatalf("unexpected GetLabels error: %s", err)
-		}
-		if len(got) == 0 && len(expectedLabelss) == 0 {
-			return
-		}
-		discoveryutil.TestEqualLabelss(t, got, expectedLabelss)
-	}
-
-	// check initial state, it must be non-empty
-	// it also inits apiConfig below
-	assertLabelss([]*promutil.Labels{
-		promutil.NewLabelsFromMap(map[string]string{
-			"__address__": "10.0.0.2:9100",
-			"job":         "node",
-			"__meta_url":  srv.URL,
-		}),
-	})
-
-	updateAPIResponse := func(response apiResponse) {
-		currentResponse.Store(&response)
-		sdc.cfg.refreshTargetsIfNeeded()
-
-	}
-
-	// change response to empty
-	updateAPIResponse(apiResponse{
-		statusCode: http.StatusOK,
-		body:       `[]`,
-	})
-	assertLabelss([]*promutil.Labels{})
-
-	// change response to non-empty
-	updateAPIResponse(apiResponse{
-		statusCode: http.StatusOK,
-		body:       `[{"targets":["10.0.0.1:9100"],"labels":{"job":"node"}},{"targets":["10.0.0.5:8429"],"labels":{"job":"vmagent"}}]`,
-	})
-	assertLabelss([]*promutil.Labels{
-		promutil.NewLabelsFromMap(map[string]string{
-			"__address__": "10.0.0.1:9100",
-			"job":         "node",
-			"__meta_url":  srv.URL,
-		}),
-		promutil.NewLabelsFromMap(map[string]string{
-			"__address__": "10.0.0.5:8429",
-			"job":         "vmagent",
-			"__meta_url":  srv.URL,
-		}),
-	})
-
-	// change response to error
-	updateAPIResponse(apiResponse{
-		statusCode: http.StatusServiceUnavailable,
-		body:       `Internal Server Error`,
-	})
-	_, err := sdc.GetLabels(".")
-	if err == nil {
-		t.Fatalf("unexpected empty error")
-	}
-
-	// transit back to correct api response
-	updateAPIResponse(apiResponse{
-		statusCode: http.StatusOK,
-		body:       `[{"targets":["10.0.0.1:9100"],"labels":{"job":"node"}},{"targets":["10.0.0.5:8429"],"labels":{"job":"vmagent"}}]`,
-	})
-	assertLabelss([]*promutil.Labels{
-		promutil.NewLabelsFromMap(map[string]string{
-			"__address__": "10.0.0.1:9100",
-			"job":         "node",
-			"__meta_url":  srv.URL,
-		}),
-		promutil.NewLabelsFromMap(map[string]string{
-			"__address__": "10.0.0.5:8429",
-			"job":         "vmagent",
-			"__meta_url":  srv.URL,
-		}),
-	})
-
-	// make sure that api response is properly cached
-	before := sdc.cfg.targetLabels.Load()
-	updateAPIResponse(apiResponse{statusCode: http.StatusOK,
-		body: `[{"targets":["10.0.0.1:9100"],"labels":{"job":"node"}},{"targets":["10.0.0.5:8429"],"labels":{"job":"vmagent"}}]`})
-
-	if sdc.cfg.targetLabels.Load() != before {
-		t.Fatalf("expected identical response to be deduplicated")
-	}
-}

lib/storage/index_db_test.go

@@ -1949,11 +1949,10 @@ func newTestStorage() *Storage {
 	s := &Storage{
 		cachePath: "test-storage-cache",
 
-		metricIDCache:       workingsetcache.New(1234),
-		metricNameCache:     workingsetcache.New(1234),
-		tsidCache:           workingsetcache.New(1234),
-		retentionMsecs:      retentionMax.Milliseconds(),
-		maxBackfillAgeMsecs: retentionMax.Milliseconds(),
+		metricIDCache:   workingsetcache.New(1234),
+		metricNameCache: workingsetcache.New(1234),
+		tsidCache:       workingsetcache.New(1234),
+		retentionMsecs:  retentionMax.Milliseconds(),
 	}
 	return s
 }

lib/storage/part.go

@@ -42,8 +42,7 @@ type part struct {
 	valuesFile     fs.MustReadAtCloser
 	indexFile      fs.MustReadAtCloser
 
-	metaindex          []metaindexRow
-	metaindexSizeBytes uint64
+	metaindex []metaindexRow
 }
 
 // mustOpenFilePart opens file-based part from the given path.
@@ -103,7 +102,6 @@ func newPart(ph *partHeader, path string, size uint64, metaindexReader filestrea
 	p.valuesFile = valuesFile
 	p.indexFile = indexFile
 	p.metaindex = metaindex
-	p.metaindexSizeBytes = metaindexSizeBytes(metaindex)
 
 	return &p
 }
@@ -130,10 +128,6 @@ func (p *part) MustClose() {
 	ibCache.RemoveBlocksForPart(p)
 }
 
-func metaindexSizeBytes(metaindex []metaindexRow) uint64 {
-	return uint64(cap(metaindex)) * uint64(unsafe.Sizeof(metaindexRow{}))
-}
-
 type indexBlock struct {
 	bhs []blockHeader
 }

lib/storage/partition.go

@@ -334,10 +334,9 @@ type partitionMetrics struct {
 	IndexBlocksCacheRequests     uint64
 	IndexBlocksCacheMisses       uint64
 
-	InmemorySizeBytes  uint64
-	SmallSizeBytes     uint64
-	BigSizeBytes       uint64
-	MetaindexSizeBytes uint64
+	InmemorySizeBytes uint64
+	SmallSizeBytes    uint64
+	BigSizeBytes      uint64
 
 	InmemoryRowsCount uint64
 	SmallRowsCount    uint64
@@ -398,7 +397,6 @@ func (pt *partition) UpdateMetrics(m *partitionMetrics) {
 		m.InmemoryRowsCount += p.ph.RowsCount
 		m.InmemoryBlocksCount += p.ph.BlocksCount
 		m.InmemorySizeBytes += p.size
-		m.MetaindexSizeBytes += p.metaindexSizeBytes
 		m.InmemoryPartsRefCount += uint64(pw.refCount.Load())
 		if isDedupScheduled {
 			m.ScheduledDownsamplingPartitionsSize += p.size
@@ -409,7 +407,6 @@ func (pt *partition) UpdateMetrics(m *partitionMetrics) {
 		m.SmallRowsCount += p.ph.RowsCount
 		m.SmallBlocksCount += p.ph.BlocksCount
 		m.SmallSizeBytes += p.size
-		m.MetaindexSizeBytes += p.metaindexSizeBytes
 		m.SmallPartsRefCount += uint64(pw.refCount.Load())
 		if isDedupScheduled {
 			m.ScheduledDownsamplingPartitionsSize += p.size
@@ -420,7 +417,6 @@ func (pt *partition) UpdateMetrics(m *partitionMetrics) {
 		m.BigRowsCount += p.ph.RowsCount
 		m.BigBlocksCount += p.ph.BlocksCount
 		m.BigSizeBytes += p.size
-		m.MetaindexSizeBytes += p.metaindexSizeBytes
 		m.BigPartsRefCount += uint64(pw.refCount.Load())
 		if isDedupScheduled {
 			m.ScheduledDownsamplingPartitionsSize += p.size

lib/storage/storage.go

@@ -65,7 +65,6 @@ type Storage struct {
 	cachePath                   string
 	retentionMsecs              int64
 	futureRetentionMsecs        int64
-	maxBackfillAgeMsecs         int64
 	denyQueriesOutsideRetention bool
 
 	// lock file for exclusive access to the storage on the given path.
@@ -165,7 +164,6 @@ type Storage struct {
 type OpenOptions struct {
 	Retention                   time.Duration
 	FutureRetention             time.Duration
-	MaxBackfillAge              time.Duration
 	DenyQueriesOutsideRetention bool
 	MaxHourlySeries             int
 	MaxDailySeries              int
@@ -189,10 +187,6 @@ func MustOpenStorage(path string, opts OpenOptions) *Storage {
 		retention = retentionMax
 	}
 	futureRetention := max(opts.FutureRetention, retention2Days)
-	maxBackfillAge := opts.MaxBackfillAge
-	if maxBackfillAge <= 0 || maxBackfillAge > retention {
-		maxBackfillAge = retention
-	}
 	idbPrefillStart := opts.IDBPrefillStart
 	if idbPrefillStart <= 0 {
 		idbPrefillStart = time.Hour
@@ -202,7 +196,6 @@ func MustOpenStorage(path string, opts OpenOptions) *Storage {
 		cachePath:                   filepath.Join(path, cacheDirname),
 		retentionMsecs:              retention.Milliseconds(),
 		futureRetentionMsecs:        futureRetention.Milliseconds(),
-		maxBackfillAgeMsecs:         maxBackfillAge.Milliseconds(),
 		denyQueriesOutsideRetention: opts.DenyQueriesOutsideRetention,
 		stopCh:                      make(chan struct{}),
 		idbPrefillStartSeconds:      idbPrefillStart.Milliseconds() / 1000,
@@ -1243,7 +1236,7 @@ func (s *Storage) checkTimeRange(tr TimeRange) error {
 		return nil
 	}
 
-	minTimestamp, maxTimestamp := s.tb.getMinMaxRetentionTimestamps()
+	minTimestamp, maxTimestamp := s.tb.getMinMaxTimestamps()
 	if minTimestamp <= tr.MinTimestamp && tr.MaxTimestamp <= maxTimestamp {
 		return nil
 	}
@@ -1903,7 +1896,7 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
 	var newSeriesCount uint64
 	var seriesRepopulated uint64
 
-	minTimestamp, maxTimestamp := s.tb.getMinMaxIngestionTimestamps()
+	minTimestamp, maxTimestamp := s.tb.getMinMaxTimestamps()
 
 	var lTSID legacyTSID
 	var ptw *partitionWrapper
@@ -1925,11 +1918,11 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
 			}
 		}
 		if mr.Timestamp < minTimestamp {
-			// Skip rows with too small timestamps outside the retention or -maxBackfillAge.
+			// Skip rows with too small timestamps outside the retention.
 			if firstWarn == nil {
 				metricName := getUserReadableMetricName(mr.MetricNameRaw)
-				firstWarn = fmt.Errorf("cannot insert row with too small timestamp %d; minimum allowed timestamp is %d; "+
-					"probably you need updating -retentionPeriod or -maxBackfillAge command-line flags; metricName: %s",
+				firstWarn = fmt.Errorf("cannot insert row with too small timestamp %d outside the retention; minimum allowed timestamp is %d; "+
+					"probably you need updating -retentionPeriod command-line flag; metricName: %s",
 					mr.Timestamp, minTimestamp, metricName)
 			}
 			s.tooSmallTimestampRows.Add(1)

lib/storage/storage_synctest_test.go

@@ -1410,80 +1410,3 @@ func TestStorage_denyQueriesOutsideRetention(t *testing.T) {
 
 	})
 }
-
-func TestStorageAddRows_MaxBackfillAge(t *testing.T) {
-	defer testRemoveAll(t)
-
-	mn := MetricName{
-		MetricGroup: []byte("metric"),
-	}
-	mr := MetricRow{
-		MetricNameRaw: mn.marshalRaw(nil),
-		Value:         123,
-	}
-
-	f := func(s *Storage, age time.Duration, want uint64) {
-		t.Helper()
-
-		mr.Timestamp = time.Now().UTC().Add(-age).UnixMilli()
-		s.AddRows([]MetricRow{mr}, defaultPrecisionBits)
-		s.DebugFlush()
-		if got := s.tooSmallTimestampRows.Load(); got != want {
-			t.Fatalf("unexpected number of tooSmallTimestampRows: got %d, want %d", got, want)
-		}
-	}
-
-	synctest.Test(t, func(t *testing.T) {
-		// synctest time begins at 2000-01-01T00:00:00Z.
-
-		retention1y := 365 * 24 * time.Hour
-		var s *Storage
-
-		s = MustOpenStorage(t.Name(), OpenOptions{
-			Retention: retention1y,
-			// By default MaxBackfillAge must be the same as Retention
-		})
-		// Verify that the sample with timestamp 1ms older than retention is
-		// rejected.
-		f(s, retention1y+time.Millisecond, 1)
-		// Verify that the sample with timestamp which is exactly at retention
-		// boundary is accepted.
-		f(s, retention1y, 1)
-
-		// Restart storage with negative MaxBackfillAge. In this case,
-		// MaxBackfillAge must be the same as Retention.
-		// Also advance time a bit so that the storage will not use the same
-		// nanosecond for creating a new part for storing the samples.
-		s.MustClose()
-		time.Sleep(time.Nanosecond)
-		s = MustOpenStorage(t.Name(), OpenOptions{
-			Retention:      retention1y,
-			MaxBackfillAge: -1,
-		})
-		f(s, retention1y+time.Millisecond, 1)
-		f(s, retention1y, 1)
-
-		// Restart storage with MaxBackfillAge bigger than Retention. In this
-		// case, MaxBackfillAge must be the same as Retention.
-		s.MustClose()
-		time.Sleep(time.Nanosecond)
-		s = MustOpenStorage(t.Name(), OpenOptions{
-			Retention:      retention1y,
-			MaxBackfillAge: retention1y + time.Millisecond,
-		})
-		f(s, retention1y+time.Millisecond, 1)
-		f(s, retention1y, 1)
-
-		// Restart storage with MaxBackfillAge smaller than Retention.
-		s.MustClose()
-		time.Sleep(time.Nanosecond)
-		s = MustOpenStorage(t.Name(), OpenOptions{
-			Retention:      retention1y,
-			MaxBackfillAge: retention1y - time.Millisecond,
-		})
-		f(s, retention1y, 1)
-		f(s, retention1y-time.Millisecond, 1)
-
-		s.MustClose()
-	})
-}

lib/storage/storage_test.go

@@ -1807,19 +1807,18 @@ func TestStorageRowsNotAdded(t *testing.T) {
 	defer testRemoveAll(t)
 
 	type options struct {
-		name           string
-		retention      time.Duration
-		maxBackfillAge time.Duration
-		mrs            []MetricRow
-		tr             TimeRange
-		wantMetrics    *Metrics
+		name        string
+		retention   time.Duration
+		mrs         []MetricRow
+		tr          TimeRange
+		wantMetrics *Metrics
 	}
 	f := func(opts *options) {
 		t.Helper()
 
 		var gotMetrics Metrics
 		path := fmt.Sprintf("%s/%s", t.Name(), opts.name)
-		s := MustOpenStorage(path, OpenOptions{Retention: opts.retention, MaxBackfillAge: opts.maxBackfillAge})
+		s := MustOpenStorage(path, OpenOptions{Retention: opts.retention})
 		defer s.MustClose()
 		s.AddRows(opts.mrs, defaultPrecisionBits)
 		s.DebugFlush()
@@ -1891,22 +1890,6 @@ func TestStorageRowsNotAdded(t *testing.T) {
 		},
 	})
 
-	retention = retentionMax
-	maxBackfillAge := 48 * time.Hour
-	minTimestamp = time.Now().Add(-maxBackfillAge - time.Hour).UnixMilli()
-	maxTimestamp = minTimestamp + 1000
-	f(&options{
-		name:           "TooSmallTimestampsForMaxBackfillAge",
-		retention:      retention,
-		maxBackfillAge: maxBackfillAge,
-		mrs:            testGenerateMetricRows(rng, numRows, minTimestamp, maxTimestamp),
-		tr:             TimeRange{minTimestamp, maxTimestamp},
-		wantMetrics: &Metrics{
-			RowsReceivedTotal:     numRows,
-			TooSmallTimestampRows: numRows,
-		},
-	})
-
 	minTimestamp = time.Now().UnixMilli()
 	maxTimestamp = minTimestamp + 1000
 	mrs = testGenerateMetricRows(rng, numRows, minTimestamp, maxTimestamp)

lib/storage/table.go

@@ -368,7 +368,7 @@ func (tb *table) MustAddRows(rows []rawRow) {
 	// The slowest path - there are rows that don't fit any existing partition.
 	// Create new partitions for these rows.
 	// Do this under tb.ptwsLock.
-	minTimestamp, maxTimestamp := tb.getMinMaxIngestionTimestamps()
+	minTimestamp, maxTimestamp := tb.getMinMaxTimestamps()
 	tb.ptwsLock.Lock()
 	for i := range missingRows {
 		r := &missingRows[i]
@@ -407,28 +407,9 @@ func (tb *table) MustGetIndexDBIDByHour(hour uint64) uint64 {
 	return ptw.pt.idb.id
 }
 
-// getMinMaxRetentionTimestamps returns the minimum and maximum timestamps
-// allowed by the configured -retentionPeriod and -futureRetention.
-//
-// It is used for checking whether the given time range is fully covered
-// by the retention, e.g. for -denyQueriesOutsideRetention.
-func (tb *table) getMinMaxRetentionTimestamps() (int64, int64) {
-	return tb.getMinMaxTimestampsForAge(tb.s.retentionMsecs)
-}
-
-// getMinMaxIngestionTimestamps returns the minimum and maximum timestamps
-// allowed for newly ingested rows.
-//
-// The minimum timestamp is bound by -maxBackfillAge instead of -retentionPeriod,
-// since -maxBackfillAge can be configured to reject backfilled rows with historical
-// timestamps stricter than the full -retentionPeriod window.
-func (tb *table) getMinMaxIngestionTimestamps() (int64, int64) {
-	return tb.getMinMaxTimestampsForAge(tb.s.maxBackfillAgeMsecs)
-}
-
-func (tb *table) getMinMaxTimestampsForAge(minAgeMsecs int64) (int64, int64) {
+func (tb *table) getMinMaxTimestamps() (int64, int64) {
 	now := int64(fasttime.UnixTimestamp() * 1000)
-	minTimestamp := now - minAgeMsecs
+	minTimestamp := now - tb.s.retentionMsecs
 	if minTimestamp < 0 {
 		// Negative timestamps aren't supported by the storage.
 		minTimestamp = 0