Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1381 (2021-06-14 13:06:21)

promql: fix `increase_pure` calculation for cases with stale series
Due to staleness handling, increase_pure were using incorrect previous value
during calculation in cases where series disappears for period longer
than staleness period and then returns back. The fix suppose to account
for a real datapoint value before staleness takes place. The fix should
remove unexpected spikes while using `increase_pure` for staled series.

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
> Merging [#1381](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (7d4ada8) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/729c4eeb9cf2e54c64fcf6d3480011eec6d22a2c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (729c4ee) will **decrease** coverage by `0.01%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)

```diff
@@            Coverage Diff             @@
##           master    #1381      +/-   ##
==========================================
- Coverage   60.73%   60.72%   -0.02%     
==========================================
  Files         224      224              
  Lines       26472    26472              
==========================================
- Hits        16078    16075       -3     
- Misses       8903     8904       +1     
- Partials     1491     1493       +2     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [app/vmselect/promql/rollup.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9yb2xsdXAuZ28=) | `85.04% <100.00%> (-0.24%)` | :arrow_down: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `57.00% <0.00%> (-3.74%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `69.85% <0.00%> (-0.30%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `86.43% <0.00%> (ø)` | |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `67.13% <0.00%> (+0.15%)` | :arrow_up: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `65.63% <0.00%> (+0.23%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Last update [05bc966...7d4ada8](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1381?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/README.md b/README.md
index 94e86d5bb..c3ed564a8 100644
--- a/README.md
+++ b/README.md
@@ -94,7 +94,7 @@ Alphabetically sorted links to case studies:
   * [Arbitrary CSV data](#how-to-import-csv-data).
 * Supports metrics' relabeling. See [these docs](#relabeling) for details.
 * Can deal with high cardinality and high churn rate issues using [series limiter](#cardinality-limiter).
-* Ideally works with big amounts of time series data from Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
+* Ideally works with big amounts of time series data from APM, Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
 * Has open source [cluster version](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/cluster).
 * See also technical [Articles about VictoriaMetrics](https://docs.victoriametrics.com/Articles.html).
 
diff --git a/app/vmselect/promql/rollup.go b/app/vmselect/promql/rollup.go
index 4bdb73fca..4e9eb763f 100644
--- a/app/vmselect/promql/rollup.go
+++ b/app/vmselect/promql/rollup.go
@@ -1332,7 +1332,8 @@ func rollupIncreasePure(rfa *rollupFuncArg) float64 {
 	// There is no need in handling NaNs here, since they must be cleaned up
 	// before calling rollup funcs.
 	values := rfa.values
-	prevValue := rfa.prevValue
+	// restore to the real value because of potential staleness reset
+	prevValue := rfa.realPrevValue
 	if math.IsNaN(prevValue) {
 		if len(values) == 0 {
 			return nan
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 212840203..2b0d1848a 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -9,6 +9,8 @@ sort: 15
 * FEATURE: vmagent: add service discovery for DigitalOcean (aka [digitalocean_sd_config](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#digitalocean_sd_config)). See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1367).
 * FEATURE: vmagent: show the number of samples the target returned during the last scrape on `/targets` and `/api/v1/targets` pages. This should simplify debugging targets, which may return too big or too low number of samples. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1377).
 
+* BUGFIX: prevent from adding new samples to deleted time series after the rotation of the inverted index (the rotation is performed once per `-retentionPeriod`). See [this comment](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1347#issuecomment-861232136) for details.
+
 
 ## [v1.61.1](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.61.1)
 
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 012c133de..234a49807 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -98,7 +98,7 @@ Alphabetically sorted links to case studies:
   * [Arbitrary CSV data](#how-to-import-csv-data).
 * Supports metrics' relabeling. See [these docs](#relabeling) for details.
 * Can deal with high cardinality and high churn rate issues using [series limiter](#cardinality-limiter).
-* Ideally works with big amounts of time series data from Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
+* Ideally works with big amounts of time series data from APM, Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
 * Has open source [cluster version](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/cluster).
 * See also technical [Articles about VictoriaMetrics](https://docs.victoriametrics.com/Articles.html).
 
diff --git a/lib/storage/index_db.go b/lib/storage/index_db.go
index f2244438a..5123f98c3 100644
--- a/lib/storage/index_db.go
+++ b/lib/storage/index_db.go
@@ -103,15 +103,6 @@ type indexDB struct {
 	loopsPerDateTagFilterCache *workingsetcache.Cache
 
 	indexSearchPool sync.Pool
-
-	// An inmemory set of deleted metricIDs.
-	//
-	// The set holds deleted metricIDs for the current db and for the extDB.
-	//
-	// It is safe to keep the set in memory even for big number of deleted
-	// metricIDs, since it usually requires 1 bit per deleted metricID.
-	deletedMetricIDs           atomic.Value
-	deletedMetricIDsUpdateLock sync.Mutex
 }
 
 // openIndexDB opens index db from the given path with the given caches.
@@ -140,14 +131,6 @@ func openIndexDB(path string, s *Storage) (*indexDB, error) {
 		uselessTagFiltersCache:     workingsetcache.New(mem/128, time.Hour),
 		loopsPerDateTagFilterCache: workingsetcache.New(mem/128, time.Hour),
 	}
-
-	is := db.getIndexSearch(noDeadline)
-	dmis, err := is.loadDeletedMetricIDs()
-	db.putIndexSearch(is)
-	if err != nil {
-		return nil, fmt.Errorf("cannot load deleted metricIDs: %w", err)
-	}
-	db.setDeletedMetricIDs(dmis)
 	return db, nil
 }
 
@@ -214,7 +197,7 @@ func (db *indexDB) UpdateMetrics(m *IndexDBMetrics) {
 	m.UselessTagFiltersCacheRequests += cs.GetCalls
 	m.UselessTagFiltersCacheMisses += cs.Misses
 
-	m.DeletedMetricsCount += uint64(db.getDeletedMetricIDs().Len())
+	m.DeletedMetricsCount += uint64(db.s.getDeletedMetricIDs().Len())
 
 	m.IndexDBRefCount += atomic.LoadUint64(&db.refCount)
 	m.NewTimeseriesCreated += atomic.LoadUint64(&db.newTimeseriesCreated)
@@ -260,12 +243,6 @@ func (db *indexDB) doExtDB(f func(extDB *indexDB)) bool {
 //
 // It decrements refCount for the previous extDB.
 func (db *indexDB) SetExtDB(extDB *indexDB) {
-	// Add deleted metricIDs from extDB to db.
-	if extDB != nil {
-		dmisExt := extDB.getDeletedMetricIDs()
-		db.updateDeletedMetricIDs(dmisExt)
-	}
-
 	db.extDBLock.Lock()
 	prevExtDB := db.extDB
 	db.extDB = extDB
@@ -737,7 +714,7 @@ func (is *indexSearch) searchTagKeysOnDate(tks map[string]struct{}, date uint64,
 	kb := &is.kb
 	mp := &is.mp
 	mp.Reset()
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
 	kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixDateTagToMetricIDs)
 	kb.B = encoding.MarshalUint64(kb.B, date)
@@ -817,7 +794,7 @@ func (is *indexSearch) searchTagKeys(tks map[string]struct{}, maxTagKeys int) er
 	kb := &is.kb
 	mp := &is.mp
 	mp.Reset()
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
 	kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
 	prefix := kb.B
@@ -935,7 +912,7 @@ func (is *indexSearch) searchTagValuesOnDate(tvs map[string]struct{}, tagKey []b
 	kb := &is.kb
 	mp := &is.mp
 	mp.Reset()
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
 	kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixDateTagToMetricIDs)
 	kb.B = encoding.MarshalUint64(kb.B, date)
@@ -1021,7 +998,7 @@ func (is *indexSearch) searchTagValues(tvs map[string]struct{}, tagKey []byte, m
 	kb := &is.kb
 	mp := &is.mp
 	mp.Reset()
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
 	kb.B = is.marshalCommonPrefix(kb.B[:0], nsPrefixTagToMetricIDs)
 	kb.B = marshalTagValue(kb.B, tagKey)
@@ -1175,7 +1152,7 @@ func (is *indexSearch) searchTagValueSuffixesForPrefix(tvss map[string]struct{},
 	ts := &is.ts
 	mp := &is.mp
 	mp.Reset()
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
 	ts.Seek(prefix)
 	for len(tvss) < maxTagValueSuffixes && ts.NextItem() {
@@ -1616,7 +1593,7 @@ func (db *indexDB) deleteMetricIDs(metricIDs []uint64) error {
 	// atomically add deleted metricIDs to an inmemory map.
 	dmis := &uint64set.Set{}
 	dmis.AddMulti(metricIDs)
-	db.updateDeletedMetricIDs(dmis)
+	db.s.updateDeletedMetricIDs(dmis)
 
 	// Reset TagFilters -> TSIDS cache, since it may contain deleted TSIDs.
 	invalidateTagCache()
@@ -1643,21 +1620,14 @@ func (db *indexDB) deleteMetricIDs(metricIDs []uint64) error {
 	return err
 }
 
-func (db *indexDB) getDeletedMetricIDs() *uint64set.Set {
-	return db.deletedMetricIDs.Load().(*uint64set.Set)
-}
-
-func (db *indexDB) setDeletedMetricIDs(dmis *uint64set.Set) {
-	db.deletedMetricIDs.Store(dmis)
-}
-
-func (db *indexDB) updateDeletedMetricIDs(metricIDs *uint64set.Set) {
-	db.deletedMetricIDsUpdateLock.Lock()
-	dmisOld := db.getDeletedMetricIDs()
-	dmisNew := dmisOld.Clone()
-	dmisNew.Union(metricIDs)
-	db.setDeletedMetricIDs(dmisNew)
-	db.deletedMetricIDsUpdateLock.Unlock()
+func (db *indexDB) loadDeletedMetricIDs() (*uint64set.Set, error) {
+	is := db.getIndexSearch(noDeadline)
+	dmis, err := is.loadDeletedMetricIDs()
+	db.putIndexSearch(is)
+	if err != nil {
+		return nil, err
+	}
+	return dmis, nil
 }
 
 func (is *indexSearch) loadDeletedMetricIDs() (*uint64set.Set, error) {
@@ -1751,7 +1721,7 @@ func (db *indexDB) searchTSIDs(tfss []*TagFilters, tr TimeRange, maxMetrics int,
 var tagFiltersKeyBufPool bytesutil.ByteBufferPool
 
 func (is *indexSearch) getTSIDByMetricName(dst *TSID, metricName []byte) error {
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	ts := &is.ts
 	kb := &is.kb
 	kb.B = append(kb.B[:0], nsPrefixMetricNameToTSID)
@@ -2315,7 +2285,7 @@ func (is *indexSearch) searchMetricIDs(tfss []*TagFilters, tr TimeRange, maxMetr
 	sortedMetricIDs := metricIDs.AppendTo(nil)
 
 	// Filter out deleted metricIDs.
-	dmis := is.db.getDeletedMetricIDs()
+	dmis := is.db.s.getDeletedMetricIDs()
 	if dmis.Len() > 0 {
 		metricIDsFiltered := sortedMetricIDs[:0]
 		for _, metricID := range sortedMetricIDs {
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index 239dce203..bcf129207 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -120,6 +120,13 @@ type Storage struct {
 
 	// The minimum timestamp when composite index search can be used.
 	minTimestampForCompositeIndex int64
+
+	// An inmemory set of deleted metricIDs.
+	//
+	// It is safe to keep the set in memory even for big number of deleted
+	// metricIDs, since it usually requires 1 bit per deleted metricID.
+	deletedMetricIDs           atomic.Value
+	deletedMetricIDsUpdateLock sync.Mutex
 }
 
 // OpenStorage opens storage on the given path with the given retentionMsecs.
@@ -208,6 +215,18 @@ func OpenStorage(path string, retentionMsecs int64, maxHourlySeries, maxDailySer
 	idbCurr.SetExtDB(idbPrev)
 	s.idbCurr.Store(idbCurr)
 
+	// Load deleted metricIDs from idbCurr and idbPrev
+	dmisCurr, err := idbCurr.loadDeletedMetricIDs()
+	if err != nil {
+		return nil, fmt.Errorf("cannot load deleted metricIDs for the current indexDB: %w", err)
+	}
+	dmisPrev, err := idbPrev.loadDeletedMetricIDs()
+	if err != nil {
+		return nil, fmt.Errorf("cannot load deleted metricIDs for the previous indexDB: %w", err)
+	}
+	s.setDeletedMetricIDs(dmisCurr)
+	s.updateDeletedMetricIDs(dmisPrev)
+
 	// Load data
 	tablePath := path + "/data"
 	tb, err := openTable(tablePath, s.getDeletedMetricIDs, retentionMsecs)
@@ -224,16 +243,29 @@ func OpenStorage(path string, retentionMsecs int64, maxHourlySeries, maxDailySer
 	return s, nil
 }
 
+func (s *Storage) getDeletedMetricIDs() *uint64set.Set {
+	return s.deletedMetricIDs.Load().(*uint64set.Set)
+}
+
+func (s *Storage) setDeletedMetricIDs(dmis *uint64set.Set) {
+	s.deletedMetricIDs.Store(dmis)
+}
+
+func (s *Storage) updateDeletedMetricIDs(metricIDs *uint64set.Set) {
+	s.deletedMetricIDsUpdateLock.Lock()
+	dmisOld := s.getDeletedMetricIDs()
+	dmisNew := dmisOld.Clone()
+	dmisNew.Union(metricIDs)
+	s.setDeletedMetricIDs(dmisNew)
+	s.deletedMetricIDsUpdateLock.Unlock()
+}
+
 // DebugFlush flushes recently added storage data, so it becomes visible to search.
 func (s *Storage) DebugFlush() {
 	s.tb.flushRawRows()
 	s.idb().tb.DebugFlush()
 }
 
-func (s *Storage) getDeletedMetricIDs() *uint64set.Set {
-	return s.idb().getDeletedMetricIDs()
-}
-
 // CreateSnapshot creates snapshot for s and returns the snapshot name.
 func (s *Storage) CreateSnapshot() (string, error) {
 	logger.Infof("creating Storage snapshot for %q...", s.path)


Test patch:
diff --git a/lib/storage/index_db_test.go b/lib/storage/index_db_test.go
index 681dd8525..6cdeac28a 100644
--- a/lib/storage/index_db_test.go
+++ b/lib/storage/index_db_test.go
@@ -1711,13 +1711,15 @@ func toTFPointers(tfs []tagFilter) []*tagFilter {
 }
 
 func newTestStorage() *Storage {
-	return &Storage{
+	s := &Storage{
 		cachePath: "test-storage-cache",
 
 		metricIDCache:   workingsetcache.New(1234, time.Hour),
 		metricNameCache: workingsetcache.New(1234, time.Hour),
 		tsidCache:       workingsetcache.New(1234, time.Hour),
 	}
+	s.setDeletedMetricIDs(&uint64set.Set{})
+	return s
 }
 
 func stopTestStorage(s *Storage) {