Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7535 (2024-11-14 10:11:14)

lib/storage: properly check for minMissingTimestamps
Previously, check was performed without context of the index search prefix. It worked fine for Single node version, but for cluster version a different prefix was used for metricID search requests. And it may lead to incomplete results, if minimal missing timestamp was cached for the tenant with different ingestion patterns.

 Image the following case:
- metrics were ingested for tenants 0 and 1
- at some point in time metrics ingestion for tenant 1 stopped
- index records have the following timestamps layout: tenant 0:  1,2,3,4,5,6 tenant 1:  1,2,3,4
- after indexDB rotation, containsTimeRange lookups may produce incorrect results: time range request for tenant 1 - 5:6 caches 5 as min timestamp request for the same or smaller time range for tenant 0 now returns empty results.

 This commit replaces single atomic with map of search prefix keys. It should have slight performance overhead,
but work consistenetely for cluster version. It caches minMissingTimestamp by prefix search key and resolves issue.

 Since it will be only populated at runtime, it doesn't hold unused tenants for queries.

Related issue: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7417

### Describe Your Changes

Please provide a brief description of the changes you made. Be as specific as possible to help others understand the purpose and impact of your modifications.

### Checklist

The following checks are **mandatory**:

- [ ] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).


f41gh7:  This commit could be also backported to the master branch. It brings slight overhead compared to atomic version. But it keeps code the same and simplifies conflicts resolution. 

