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/Articles.md

@@ -154,13 +154,6 @@ See [our blog](https://victoriametrics.com/blog) for the latest articles written
 * [Why irate from Prometheus doesn't capture spikes](https://valyala.medium.com/why-irate-from-prometheus-doesnt-capture-spikes-45f9896d7832)
 * [VictoriaMetrics: PromQL compliance](https://medium.com/@romanhavronenko/victoriametrics-promql-compliance-d4318203f51e)
 * [How do open source solutions for logs work: Elasticsearch, Loki and VictoriaLogs](https://itnext.io/how-do-open-source-solutions-for-logs-work-elasticsearch-loki-and-victorialogs-9f7097ecbc2f)
-* [How vmagent Collects and Ships Metrics Fast with Aggregation, Deduplication, and More](https://victoriametrics.com/blog/vmagent-how-it-works/)
-* [When Metrics Meet vminsert: A Data-Delivery Story](https://victoriametrics.com/blog/vminsert-how-it-works/)
-* [How vmstorage Handles Data Ingestion From vminsert](https://victoriametrics.com/blog/vmstorage-how-it-handles-data-ingestion/)
-* [How vmstorage Processes Data: Retention, Merging, Deduplication...](https://victoriametrics.com/blog/vmstorage-retention-merging-deduplication/)
-* [How vmstorage's IndexDB Works](https://victoriametrics.com/blog/vmstorage-how-indexdb-works/)
-* [How vmstorage Handles Query Requests From vmselect](https://victoriametrics.com/blog/vmstorage-how-it-handles-query-requests/)
-* [Inside vmselect: The Query Processing Engine of VictoriaMetrics](https://victoriametrics.com/blog/vmselect-how-it-works/)
 
 ### Tutorials, guides and how-to articles
 
@@ -178,12 +171,6 @@ See [our guides](https://docs.victoriametrics.com/guides/) for the up-to-date gu
 * [Prometheus storage: tech terms for humans](https://valyala.medium.com/prometheus-storage-technical-terms-for-humans-4ab4de6c3d48)
 * [Cardinality explorer](https://victoriametrics.com/blog/cardinality-explorer/)
 * [Rules backfilling via vmalert](https://victoriametrics.com/blog/rules-replay/)
-* [vmagent: Key Features Explained in Under 15 Minutes](https://victoriametrics.com/blog/vmagent-key-features-explained/)
-* [Prometheus Metrics Explained: Counters, Gauges, Histograms & Summaries](https://victoriametrics.com/blog/prometheus-monitoring-metrics-counters-gauges-histogram-summaries/)
-* [Prometheus Monitoring: Instant Queries and Range Queries Explained](https://victoriametrics.com/blog/prometheus-monitoring-instant-range-query/)
-* [Prometheus Monitoring: Functions, Subqueries, Operators, and Modifiers](https://victoriametrics.com/blog/prometheus-monitoring-function-operator-modifier/)
-* [Prometheus Alerting 101: Rules, Recording Rules, and Alertmanager](https://victoriametrics.com/blog/alerting-recording-rules-alertmanager/)
-* [Alerting Best Practices](https://victoriametrics.com/blog/alerting-best-practices/)
 
 ### Other articles
 

docs/victoriametrics/Cluster-VictoriaMetrics.md

@@ -59,11 +59,6 @@ It increases cluster availability, and simplifies cluster maintenance as well as
 
 ![Cluster Scheme](Cluster-VictoriaMetrics-components.webp)
 
-> Further reading, deep dives into how each service works internally:
-> - `vmstorage`: [How vmstorage Handles Data Ingestion From vminsert](https://victoriametrics.com/blog/vmstorage-how-it-handles-data-ingestion/), [How vmstorage's IndexDB Works](https://victoriametrics.com/blog/vmstorage-how-indexdb-works/), [How vmstorage Handles Query Requests From vmselect](https://victoriametrics.com/blog/vmstorage-how-it-handles-query-requests/).
-> - `vminsert`: [When Metrics Meet vminsert: A Data-Delivery Story](https://victoriametrics.com/blog/vminsert-how-it-works/).
-> - `vmselect`: [Inside vmselect: The Query Processing Engine of VictoriaMetrics](https://victoriametrics.com/blog/vmselect-how-it-works/).
-
 ## vmui
 
 VictoriaMetrics cluster version provides UI for query troubleshooting and exploration. The UI is available at
@@ -839,7 +834,7 @@ This ensures that incoming metrics are evenly distributed across all `vmstorage`
 The downside is that a single slow vmstorage node can throttle the entire cluster.
 
 When `-disableRerouting=false` is enabled on `vminsert`, 
-the cluster will automatically [re-route writes](https://victoriametrics.com/blog/vminsert-how-it-works/#31-rerouting) away from the slowest vmstorage node to preserve maximum ingestion throughput.
+the cluster will automatically re-route writes away from the slowest vmstorage node to preserve maximum ingestion throughput.
 
 Re-routing occurs only when all of the following conditions hold:
 - the storage send buffer is full.
@@ -882,7 +877,7 @@ See also [resource usage limits docs](#resource-usage-limits).
 
 ## Rebalancing
 
-Every `vminsert` node [evenly spreads (shards) incoming data](https://victoriametrics.com/blog/vminsert-how-it-works/#3-sharding-and-buffering) among `vmstorage` nodes specified in the `-storageNode` command-line flag.
+Every `vminsert` node evenly spreads (shards) incoming data among `vmstorage` nodes specified in the `-storageNode` command-line flag.
 This guarantees even distribution of the ingested data among `vmstorage` nodes. When new `vmstorage` nodes are added to the `-storageNode`
 command-line flag at `vminsert`, then only newly ingested data is distributed evenly among old and new `vmstorage` nodes, while
 historical data remains on the old `vmstorage` nodes. This speeds up data ingestion and querying for the majority of production workloads,
@@ -1030,7 +1025,7 @@ By default, VictoriaMetrics offloads replication to the underlying storage point
 which guarantees data durability. VictoriaMetrics supports application-level replication if replicated durable persistent disks cannot be used for some reason.
 
 The replication can be enabled by passing `-replicationFactor=N` command-line flag to `vminsert`. This instructs `vminsert` to store `N` copies for every ingested sample
-on `N` distinct `vmstorage` nodes. This guarantees that all the stored data remains available for querying if up to `N-1` `vmstorage` nodes are unavailable. See [how `vminsert` replicates each sample to `N` `vmstorage` nodes](https://victoriametrics.com/blog/vminsert-how-it-works/#4-replication-and-sending-data-to-vmstorage) for details.
+on `N` distinct `vmstorage` nodes. This guarantees that all the stored data remains available for querying if up to `N-1` `vmstorage` nodes are unavailable.
 
 Passing `-replicationFactor=N` command-line flag to `vmselect` instructs it to not mark responses as `partial` if less than `-replicationFactor` vmstorage nodes are unavailable during the query.
 See [cluster availability docs](#cluster-availability) for details.
@@ -1065,7 +1060,7 @@ deduplication can't be guaranteed when samples and sample duplicates for the sam
 - when `vmstorage` node has no enough capacity for processing incoming data stream. Then `vminsert` re-routes new samples to other `vmstorage` nodes.
 
 It is recommended to set **the same** `-dedup.minScrapeInterval` command-line flag value to both `vmselect` and `vmstorage` nodes
-to ensure query results consistency, even if [storage layer didn't complete deduplication](https://victoriametrics.com/blog/vmstorage-retention-merging-deduplication/#deduplication) yet.
+to ensure query results consistency, even if storage layer didn't complete deduplication yet.
 
 ## Metrics Metadata
 

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.
@@ -1405,9 +1402,6 @@ for fast block lookups, which belong to the given `TSID` and cover the given tim
 * various background maintenance tasks such as [de-duplication](#deduplication), [downsampling](#downsampling)
   and [freeing up disk space for the deleted time series](#how-to-delete-time-series) are performed during the merge
 
-See how `vmstorage` [selects parts for background merging](https://victoriametrics.com/blog/vmstorage-retention-merging-deduplication/#merge-process),
-including merge limits and monitoring metrics.
-
 Newly added `parts` either successfully appear in the storage or fail to appear.
 The newly added `part` is atomically registered in the `parts.json` file under the corresponding partition
 after it is fully written and [fsynced](https://man7.org/linux/man-pages/man2/fsync.2.html) to the storage.
@@ -1535,8 +1529,6 @@ are **eventually deleted** during [background merge](https://medium.com/@valyala
 The time range covered by data part is **not limited by retention period unit**. One data part can cover hours or days of
 data. Hence, a data part can be deleted only **when fully outside the configured retention**.
 See more about partitions and parts in the [Storage section](#storage).
-See how the [retention and free-disk watchers manage storage](https://victoriametrics.com/blog/vmstorage-retention-merging-deduplication/#retention-free-disk-space-guard-and-downsampling)
-for implementation details and monitoring metrics.
 
 The maximum disk space usage for a given `-retentionPeriod` is going to be (`-retentionPeriod` + 1) months.
 For example, if `-retentionPeriod` is set to 1, data for January is deleted on March 1st.
@@ -1556,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
+* 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.
-* 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).
 
 ## [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

@@ -141,7 +141,6 @@ to other remote storage systems that support Prometheus `remote_write` protocol
 If a single remote storage instance is temporarily unavailable, the collected data remains available on the other remote storage instances.
 `vmagent` buffers the collected data in files at `-remoteWrite.tmpDataPath` until the remote storage becomes available again.
 Then it sends the buffered data to the remote storage in order to prevent data gaps.
-See how `vmagent` [selects shards and places replicas](https://victoriametrics.com/blog/vmagent-how-it-works/#step-4-sharding--replication) for implementation details.
 
 [VictoriaMetrics cluster](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) already supports replication,
 so there is no need to specify multiple `-remoteWrite.url` flags when writing data to the same cluster.
@@ -151,8 +150,7 @@ See [these docs](https://docs.victoriametrics.com/victoriametrics/cluster-victor
 
 `vmagent` can add, remove, or update labels on the collected data before sending it to the remote storage.
 It can filter scrape targets or remove unwanted samples via Prometheus-like relabeling.
-Please see the [Relabeling cookbook](https://docs.victoriametrics.com/victoriametrics/relabeling/) for configuration examples.
-For ingestion pipeline internals, see how `vmagent` [applies global relabeling and cardinality limits](https://victoriametrics.com/blog/vmagent-how-it-works/#step-2-global-relabeling-cardinality-reduction).
+Please see the [Relabeling cookbook](https://docs.victoriametrics.com/victoriametrics/relabeling/) for details.
 
 ### Sharding among remote storages
 
@@ -270,9 +268,7 @@ for the collected samples. Examples:
   ```sh
   ./vmagent -remoteWrite.url=http://remote-storage/api/v1/write -streamAggr.dropInputLabels=replica -streamAggr.dedupInterval=60s
   ```
-
-See how `vmagent` [orders global deduplication and stream aggregation](https://victoriametrics.com/blog/vmagent-how-it-works/#step-3-global-deduplication--stream-aggregation) in the ingestion pipeline.
-
+  
 ### Monitoring Data eXchange
 
 The Monitoring Data eXchange (MDX){{% available_from "v1.147.0" %}} feature allows `vmagent` to forward only VictoriaMetrics metrics to selected `-remoteWrite.url` destinations while dropping metrics from non-VictoriaMetrics services.
@@ -327,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>]
@@ -335,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:
@@ -361,8 +359,6 @@ in addition to the pull-based Prometheus-compatible targets' scraping:
 * Prometheus exposition format via `http://<vmagent>:8429/api/v1/import/prometheus`. See [these docs](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-data-in-prometheus-exposition-format) for details.
 * Arbitrary CSV data via `http://<vmagent>:8429/api/v1/import/csv`. See [these docs](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-import-csv-data).
 
-See how `vmagent` [handles concurrency, decompression, and stream parsing](https://victoriametrics.com/blog/vmagent-how-it-works/#step-1-receiving-data-via-api-or-scrape) during ingestion.
-
 ## How to collect metrics in Prometheus format
 
 Specify the path to the `prometheus.yml` file via the `-promscrape.config` command-line flag. `vmagent` takes into account the following
@@ -652,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)
@@ -1004,7 +1019,6 @@ This behavior can be changed with the `-remoteWrite.inmemoryQueues` {{% availabl
 When set to a non-zero value, vmagent starts the given number of additional workers,
 which send only recently ingested data from the in-memory queue, while the workers configured via `-remoteWrite.queues` drain the file-based backlog concurrently.
 This reduces the delivery lag for fresh samples after remote storage outages or slowdowns. The flag can be set individually per each `-remoteWrite.url`.
-See how the [in-memory and file-based queues manage blocks](https://victoriametrics.com/blog/vmagent-how-it-works/#in-memory-queue) for implementation details.
 
 Note that these workers are started in addition to the workers configured via `-remoteWrite.queues`, so the total number of concurrent connections to
 the remote storage becomes the sum of both flags. Take this into account if the remote storage limits the number of concurrent requests.

docs/victoriametrics/vmalert.md

@@ -982,9 +982,6 @@ Try the following tips to avoid common issues:
     In that case, the default step will be used (`-datasource.queryStep`) and may cause unexpected results compared to
     executing this query in vmui/Grafana, where step is adjusted differently.
 
-See [practical examples for reducing alert noise](https://victoriametrics.com/blog/alerting-best-practices/#reducing-noise),
-including aggregating alerts and configuring inhibition.
-
 ### Rule state
 
 vmalert keeps the last `-rule.updateEntriesLimit` updates (or `update_entries_limit` [per-rule config](https://docs.victoriametrics.com/victoriametrics/vmalert/#alerting-rules))
@@ -1040,9 +1037,6 @@ Sometimes, it's hard to understand why a specific alert fired or not. Keep in mi
 If evaluation returns error (i.e. datasource is unavailable), alert state doesn't change.
 If at least one evaluation returns no data, then alert's `for` state resets.
 
-See [how to tune the `for` parameter](https://victoriametrics.com/blog/alerting-best-practices/#the-for-param),
-including its tradeoff with the query lookbehind window.
-
 > Note: The alert state is tracked separately for each time series returned during evaluation.
 > For example, if the 1st evaluation returns series A and B, and the 2nd evaluation returns only B – the alert will remain active **only for B**.
 

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

@@ -11,7 +11,9 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+	"unsafe"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
@@ -38,6 +40,14 @@ const maxInmemoryParts = 60
 // See appendPartsToMerge tests for details.
 const defaultPartsToMerge = 15
 
+// The number of shards for rawRow entries per partition.
+//
+// Higher number of shards reduces CPU contention and increases the max bandwidth on multi-core systems.
+var rawRowsShardsPerPartition = cgroup.AvailableCPUs()
+
+// The interval for flushing buffered rows into parts, so they become visible to search.
+const pendingRowsFlushInterval = 2 * time.Second
+
 // The interval for guaranteed flush of recently ingested data from memory to on-disk parts, so they survive process crash.
 var dataFlushInterval = 5 * time.Second
 
@@ -56,6 +66,11 @@ func SetDataFlushInterval(d time.Duration) {
 	dataFlushInterval = d
 }
 
+// The maximum number of rawRow items in rawRowsShard.
+//
+// Limit the maximum shard size to 8Mb, since this gives the lowest CPU usage under high ingestion rate.
+const maxRawRowsPerShard = (8 << 20) / int(unsafe.Sizeof(rawRow{}))
+
 // partition represents a partition.
 type partition struct {
 	activeInmemoryMerges atomic.Int64
@@ -319,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
@@ -383,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
@@ -394,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
@@ -405,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
@@ -465,29 +476,150 @@ func (pt *partition) AddRows(rows []rawRow) {
 		}
 	}
 
-	pt.rawRows.addRows(pt.flushRowssToInmemoryParts, rows)
+	pt.rawRows.addRows(pt, rows)
 }
 
 var isDebug = false
 
-func (pt *partition) flushRowssToInmemoryParts(rowss [][]rawRow) {
-	var nonEmptyRowss [][]rawRow
-	for _, rows := range rowss {
-		if len(rows) > 0 {
-			nonEmptyRowss = append(nonEmptyRowss, rows)
-		}
+type rawRowsShards struct {
+	flushDeadlineMs atomic.Int64
+
+	shardIdx atomic.Uint32
+
+	// Shards reduce lock contention when adding rows on multi-CPU systems.
+	shards []rawRowsShard
+
+	rowssToFlushLock sync.Mutex
+	rowssToFlush     [][]rawRow
+}
+
+func (rrss *rawRowsShards) init() {
+	rrss.shards = make([]rawRowsShard, rawRowsShardsPerPartition)
+}
+
+func (rrss *rawRowsShards) addRows(pt *partition, rows []rawRow) {
+	shards := rrss.shards
+	shardsLen := uint32(len(shards))
+	for len(rows) > 0 {
+		n := rrss.shardIdx.Add(1)
+		idx := n % shardsLen
+		tailRows, rowsToFlush := shards[idx].addRows(rows)
+		rrss.addRowsToFlush(pt, rowsToFlush)
+		rows = tailRows
 	}
-	if len(nonEmptyRowss) == 0 {
+}
+
+func (rrss *rawRowsShards) addRowsToFlush(pt *partition, rowsToFlush []rawRow) {
+	if len(rowsToFlush) == 0 {
+		return
+	}
+
+	var rowssToMerge [][]rawRow
+
+	rrss.rowssToFlushLock.Lock()
+	if len(rrss.rowssToFlush) == 0 {
+		rrss.updateFlushDeadline()
+	}
+	rrss.rowssToFlush = append(rrss.rowssToFlush, rowsToFlush)
+	if len(rrss.rowssToFlush) >= defaultPartsToMerge {
+		rowssToMerge = rrss.rowssToFlush
+		rrss.rowssToFlush = nil
+	}
+	rrss.rowssToFlushLock.Unlock()
+
+	pt.flushRowssToInmemoryParts(rowssToMerge)
+}
+
+func (rrss *rawRowsShards) Len() int {
+	n := 0
+	for i := range rrss.shards[:] {
+		n += rrss.shards[i].Len()
+	}
+
+	rrss.rowssToFlushLock.Lock()
+	for _, rows := range rrss.rowssToFlush {
+		n += len(rows)
+	}
+	rrss.rowssToFlushLock.Unlock()
+
+	return n
+}
+
+func (rrss *rawRowsShards) updateFlushDeadline() {
+	rrss.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
+}
+
+type rawRowsShardNopad struct {
+	flushDeadlineMs atomic.Int64
+
+	mu   sync.Mutex
+	rows []rawRow
+}
+
+type rawRowsShard struct {
+	rawRowsShardNopad
+
+	// The padding prevents false sharing
+	_ [atomicutil.CacheLineSize - unsafe.Sizeof(rawRowsShardNopad{})%atomicutil.CacheLineSize]byte
+}
+
+func (rrs *rawRowsShard) Len() int {
+	rrs.mu.Lock()
+	n := len(rrs.rows)
+	rrs.mu.Unlock()
+	return n
+}
+
+func (rrs *rawRowsShard) addRows(rows []rawRow) ([]rawRow, []rawRow) {
+	var rowsToFlush []rawRow
+
+	rrs.mu.Lock()
+	if cap(rrs.rows) == 0 {
+		rrs.rows = newRawRows()
+	}
+	if len(rrs.rows) == 0 {
+		rrs.updateFlushDeadline()
+	}
+	n := copy(rrs.rows[len(rrs.rows):cap(rrs.rows)], rows)
+	rrs.rows = rrs.rows[:len(rrs.rows)+n]
+	rows = rows[n:]
+	if len(rows) > 0 {
+		rowsToFlush = rrs.rows
+		rrs.rows = newRawRows()
+		rrs.updateFlushDeadline()
+		n = copy(rrs.rows[:cap(rrs.rows)], rows)
+		rrs.rows = rrs.rows[:n]
+		rows = rows[n:]
+	}
+	rrs.mu.Unlock()
+
+	return rows, rowsToFlush
+}
+
+func newRawRows() []rawRow {
+	return make([]rawRow, 0, maxRawRowsPerShard)
+}
+
+func (pt *partition) flushRowssToInmemoryParts(rowss [][]rawRow) {
+	if len(rowss) == 0 {
 		return
 	}
 
 	// Convert rowss into in-memory parts.
-	pws := make([]*partWrapper, len(nonEmptyRowss))
+	var pwsLock sync.Mutex
+	pws := make([]*partWrapper, 0, len(rowss))
 	wg := getWaitGroup()
-	for i, rows := range nonEmptyRowss {
+	for _, rows := range rowss {
 		inmemoryPartsConcurrencyCh <- struct{}{}
+
 		wg.Go(func() {
-			pws[i] = pt.mustCreateInmemoryPart(rows)
+			pw := pt.createInmemoryPart(rows)
+			if pw != nil {
+				pwsLock.Lock()
+				pws = append(pws, pw)
+				pwsLock.Unlock()
+			}
+
 			<-inmemoryPartsConcurrencyCh
 		})
 	}
@@ -742,14 +874,9 @@ func (pt *partition) mustMergeInmemoryPartsFinal(pws []*partWrapper) *partWrappe
 	return newPartWrapperFromInmemoryPart(mpDst, flushToDiskDeadline)
 }
 
-// mustCreateInmemoryPart creates a new in-memory part from rawRows.
-//
-// The number of rawRows cannot be zero. Otherwise the method will panic.
-//
-// The returned value is always a non-nil partWrapper.
-func (pt *partition) mustCreateInmemoryPart(rows []rawRow) *partWrapper {
+func (pt *partition) createInmemoryPart(rows []rawRow) *partWrapper {
 	if len(rows) == 0 {
-		logger.Panicf("BUG: a part cannot be created from 0 rawRows")
+		return nil
 	}
 	mp := getInmemoryPart()
 	mp.InitFromRows(rows)
@@ -1011,7 +1138,7 @@ func (pt *partition) pendingRowsFlusher() {
 }
 
 func (pt *partition) flushPendingRows(isFinal bool) {
-	pt.rawRows.flush(pt.flushRowssToInmemoryParts, isFinal)
+	pt.rawRows.flush(pt, isFinal)
 }
 
 func (pt *partition) flushInmemoryRowsToFiles() {
@@ -1037,6 +1164,66 @@ func (pt *partition) flushInmemoryPartsToFiles(isFinal bool) {
 	}
 }
 
+func (rrss *rawRowsShards) flush(pt *partition, isFinal bool) {
+	var dst [][]rawRow
+
+	currentTimeMs := time.Now().UnixMilli()
+	flushDeadlineMs := rrss.flushDeadlineMs.Load()
+	if isFinal || currentTimeMs >= flushDeadlineMs {
+		rrss.rowssToFlushLock.Lock()
+		dst = rrss.rowssToFlush
+		rrss.rowssToFlush = nil
+		rrss.rowssToFlushLock.Unlock()
+	}
+
+	for i := range rrss.shards {
+		dst = rrss.shards[i].appendRawRowsToFlush(dst, currentTimeMs, isFinal)
+	}
+
+	pt.flushRowssToInmemoryParts(dst)
+}
+
+func (rrs *rawRowsShard) appendRawRowsToFlush(dst [][]rawRow, currentTimeMs int64, isFinal bool) [][]rawRow {
+	flushDeadlineMs := rrs.flushDeadlineMs.Load()
+	if !isFinal && currentTimeMs < flushDeadlineMs {
+		// Fast path - nothing to flush
+		return dst
+	}
+
+	// Slow path - move rrs.rows to dst.
+	rrs.mu.Lock()
+	dst = appendRawRowss(dst, rrs.rows)
+	rrs.rows = rrs.rows[:0]
+	rrs.mu.Unlock()
+
+	return dst
+}
+
+func (rrs *rawRowsShard) updateFlushDeadline() {
+	rrs.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
+}
+
+func appendRawRowss(dst [][]rawRow, src []rawRow) [][]rawRow {
+	if len(src) == 0 {
+		return dst
+	}
+	if len(dst) == 0 {
+		dst = append(dst, newRawRows())
+	}
+	prows := &dst[len(dst)-1]
+	n := copy((*prows)[len(*prows):cap(*prows)], src)
+	*prows = (*prows)[:len(*prows)+n]
+	src = src[n:]
+	for len(src) > 0 {
+		rows := newRawRows()
+		n := copy(rows[:cap(rows)], src)
+		rows = rows[:len(rows)+n]
+		src = src[n:]
+		dst = append(dst, rows)
+	}
+	return dst
+}
+
 func (pt *partition) mergePartsToFiles(pws []*partWrapper, stopCh <-chan struct{}, concurrencyCh chan struct{}, useSparseCache bool) error {
 	pwsLen := len(pws)
 

lib/storage/raw_row.go

@@ -3,29 +3,11 @@ package storage
 import (
 	"sort"
 	"sync"
-	"sync/atomic"
-	"time"
-	"unsafe"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/atomicutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 )
 
-// The number of shards for rawRow entries.
-//
-// Higher number of shards reduces CPU contention and increases the max bandwidth on multi-core systems.
-var numRawRowsShards = cgroup.AvailableCPUs()
-
-// The interval for flushing buffered rows into parts, so they become visible to search.
-const pendingRowsFlushInterval = 2 * time.Second
-
-// The maximum number of rawRow items in rawRowsShard.
-//
-// Limit the maximum shard size to 8Mb, since this gives the lowest CPU usage under high ingestion rate.
-const maxRawRowsPerShard = (8 << 20) / int(unsafe.Sizeof(rawRow{}))
-
 // rawRow represents raw timeseries row.
 type rawRow struct {
 	// TSID is time series id.
@@ -167,182 +149,3 @@ func putRawRowsMarshaler(rrm *rawRowsMarshaler) {
 }
 
 var rrmPool sync.Pool
-
-type rawRowsShards struct {
-	flushDeadlineMs atomic.Int64
-
-	shardIdx atomic.Uint32
-
-	// Shards reduce lock contention when adding rows on multi-CPU systems.
-	shards []rawRowsShard
-
-	rowssToFlushLock sync.Mutex
-	rowssToFlush     [][]rawRow
-}
-
-func (rrss *rawRowsShards) init() {
-	rrss.shards = make([]rawRowsShard, numRawRowsShards)
-}
-
-func (rrss *rawRowsShards) Len() int {
-	n := 0
-	for i := range rrss.shards[:] {
-		n += rrss.shards[i].Len()
-	}
-
-	rrss.rowssToFlushLock.Lock()
-	for _, rows := range rrss.rowssToFlush {
-		n += len(rows)
-	}
-	rrss.rowssToFlushLock.Unlock()
-
-	return n
-}
-
-func (rrss *rawRowsShards) addRows(flush func([][]rawRow), rows []rawRow) {
-	shards := rrss.shards
-	shardsLen := uint32(len(shards))
-	for len(rows) > 0 {
-		n := rrss.shardIdx.Add(1)
-		idx := n % shardsLen
-		tailRows, rowsToFlush := shards[idx].addRows(rows)
-		rrss.addRowsToFlush(flush, rowsToFlush)
-		rows = tailRows
-	}
-}
-
-func (rrss *rawRowsShards) addRowsToFlush(flush func([][]rawRow), rowsToFlush []rawRow) {
-	if len(rowsToFlush) == 0 {
-		return
-	}
-
-	var rowssToMerge [][]rawRow
-
-	rrss.rowssToFlushLock.Lock()
-	if len(rrss.rowssToFlush) == 0 {
-		rrss.updateFlushDeadline()
-	}
-	rrss.rowssToFlush = append(rrss.rowssToFlush, rowsToFlush)
-	if len(rrss.rowssToFlush) >= defaultPartsToMerge {
-		rowssToMerge = rrss.rowssToFlush
-		rrss.rowssToFlush = nil
-	}
-	rrss.rowssToFlushLock.Unlock()
-
-	flush(rowssToMerge)
-}
-
-func (rrss *rawRowsShards) updateFlushDeadline() {
-	rrss.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
-}
-
-func (rrss *rawRowsShards) flush(flush func(rrs [][]rawRow), isFinal bool) {
-	var dst [][]rawRow
-
-	currentTimeMs := time.Now().UnixMilli()
-	flushDeadlineMs := rrss.flushDeadlineMs.Load()
-	if isFinal || currentTimeMs >= flushDeadlineMs {
-		rrss.rowssToFlushLock.Lock()
-		dst = rrss.rowssToFlush
-		rrss.rowssToFlush = nil
-		rrss.rowssToFlushLock.Unlock()
-	}
-
-	for i := range rrss.shards {
-		dst = rrss.shards[i].appendRawRowsToFlush(dst, currentTimeMs, isFinal)
-	}
-
-	flush(dst)
-}
-
-type rawRowsShardNopad struct {
-	flushDeadlineMs atomic.Int64
-
-	mu   sync.Mutex
-	rows []rawRow
-}
-
-type rawRowsShard struct {
-	rawRowsShardNopad
-
-	// The padding prevents false sharing
-	_ [atomicutil.CacheLineSize - unsafe.Sizeof(rawRowsShardNopad{})%atomicutil.CacheLineSize]byte
-}
-
-func (rrs *rawRowsShard) Len() int {
-	rrs.mu.Lock()
-	n := len(rrs.rows)
-	rrs.mu.Unlock()
-	return n
-}
-
-func (rrs *rawRowsShard) addRows(rows []rawRow) ([]rawRow, []rawRow) {
-	var rowsToFlush []rawRow
-
-	rrs.mu.Lock()
-	if cap(rrs.rows) == 0 {
-		rrs.rows = newRawRows()
-	}
-	if len(rrs.rows) == 0 {
-		rrs.updateFlushDeadline()
-	}
-	n := copy(rrs.rows[len(rrs.rows):cap(rrs.rows)], rows)
-	rrs.rows = rrs.rows[:len(rrs.rows)+n]
-	rows = rows[n:]
-	if len(rows) > 0 {
-		rowsToFlush = rrs.rows
-		rrs.rows = newRawRows()
-		rrs.updateFlushDeadline()
-		n = copy(rrs.rows[:cap(rrs.rows)], rows)
-		rrs.rows = rrs.rows[:n]
-		rows = rows[n:]
-	}
-	rrs.mu.Unlock()
-
-	return rows, rowsToFlush
-}
-
-func newRawRows() []rawRow {
-	return make([]rawRow, 0, maxRawRowsPerShard)
-}
-
-func (rrs *rawRowsShard) updateFlushDeadline() {
-	rrs.flushDeadlineMs.Store(time.Now().Add(pendingRowsFlushInterval).UnixMilli())
-}
-
-func (rrs *rawRowsShard) appendRawRowsToFlush(dst [][]rawRow, currentTimeMs int64, isFinal bool) [][]rawRow {
-	flushDeadlineMs := rrs.flushDeadlineMs.Load()
-	if !isFinal && currentTimeMs < flushDeadlineMs {
-		// Fast path - nothing to flush
-		return dst
-	}
-
-	// Slow path - move rrs.rows to dst.
-	rrs.mu.Lock()
-	dst = appendRawRowss(dst, rrs.rows)
-	rrs.rows = rrs.rows[:0]
-	rrs.mu.Unlock()
-
-	return dst
-}
-
-func appendRawRowss(dst [][]rawRow, src []rawRow) [][]rawRow {
-	if len(src) == 0 {
-		return dst
-	}
-	if len(dst) == 0 {
-		dst = append(dst, newRawRows())
-	}
-	prows := &dst[len(dst)-1]
-	n := copy((*prows)[len(*prows):cap(*prows)], src)
-	*prows = (*prows)[:len(*prows)+n]
-	src = src[n:]
-	for len(src) > 0 {
-		rows := newRawRows()
-		n := copy(rows[:cap(rows)], src)
-		rows = rows[:len(rows)+n]
-		src = src[n:]
-		dst = append(dst, rows)
-	}
-	return dst
-}

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