Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7118 (2024-09-27 10:05:19)

stream aggregation: fix possible duplicated aggregation results
When ingesting samples with the same labels(duplicated samples or samples with the same labels after `by` or `without` options). They could register different entries for the same labelset in LabelsCompressor. 
For example, both index 99 and 100 can be assigned to label `foo=1` in two concurrent pushes. Then due to differing label indexes in encoded keys, the samples will appear as distinct in aggrState, resulting in duplicated results after decompressing the label indexes.
https://github.com/VictoriaMetrics/VictoriaMetrics/blob/fbde238cdcdf4e2c892d85a3e9e2be6e54e69cef/lib/streamaggr/streamaggr.go#L933

In this pull request, since we need to store `idxToLabel` first to ensure the idx can be searched after `lc.labelToIdxStore`,
the `lc.idxToLabel` still could contain a duplicated entries [100]="foo=1". But given the low likelihood of this issue and the size of idxToLabel, it should be fine.



Patch:
diff --git a/app/vmselect/prometheus/prometheus.go b/app/vmselect/prometheus/prometheus.go
index 413a58d4d..73afe7e21 100644
--- a/app/vmselect/prometheus/prometheus.go
+++ b/app/vmselect/prometheus/prometheus.go
@@ -52,7 +52,7 @@ var (
 	maxExportSeries     = flag.Int("search.maxExportSeries", 10e6, "The maximum number of time series, which can be returned from /api/v1/export* APIs. This option allows limiting memory usage")
 	maxTSDBStatusSeries = flag.Int("search.maxTSDBStatusSeries", 10e6, "The maximum number of time series, which can be processed during the call to /api/v1/status/tsdb. This option allows limiting memory usage")
 	maxSeriesLimit      = flag.Int("search.maxSeries", 30e3, "The maximum number of time series, which can be returned from /api/v1/series. This option allows limiting memory usage")
-	maxDeleteSeries     = flag.Int("search.maxDeleteSeries", 30e3, "The maximum number of time series, which can be deleted using /api/v1/admin/tsdb/delete_series. This option allows limiting memory usage")
+	maxDeleteSeries     = flag.Int("search.maxDeleteSeries", 1e6, "The maximum number of time series, which can be deleted using /api/v1/admin/tsdb/delete_series. This option allows limiting memory usage")
 	maxLabelsAPISeries  = flag.Int("search.maxLabelsAPISeries", 1e6, "The maximum number of time series, which could be scanned when searching for the matching time series "+
 		"at /api/v1/labels and /api/v1/label/.../values. This option allows limiting memory usage and CPU usage. See also -search.maxLabelsAPIDuration, "+
 		"-search.maxTagKeys, -search.maxTagValues and -search.ignoreExtraFiltersAtLabelsAPI")
diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index b6516630c..f01ed1c59 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -15,6 +15,8 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
+* BUGFIX: do not return field values with zero matching logs from [`field_values`](https://docs.victoriametrics.com/victorialogs/logsql/#field_values-pipe), [`top`](https://docs.victoriametrics.com/victorialogs/logsql/#top-pipe) and [`uniq`](https://docs.victoriametrics.com/victorialogs/logsql/#uniq-pipe) pipes. See [this issue](https://github.com/VictoriaMetrics/victorialogs-datasource/issues/72#issuecomment-2352078483).
+
 ## [v0.32.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.32.0-victorialogs)
 
 Released at 2024-09-29
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index 196a5ae1d..3e332c86f 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -18,6 +18,8 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 ## tip
 
+**Update note 1: `*.passwordFile` and similar flags are trimming trailing whitespaces at the end of content. If authorization check performed with `*.passwordFile` content, make sure to update authorization settings to not include trailing whitespaces before the upgrade. In case of [operator](https://docs.victoriametrics.com/operator/) managed installations, make sure to update operator version to [v0.48.*](https://docs.victoriametrics.com/operator/changelog/#v0480---25-sep-2024). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6986) for the details. This change reverts behavior introduced at [v1.102.0-rc2](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.102.0-rc2) release**
+
 * SECURITY: upgrade Go builder from Go1.23.0 to Go1.23.1. See the list of issues addressed in [Go1.23.1](https://github.com/golang/go/issues?q=milestone%3AGo1.23.1+label%3ACherryPickApproved).
 * SECURITY: upgrade base docker image (Alpine) from 3.20.2 to 3.20.3. See [alpine 3.20.3 release notes](https://alpinelinux.org/posts/Alpine-3.17.10-3.18.9-3.19.4-3.20.3-released.html).
 
@@ -38,6 +40,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent/): properly consume messages [from kafka](https://docs.victoriametrics.com/vmagent/#kafka-integration). Previously vmagent could skip some messages during start-up.
 * BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): perform deduplication for all received data when specifying `-streamAggr.dedupInterval` or `-remoteWrite.streamAggr.dedupInterval` command-line flags are set. Previously, if the `-remoteWrite.streamAggr.config` or `-streamAggr.config` is set, only series that matched aggregation config were deduplicated. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6711#issuecomment-2288361213) for details.
 * BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): fix  `-remoteWrite.streamAggr.dropInputLabels` labels parsing. Now, this flag allows specifying a list of labels to drop (by using '^^' separator, i.e. `dropInputLabels='replica^^az,replica'`) per each corresponding `remoteWrite.url`. Before, `-remoteWrite.streamAggr.dropInputLabels` labels were incorrectly applied to all configured `remoteWrite.url`s. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6780) for the details.
+* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): fix duplicated aggregation results if there are multiple concurrent writing samples with the same label. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7118).
 * BUGFIX: [vmagent dashboard](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/dashboards/vmagent.json): fix legend captions for stream aggregation related panels. Before they were displaying wrong label names.
 * BUGFIX: [vmgateway](https://docs.victoriametrics.com/vmgateway/): add missing `datadog`, `newrelic`, `opentelemetry` and `pushgateway` routes to the `JWT` authorization routes. Allows prefixed (`promtheus/graphite`) routes for query requests.
 * BUGFIX: [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): properly cache empty list of matching time series for the given [labels filter](https://docs.victoriametrics.com/keyconcepts/#filtering). This type of caching was broken since [v1.97.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.97.0), which could result in the increased CPU usage when performing queries, which match zero time series. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7009).
diff --git a/lib/logstorage/pipe_top.go b/lib/logstorage/pipe_top.go
index 034038b0a..fb196cacb 100644
--- a/lib/logstorage/pipe_top.go
+++ b/lib/logstorage/pipe_top.go
@@ -207,6 +207,10 @@ func (shard *pipeTopProcessorShard) writeBlock(br *blockResult) {
 }
 
 func (shard *pipeTopProcessorShard) updateState(v string, hits uint64) {
+	if hits == 0 {
+		return
+	}
+
 	m := shard.getM()
 	pHits, ok := m[v]
 	if !ok {
diff --git a/lib/logstorage/pipe_uniq.go b/lib/logstorage/pipe_uniq.go
index c33cfc7f0..4803ac45a 100644
--- a/lib/logstorage/pipe_uniq.go
+++ b/lib/logstorage/pipe_uniq.go
@@ -166,24 +166,18 @@ func (shard *pipeUniqProcessorShard) writeBlock(br *blockResult) bool {
 			return true
 		}
 		if c.valueType == valueTypeDict {
-			if needHits {
-				a := encoding.GetUint64s(len(c.dictValues))
-				hits := a.A
-				clear(hits)
-				valuesEncoded := c.getValuesEncoded(br)
-				for _, v := range valuesEncoded {
-					idx := unmarshalUint8(v)
-					hits[idx]++
-				}
-				for i, v := range c.dictValues {
-					shard.updateState(v, hits[i])
-				}
-				encoding.PutUint64s(a)
-			} else {
-				for _, v := range c.dictValues {
-					shard.updateState(v, 0)
-				}
+			a := encoding.GetUint64s(len(c.dictValues))
+			hits := a.A
+			clear(hits)
+			valuesEncoded := c.getValuesEncoded(br)
+			for _, v := range valuesEncoded {
+				idx := unmarshalUint8(v)
+				hits[idx]++
 			}
+			for i, v := range c.dictValues {
+				shard.updateState(v, hits[i])
+			}
+			encoding.PutUint64s(a)
 			return true
 		}
 
@@ -230,6 +224,10 @@ func (shard *pipeUniqProcessorShard) writeBlock(br *blockResult) bool {
 }
 
 func (shard *pipeUniqProcessorShard) updateState(v string, hits uint64) {
+	if hits == 0 {
+		return
+	}
+
 	m := shard.getM()
 	pHits, ok := m[v]
 	if !ok {
diff --git a/lib/promutils/labelscompressor.go b/lib/promutils/labelscompressor.go
index 391b8f3c5..24c9d468d 100644
--- a/lib/promutils/labelscompressor.go
+++ b/lib/promutils/labelscompressor.go
@@ -59,8 +59,19 @@ func (lc *LabelsCompressor) compress(dst []uint64, labels []prompbmarshal.Label)
 			idx := lc.nextIdx.Add(1)
 			v = idx
 			labelCopy := cloneLabel(label)
+
+			// Must store idxToLabel entry before labelToIdx,
+			// so it can be found by possible concurrent goroutines.
+			//
+			// We might store duplicated entries for single label with different indexes,
+			// and it's fine, see https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7118.
 			lc.idxToLabel.Store(idx, labelCopy)
-			lc.labelToIdx.Store(labelCopy, v)
+			vNew, loaded := lc.labelToIdx.LoadOrStore(labelCopy, v)
+			if loaded {
+				// This label has been stored by a concurrent goroutine with different index,
+				// use it for key consistency in aggrState.
+				v = vNew
+			}
 
 			// Update lc.totalSizeBytes
 			labelSizeBytes := uint64(len(label.Name) + len(label.Value))


Test patch:
diff --git a/lib/promutils/labelscompressor_test.go b/lib/promutils/labelscompressor_test.go
index b5d787be7..9b520dba8 100644
--- a/lib/promutils/labelscompressor_test.go
+++ b/lib/promutils/labelscompressor_test.go
@@ -68,6 +68,7 @@ func TestLabelsCompressorSerial(t *testing.T) {
 func TestLabelsCompressorConcurrent(t *testing.T) {
 	const concurrency = 5
 	var lc LabelsCompressor
+	var expectCompressedKeys sync.Map
 
 	var wg sync.WaitGroup
 	for i := 0; i < concurrency; i++ {
@@ -75,9 +76,14 @@ func TestLabelsCompressorConcurrent(t *testing.T) {
 		go func() {
 			defer wg.Done()
 			series := newTestSeries(100, 20)
-			for i, labels := range series {
+			for n, labels := range series {
 				sExpected := labelsToString(labels)
 				data := lc.Compress(nil, labels)
+				if expectData, ok := expectCompressedKeys.LoadOrStore(n, data); ok {
+					if string(data) != string(expectData.([]byte)) {
+						panic(fmt.Errorf("unexpected compress result at series/%d in iteration %d ", n, i))
+					}
+				}
 				labelsResult := lc.Decompress(nil, data)
 				sResult := labelsToString(labelsResult)
 				if sExpected != sResult {