Patch:
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index 138c08afd..26a302ce2 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -23,10 +23,9 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl/): drop rows that do not belong to the current series during import. The dropped rows should belong to another series whose tags are a superset of the current series. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7301) and [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7330). Thanks to @dpedu for reporting and cooperating with the test.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/), `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): keep the order of resulting time series when `limit_offset` is applied. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7068).
 * BUGFIX: [graphite](https://docs.victoriametrics.com/#graphite-render-api-usage): properly handle xFilesFactor=0 for `transformRemoveEmptySeries` function. See [this PR](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7337) for details.
-* BUGFIX: [vmauth](https://docs.victoriametrics.com/vmauth/): properly check availability of all the backends before giving up when proxying requests. Previously, vmauth could return an error even if there were healthy backends available. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3061) for details.
-* BUGFIX: [vmauth](https://docs.victoriametrics.com/vmauth/): properly inherit [`drop_src_path_prefix_parts`](https://docs.victoriametrics.com/vmauth/#dropping-request-path-prefix), [`load_balancing_policy`](https://docs.victoriametrics.com/vmauth/#high-availability), [`retry_status_codes`](https://docs.victoriametrics.com/vmauth/#load-balancing) and [`discover_backend_ips`](https://docs.victoriametrics.com/vmauth/#discovering-backend-ips) options by `url_map` entries if `url_prefix` option isn't set at the [user config level](https://docs.victoriametrics.com/vmauth/#auth-config). These options were inherited only when the `url_prefix` option was set. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7519).
-* BUGFIX: [dashboards](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/dashboards): add `file` label filter to vmalert dashboard panels. Previously, metrics from groups with the same name but different rule files could be mixed in the results.
-BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): Properly handle [multitenant](https://docs.victoriametrics.com/cluster-victoriametrics/#multitenancy-via-labels) query request errors and correctly perform search for available tenants. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7549) for details.
+* BUGFIX: [vmauth](https://docs.victoriametrics.com/vmauth): properly check availability of all the backends before giving up when proxying requests. Previously, vmauth could return an error even if there were healthy backends available. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3061) for details.
+* BUGFIX: `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): Properly return query results for search requests after index rotation. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7417) for details.
+* BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): Properly handle [multitenant](https://docs.victoriametrics.com/cluster-victoriametrics/#multitenancy-via-labels) query request errors and correctly perform search for available tenants. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7549) for details.
 
 ## [v1.106.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.106.0)
 
diff --git a/lib/storage/index_db.go b/lib/storage/index_db.go
index 333725b8d..cbdc66a0d 100644
--- a/lib/storage/index_db.go
+++ b/lib/storage/index_db.go
@@ -91,13 +91,17 @@ type indexDB struct {
 	// The db must be automatically recovered after that.
 	missingMetricNamesForMetricID atomic.Uint64
 
-	// minMissingTimestamp is the minimum timestamp, which is missing in the given indexDB.
+	// minMissingTimestampByKey holds the minimum timestamps by index search key,
+	// which is missing in the given indexDB.
+	// Key must be formed with marshalCommonPrefix function.
 	//
 	// This field is used at containsTimeRange() function only for the previous indexDB,
 	// since this indexDB is readonly.
 	// This field cannot be used for the current indexDB, since it may receive data
 	// with bigger timestamps at any time.
-	minMissingTimestamp atomic.Int64
+	minMissingTimestampByKey map[string]int64
+	// protects minMissingTimestampByKey
+	minMissingTimestampByKeyLock sync.RWMutex
 
 	// generation identifies the index generation ID
 	// and is used for syncing items from different indexDBs
@@ -162,6 +166,7 @@ func mustOpenIndexDB(path string, s *Storage, isReadOnly *atomic.Bool) *indexDB
 		tb:         tb,
 		name:       name,
 
+		minMissingTimestampByKey:   make(map[string]int64),
 		tagFiltersToMetricIDsCache: workingsetcache.New(tagFiltersCacheSize),
 		s:                          s,
 		loopsPerDateTagFilterCache: workingsetcache.New(mem / 128),
@@ -2100,25 +2105,36 @@ func (is *indexSearch) containsTimeRange(tr TimeRange) bool {
 		// This means that it may contain data for the given tr with probability close to 100%.
 		return true
 	}
-
 	// The db corresponds to the previous indexDB, which is readonly.
 	// So it is safe caching the minimum timestamp, which isn't covered by the db.
-	minMissingTimestamp := db.minMissingTimestamp.Load()
-	if minMissingTimestamp != 0 && tr.MinTimestamp >= minMissingTimestamp {
+
+	// use common prefix as a key for minMissingTimestamp
+	// it's needed to properly track timestamps for cluster version
+	// which uses tenant labels for the index search
+	kb := &is.kb
+	kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixDateToMetricID)
+	key := kb.B
+
+	db.minMissingTimestampByKeyLock.RLock()
+	minMissingTimestamp, ok := db.minMissingTimestampByKey[string(key)]
+	db.minMissingTimestampByKeyLock.RUnlock()
+
+	if ok && tr.MinTimestamp >= minMissingTimestamp {
 		return false
 	}
-
-	if is.containsTimeRangeSlow(tr) {
+	if is.containsTimeRangeSlowForPrefixBuf(kb, tr) {
 		return true
 	}
 
-	db.minMissingTimestamp.CompareAndSwap(minMissingTimestamp, tr.MinTimestamp)
+	db.minMissingTimestampByKeyLock.Lock()
+	db.minMissingTimestampByKey[string(key)] = tr.MinTimestamp
+	db.minMissingTimestampByKeyLock.Unlock()
+
 	return false
 }
 
-func (is *indexSearch) containsTimeRangeSlow(tr TimeRange) bool {
+func (is *indexSearch) containsTimeRangeSlowForPrefixBuf(prefixBuf *bytesutil.ByteBuffer, tr TimeRange) bool {
 	ts := &is.ts
-	kb := &is.kb
 
 	// Verify whether the tr.MinTimestamp is included into `ts` or is smaller than the minimum date stored in `ts`.
 	// Do not check whether tr.MaxTimestamp is included into `ts` or is bigger than the max date stored in `ts` for performance reasons.
@@ -2127,13 +2143,12 @@ func (is *indexSearch) containsTimeRangeSlow(tr TimeRange) bool {
 	// The main practical case allows skipping searching in prev indexdb (`ts`) when `tr`
 	// is located above the max date stored there.
 	minDate := uint64(tr.MinTimestamp) / msecPerDay
-	kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixDateToMetricID)
-	prefix := kb.B
-	kb.B = encoding.MarshalUint64(kb.B, minDate)
-	ts.Seek(kb.B)
+	prefix := prefixBuf.B
+	prefixBuf.B = encoding.MarshalUint64(prefixBuf.B, minDate)
+	ts.Seek(prefixBuf.B)
 	if !ts.NextItem() {
 		if err := ts.Error(); err != nil {
-			logger.Panicf("FATAL: error when searching for minDate=%d, prefix %q: %w", minDate, kb.B, err)
+			logger.Panicf("FATAL: error when searching for minDate=%d, prefix %q: %w", minDate, prefixBuf.B, err)
 		}
 		return false
 	}


Test patch:
diff --git a/lib/storage/index_db_test.go b/lib/storage/index_db_test.go
index 9c6e5c10e..0c1530e64 100644
--- a/lib/storage/index_db_test.go
+++ b/lib/storage/index_db_test.go
@@ -2216,3 +2216,196 @@ func stopTestStorage(s *Storage) {
 	s.tsidCache.Stop()
 	fs.MustRemoveDirAtomic(s.cachePath)
 }
+
+func TestSearchContainsTimeRange(t *testing.T) {
+	path := t.Name()
+	os.RemoveAll(path)
+	s := MustOpenStorage(path, retentionMax, 0, 0)
+	db := s.idb()
+
+	is := db.getIndexSearch(0, 0, noDeadline)
+
+	// Create a bunch of per-day time series
+	const (
+		days                = 6
+		tenant2IngestionDay = 8
+		metricsPerDay       = 1000
+	)
+	rotationDay := time.Date(2019, time.October, 15, 5, 1, 0, 0, time.UTC)
+	rotationMillis := uint64(rotationDay.UnixMilli())
+	rotationDate := rotationMillis / msecPerDay
+	var metricNameBuf []byte
+	perDayMetricIDs := make(map[uint64]*uint64set.Set)
+	labelNames := []string{
+		"__name__", "constant", "day", "UniqueId", "some_unique_id",
+	}
+
+	sort.Strings(labelNames)
+
+	newMN := func(name string, day, metric int) MetricName {
+		var mn MetricName
+		mn.MetricGroup = []byte(name)
+		mn.AddTag(
+			"constant",
+			"const",
+		)
+		mn.AddTag(
+			"day",
+			fmt.Sprintf("%v", day),
+		)
+		mn.AddTag(
+			"UniqueId",
+			fmt.Sprintf("%v", metric),
+		)
+		mn.AddTag(
+			"some_unique_id",
+			fmt.Sprintf("%v", day),
+		)
+		mn.sortTags()
+		return mn
+	}
+
+	// ingest metrics for tenant 0:0
+	for day := 0; day < days; day++ {
+		date := rotationDate - uint64(day)
+
+		var metricIDs uint64set.Set
+		for metric := range metricsPerDay {
+			mn := newMN("testMetric", day, metric)
+			metricNameBuf = mn.Marshal(metricNameBuf[:0])
+			var genTSID generationTSID
+			if !is.getTSIDByMetricName(&genTSID, metricNameBuf, date) {
+				generateTSID(&genTSID.TSID, &mn)
+				createAllIndexesForMetricName(is, &mn, &genTSID.TSID, date)
+			}
+			metricIDs.Add(genTSID.TSID.MetricID)
+		}
+
+		perDayMetricIDs[date] = &metricIDs
+	}
+	db.putIndexSearch(is)
+
+	// ingest metrics for tenant 1:1
+	isTenant2 := db.getIndexSearch(1, 1, noDeadline)
+	{
+		var metricIDs uint64set.Set
+		// ingestion day must be outside of tenant 0:0 data ingestion
+		date := rotationDate - uint64(tenant2IngestionDay)
+
+		for metric := range metricsPerDay {
+			mn := newMN("testMetric2", tenant2IngestionDay, metric)
+			mn.AccountID = 1
+			mn.ProjectID = 1
+			metricNameBuf = mn.Marshal(metricNameBuf[:0])
+			var genTSID generationTSID
+			if !isTenant2.getTSIDByMetricName(&genTSID, metricNameBuf, date) {
+				generateTSID(&genTSID.TSID, &mn)
+				createAllIndexesForMetricName(isTenant2, &mn, &genTSID.TSID, date)
+			}
+			metricIDs.Add(genTSID.TSID.MetricID)
+		}
+		perDayMetricIDs[date] = &metricIDs
+	}
+	db.putIndexSearch(isTenant2)
+
+	// Flush index to disk, so it becomes visible for search
+	s.DebugFlush()
+
+	is2 := db.getIndexSearch(0, 0, noDeadline)
+
+	// Check that all the metrics are found for all the days.
+	for date := rotationDate - days + 1; date <= rotationDate; date++ {
+
+		metricIDs, err := is2.getMetricIDsForDate(date, metricsPerDay)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if !perDayMetricIDs[date].Equal(metricIDs) {
+			t.Fatalf("unexpected metricIDs found;\ngot\n%d\nwant\n%d", metricIDs.AppendTo(nil), perDayMetricIDs[date].AppendTo(nil))
+		}
+	}
+
+	db.putIndexSearch(is2)
+
+	// Check that all metrics for tenant 2 are found at ingestion day
+	is2Tenant2 := db.getIndexSearch(1, 1, noDeadline)
+	{
+		date := rotationDate - uint64(tenant2IngestionDay+1)
+		metricIDs, err := is2Tenant2.getMetricIDsForDate(date, metricsPerDay)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if !perDayMetricIDs[date].Equal(metricIDs) {
+			t.Fatalf("unexpected 1:1 tenant metricIDs found;\ngot\n%d\nwant\n%d", metricIDs.AppendTo(nil), perDayMetricIDs[date].AppendTo(nil))
+		}
+	}
+	db.putIndexSearch(is2Tenant2)
+
+	// rotate indexdb
+	s.mustRotateIndexDB(rotationDay)
+	db = s.idb()
+
+	isExtTenant2 := db.extDB.getIndexSearch(1, 1, noDeadline)
+
+	// search for range covers metrics for 1:1 at prev index
+	tr := TimeRange{
+		MinTimestamp: int64(rotationMillis - msecPerDay*(tenant2IngestionDay) + 1),
+		MaxTimestamp: int64(rotationMillis),
+	}
+	if !isExtTenant2.containsTimeRange(tr) {
+		t.Fatalf("expected to have given time range at prev IndexDB")
+	}
+	// search for range missing for 1:1 at prev index
+	tr = TimeRange{
+		MinTimestamp: int64(rotationMillis - msecPerDay*(tenant2IngestionDay-1)),
+		MaxTimestamp: int64(rotationMillis),
+	}
+	if isExtTenant2.containsTimeRange(tr) {
+		t.Fatalf("not expected to have given time range at prev IndexDB")
+	}
+	key := isExtTenant2.marshalCommonPrefix(nil, nsPrefixDateToMetricID)
+
+	db.extDB.minMissingTimestampByKeyLock.Lock()
+	minMissingTimetamp := db.extDB.minMissingTimestampByKey[string(key)]
+	db.extDB.minMissingTimestampByKeyLock.Unlock()
+
+	if minMissingTimetamp != tr.MinTimestamp {
+		t.Fatalf("unexpected minMissingTimestamp for 1:1 tenant got %d, want %d", minMissingTimetamp, tr.MinTimestamp)
+	}
+	db.extDB.putIndexSearch(isExtTenant2)
+
+	// perform search for 0:0 tenant
+	// results of previous search requests shouldn't affect it
+
+	isExt := db.extDB.getIndexSearch(0, 0, noDeadline)
+	// search for range that covers prev indexDB for dates before ingestion
+	tr = TimeRange{
+		MinTimestamp: int64(rotationMillis - msecPerDay*(days)),
+		MaxTimestamp: int64(rotationMillis),
+	}
+	if !isExt.containsTimeRange(tr) {
+		t.Fatalf("expected to have given time range at prev IndexDB")
+	}
+
+	// search for range not exist at prev indexDB
+	tr = TimeRange{
+		MinTimestamp: int64(rotationMillis + msecPerDay*(days+4)),
+		MaxTimestamp: int64(rotationMillis + msecPerDay*(days+2)),
+	}
+	if isExt.containsTimeRange(tr) {
+		t.Fatalf("not expected to have given time range at prev IndexDB")
+	}
+	key = isExt.marshalCommonPrefix(key[:0], nsPrefixDateToMetricID)
+
+	db.extDB.minMissingTimestampByKeyLock.Lock()
+	minMissingTimetamp = db.extDB.minMissingTimestampByKey[string(key)]
+	db.extDB.minMissingTimestampByKeyLock.Unlock()
+
+	if minMissingTimetamp != tr.MinTimestamp {
+		t.Fatalf("unexpected minMissingTimestamp for 0:0 tenant got %d, want %d", minMissingTimetamp, tr.MinTimestamp)
+	}
+
+	db.extDB.putIndexSearch(isExt)
+	s.MustClose()
+	fs.MustRemoveAll(path)
+}