Merge branch 'public-single-node' into victorialogs-wip

docs/CHANGELOG.md

@@ -49,6 +49,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent/): prevent potential panic during [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html) if more than one `--remoteWrite.streamAggr.dedupInterval` is configured. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6205).
 * BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): set correct suffix `<output>_prometheus` for aggregation outputs [increase_prometheus](https://docs.victoriametrics.com/stream-aggregation/#increase_prometheus) and [total_prometheus](https://docs.victoriametrics.com/stream-aggregation/#total_prometheus). Before, outputs `total` and `total_prometheus` or `increase` and `increase_prometheus` had the same suffix.
 * BUGFIX: properly estimate the needed memory for query execution if it has the format [`aggr_func`](https://docs.victoriametrics.com/metricsql/#aggregate-functions)([`rollup_func[d]`](https://docs.victoriametrics.com/metricsql/#rollup-functions) (for example, `sum(rate(request_duration_seconds_bucket[5m]))`). This should allow performing aggregations over bigger number of time series when VictoriaMetrics runs in environments with small amounts of available memory. The issue has been introduced in [this commit](https://github.com/VictoriaMetrics/VictoriaMetrics/commit/5138eaeea0791caa34bcfab410e0ca9cd253cd8f) in [v1.83.0](https://docs.victoriametrics.com/changelog_2022/#v1830).
+* BUGFIX: [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): correctly apply `-inmemoryDataFlushInterval` when it's set to minimum supported value 1s.
 
 * DEPRECATION: [vmagent](https://docs.victoriametrics.com/vmagent/): removed deprecated `-remoteWrite.multitenantURL` flag from vmagent. This flag was deprecated since [v1.96.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.96.0). Use `-enableMultitenantHandlers` instead, as it is easier to use and combine with [multitenant URL at vminsert](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#multitenancy-via-labels). See these [docs for details](https://docs.victoriametrics.com/vmagent.html#multitenancy).
 

lib/mergeset/table.go

@@ -57,7 +57,7 @@ var dataFlushInterval = 5 * time.Second
 //
 // This function must be called before initializing the indexdb.
 func SetDataFlushInterval(d time.Duration) {
-	if d > pendingItemsFlushInterval {
+	if d >= time.Second {
 		dataFlushInterval = d
 	}
 }

lib/storage/partition.go

@@ -58,7 +58,7 @@ var dataFlushInterval = 5 * time.Second
 //
 // This function must be called before initializing the storage.
 func SetDataFlushInterval(d time.Duration) {
-	if d > pendingRowsFlushInterval {
+	if d >= time.Second {
 		dataFlushInterval = d
 		mergeset.SetDataFlushInterval(d)
 	}

lib/streamaggr/rate.go

@@ -19,7 +19,7 @@ type rateAggrState struct {
 
 type rateStateValue struct {
 	mu             sync.Mutex
-	lastValues     map[string]*rateLastValueState
+	lastValues     map[string]rateLastValueState
 	deleteDeadline uint64
 	deleted        bool
 }
@@ -57,7 +57,7 @@ func (as *rateAggrState) pushSamples(samples []pushSample) {
 		if !ok {
 			// The entry is missing in the map. Try creating it.
 			v = &rateStateValue{
-				lastValues: make(map[string]*rateLastValueState),
+				lastValues: make(map[string]rateLastValueState),
 			}
 			vNew, loaded := as.m.LoadOrStore(outputKey, v)
 			if loaded {
@@ -85,8 +85,6 @@ func (as *rateAggrState) pushSamples(samples []pushSample) {
 					// counter reset
 					lv.total += s.value
 				}
-			} else {
-				lv = &rateLastValueState{}
 			}
 			lv.value = s.value
 			lv.timestamp = s.timestamp
@@ -103,7 +101,8 @@ func (as *rateAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
+func (as *rateAggrState) flushState(ctx *flushCtx, resetState bool) {
+	_ = resetState // it isn't used here
 	currentTime := fasttime.UnixTimestamp()
 	currentTimeMsec := int64(currentTime) * 1000
 
@@ -132,6 +131,7 @@ func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
 				rate += v1.total * 1000 / float64(v1.timestamp-v1.prevTimestamp)
 				v1.prevTimestamp = v1.timestamp
 				v1.total = 0
+				m[k1] = v1
 			}
 		}
 		if as.suffix == "rate_avg" {

lib/streamaggr/streamaggr_timing_test.go

@@ -15,6 +15,8 @@ var benchOutputs = []string{
 	"total_prometheus",
 	"increase",
 	"increase_prometheus",
+	"rate_sum",
+	"rate_avg",
 	"count_series",
 	"count_samples",
 	"unique_samples",