lib/streamaggr: move increase and increase_prometheus outputs to a separate struct

See https://github.com/golangci/golangci-lint/releases/tag/v2.12.2

While at it, actualize .golangci.yml .

.github/workflows/build.yml

@@ -33,7 +33,8 @@ jobs:
     name: ${{ matrix.os }}-${{ matrix.arch }}
     permissions:
       contents: read
-    runs-on: ubuntu-latest
+    # Runs on dedicated runner with extra resources to increase build speed.
+    runs-on: 'vm-runner'
     strategy:
       fail-fast: false
       matrix:

.github/workflows/test.yml

@@ -30,7 +30,8 @@ jobs:
     name: lint
     permissions:
       contents: read
-    runs-on: ubuntu-latest
+    # Runs on dedicated runner with extra resources since golangci-lint requires extra memory
+    runs-on: 'vm-runner'
     steps:
       - name: Code checkout
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
@@ -64,7 +65,8 @@ jobs:
     name: unit
     permissions:
       contents: read
-    runs-on: ubuntu-latest
+    # Runs on dedicated runner with extra resources to increase tests speed.
+    runs-on: 'vm-runner'
 
     strategy:
       matrix:
@@ -95,6 +97,7 @@ jobs:
     name: apptest
     permissions:
       contents: read
+    # Runs on dedicated runner to isolate app tests from other tests.
     runs-on: apptest
 
     steps:

.golangci.yml

@@ -3,27 +3,14 @@ linters:
   settings:
     errcheck:
       exclude-functions:
-        - fmt.Fprintf
-        - fmt.Fprint
         - (net/http.ResponseWriter).Write
   exclusions:
     generated: lax
     presets:
-      - common-false-positives
-      - legacy
       - std-error-handling
     rules:
       - linters:
           - staticcheck
         text: 'SA(4003|1019|5011):'
     paths:
-      - third_party$
-      - builtin$
-      - examples$
-formatters:
-  exclusions:
-    generated: lax
-    paths:
-      - third_party$
-      - builtin$
-      - examples$
+      - ^app/vmui/

Makefile

@@ -17,7 +17,7 @@ EXTRA_GO_BUILD_TAGS ?=
 GO_BUILDINFO = -X '$(PKG_PREFIX)/lib/buildinfo.Version=$(APP_NAME)-$(DATEINFO_TAG)-$(BUILDINFO_TAG)'
 TAR_OWNERSHIP ?= --owner=1000 --group=1000
 
-GOLANGCI_LINT_VERSION := 2.9.0
+GOLANGCI_LINT_VERSION := 2.12.2
 
 .PHONY: $(MAKECMDGOALS)
 
@@ -527,7 +527,7 @@ golangci-lint: install-golangci-lint
 	golangci-lint run --build-tags 'synctest'
 
 install-golangci-lint:
-	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
+	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://golangci-lint.run/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
 
 remove-golangci-lint:
 	rm -rf `which golangci-lint`

app/vmalert/config/config.go

@@ -173,9 +173,9 @@ func (r *Rule) String() string {
 	if r.Alert != "" {
 		ruleType = "alerting"
 	}
-	b := strings.Builder{}
-	b.WriteString(fmt.Sprintf("%s rule %q", ruleType, r.Name()))
-	b.WriteString(fmt.Sprintf("; expr: %q", r.Expr))
+	var b strings.Builder
+	fmt.Fprintf(&b, "%s rule %q", ruleType, r.Name())
+	fmt.Fprintf(&b, "; expr: %q", r.Expr)
 
 	kv := sortMap(r.Labels)
 	for i := range kv {

app/vmui/packages/vmui/package.json

@@ -21,16 +21,16 @@
   },
   "dependencies": {
     "classnames": "^2.5.1",
-    "dayjs": "^1.11.20",
+    "dayjs": "^1.11.21",
     "lodash.debounce": "^4.0.8",
-    "marked": "^18.0.2",
-    "preact": "^10.29.1",
-    "qs": "^6.15.1",
+    "marked": "^18.0.5",
+    "preact": "^10.29.2",
+    "qs": "^6.15.2",
     "react-input-mask": "^2.0.4",
-    "react-router-dom": "^7.14.1",
+    "react-router-dom": "^7.17.0",
     "uplot": "^1.6.32",
-    "vite": "^8.0.8",
-    "web-vitals": "^5.2.0"
+    "vite": "^8.0.16",
+    "web-vitals": "^5.3.0"
   },
   "devDependencies": {
     "@eslint/eslintrc": "^3.3.5",
@@ -39,24 +39,24 @@
     "@testing-library/jest-dom": "^6.9.1",
     "@testing-library/preact": "^3.2.4",
     "@types/lodash.debounce": "^4.0.9",
-    "@types/node": "^25.6.0",
-    "@types/qs": "^6.15.0",
-    "@types/react": "^19.2.14",
+    "@types/node": "^25.9.2",
+    "@types/qs": "^6.15.1",
+    "@types/react": "^19.2.17",
     "@types/react-input-mask": "^3.0.6",
     "@types/react-router-dom": "^5.3.3",
-    "@typescript-eslint/eslint-plugin": "^8.58.2",
-    "@typescript-eslint/parser": "^8.58.2",
+    "@typescript-eslint/eslint-plugin": "^8.61.0",
+    "@typescript-eslint/parser": "^8.61.0",
     "cross-env": "^10.1.0",
     "eslint": "^9.39.2",
     "eslint-plugin-react": "^7.37.5",
     "eslint-plugin-unused-imports": "^4.4.1",
-    "globals": "^17.5.0",
-    "http-proxy-middleware": "^3.0.5",
-    "jsdom": "^29.0.2",
-    "postcss": "^8.5.10",
-    "sass-embedded": "^1.99.0",
-    "typescript": "^6.0.2",
-    "vitest": "^4.1.4"
+    "globals": "^17.6.0",
+    "http-proxy-middleware": "^4.1.0",
+    "jsdom": "^29.1.1",
+    "postcss": "^8.5.15",
+    "sass-embedded": "^1.100.0",
+    "typescript": "^6.0.3",
+    "vitest": "^4.1.8"
   },
   "browserslist": {
     "production": [

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,6 +26,8 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 ## tip
 
+* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/victoriametrics/stream-aggregation/): fix issue with producing aggregated samples with identical timestamps between flushes. See PR [#10808](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/10808) for details.
+
 ## [v1.145.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.145.0)
 
 Released at 2026-06-08

lib/promscrape/scrapework_test.go

@@ -667,11 +667,11 @@ func TestScrapeWorkScrapeInternalStreamConcurrency(t *testing.T) {
 	}
 
 	generateScrape := func(n int) string {
-		w := strings.Builder{}
+		var w strings.Builder
 		for i := range n {
-			w.WriteString(fmt.Sprintf("fooooo_%d 1\n", i))
+			fmt.Fprintf(&w, "fooooo_%d 1\n", i)
 			if i%100 == 0 {
-				w.WriteString(fmt.Sprintf("# HELP fooooo_%d This is a test\n", i))
+				fmt.Fprintf(&w, "# HELP fooooo_%d This is a test\n", i)
 			}
 		}
 		return w.String()
@@ -1005,9 +1005,9 @@ func TestSendStaleSeries(t *testing.T) {
 		}
 	}
 	generateScrape := func(n int) string {
-		w := strings.Builder{}
+		var w strings.Builder
 		for i := range n {
-			w.WriteString(fmt.Sprintf("foo_%d 1\n", i))
+			fmt.Fprintf(&w, "foo_%d 1\n", i)
 		}
 		return w.String()
 	}

lib/streamaggr/deduplicator.go

@@ -231,6 +231,7 @@ func (d *Deduplicator) flush(pushFunc PushFunc) {
 		logger.Warnf("deduplication couldn't be finished in the configured dedupInterval=%s; it took %.03fs; "+
 			"possible solutions: increase dedupInterval; reduce samples' ingestion rate", d.interval, duration.Seconds())
 	}
+	deadlineTime = deadlineTime.Add(d.interval)
 	for time.Now().After(deadlineTime) {
 		deadlineTime = deadlineTime.Add(d.interval)
 	}

lib/streamaggr/histogram_bucket.go

@@ -14,16 +14,10 @@ func (av *histogramBucketAggrValue) pushSample(_ aggrConfig, sample *pushSample,
 	av.h.Update(sample.value)
 }
 
-func (av *histogramBucketAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, _ bool) {
-	ac := c.(*histogramBucketAggrConfig)
-	shared := av.shared
-	if ac.useSharedState {
-		shared.Merge(&av.h)
-		av.h.Reset()
-	} else {
-		shared = &av.h
-	}
-	shared.VisitNonZeroBuckets(func(vmrange string, count uint64) {
+func (av *histogramBucketAggrValue) flush(_ aggrConfig, ctx *flushCtx, key string, _ bool) {
+	av.shared.Merge(&av.h)
+	av.h.Reset()
+	av.shared.VisitNonZeroBuckets(func(vmrange string, count uint64) {
 		ctx.appendSeriesWithExtraLabel(key, "histogram_bucket", float64(count), "vmrange", vmrange)
 	})
 }
@@ -32,26 +26,17 @@ func (av *histogramBucketAggrValue) state() any {
 	return av.shared
 }
 
-func newHistogramBucketAggrConfig(useSharedState bool) aggrConfig {
-	return &histogramBucketAggrConfig{
-		useSharedState: useSharedState,
-	}
+func newHistogramBucketAggrConfig() aggrConfig {
+	return &histogramBucketAggrConfig{}
 }
 
-type histogramBucketAggrConfig struct {
-	useSharedState bool
-}
+type histogramBucketAggrConfig struct{}
 
-func (ac *histogramBucketAggrConfig) getValue(s any) aggrValue {
-	var shared *metrics.Histogram
-	if ac.useSharedState {
-		if s == nil {
-			shared = &metrics.Histogram{}
-		} else {
-			shared = s.(*metrics.Histogram)
-		}
+func (*histogramBucketAggrConfig) getValue(s any) aggrValue {
+	if s == nil {
+		s = &metrics.Histogram{}
 	}
 	return &histogramBucketAggrValue{
-		shared: shared,
+		shared: s.(*metrics.Histogram),
 	}
 }

lib/streamaggr/increase.go

@@ -0,0 +1,109 @@
+package streamaggr
+
+import (
+	"fmt"
+
+	"github.com/VictoriaMetrics/metrics"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+type increaseLastValue struct {
+	value          float64
+	timestamp      int64
+	deleteDeadline int64
+}
+
+type increaseAggrConfig struct {
+	keepFirstSample bool
+
+	// The first sample per each new series is ignored until this unix timestamp deadline in seconds even if keepFirstSample is set.
+	// This allows avoiding an initial spike of the output values at startup when new time series
+	// cannot be distinguished from already existing series. This is tracked with ignoreFirstSampleDeadline.
+	ignoreFirstSampleDeadline uint64
+	counterResetsTotal        *metrics.Counter
+}
+
+type increaseAggrValue struct {
+	total  *float64
+	shared map[string]increaseLastValue
+}
+
+func (av *increaseAggrValue) pushSample(c aggrConfig, sample *pushSample, key string, deleteDeadline int64) {
+	ac := c.(*increaseAggrConfig)
+	currentTime := fasttime.UnixTimestamp()
+	keepFirstSample := ac.keepFirstSample && currentTime >= ac.ignoreFirstSampleDeadline
+	lv, ok := av.shared[key]
+	if av.total == nil {
+		av.total = new(float64)
+	}
+	if ok {
+		if sample.timestamp < lv.timestamp {
+			// Skip out of order sample
+			return
+		}
+		if sample.value >= lv.value {
+			*av.total += sample.value - lv.value
+		} else {
+			// counter reset
+			*av.total += sample.value
+			ac.counterResetsTotal.Inc()
+		}
+	} else if keepFirstSample {
+		*av.total += sample.value
+	}
+	lv.value = sample.value
+	lv.timestamp = sample.timestamp
+	lv.deleteDeadline = deleteDeadline
+	key = bytesutil.InternString(key)
+	av.shared[key] = lv
+}
+
+func (av *increaseAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, isLast bool) {
+	ac := c.(*increaseAggrConfig)
+	for lk, lv := range av.shared {
+		if ctx.flushTimestamp > lv.deleteDeadline || isLast {
+			delete(av.shared, lk)
+		}
+	}
+	if av.total == nil {
+		return
+	}
+	total := *av.total
+	av.total = nil
+	ctx.appendSeries(key, ac.getSuffix(), total)
+}
+
+func (av *increaseAggrValue) state() any {
+	return av.shared
+}
+
+func newIncreaseAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleIntervalSecs uint64, keepFirstSample bool) aggrConfig {
+	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + ignoreFirstSampleIntervalSecs
+	cfg := &increaseAggrConfig{
+		keepFirstSample:           keepFirstSample,
+		ignoreFirstSampleDeadline: ignoreFirstSampleDeadline,
+	}
+	cfg.counterResetsTotal = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_counter_resets_total{%s}`, metricLabels))
+	return cfg
+}
+
+func (*increaseAggrConfig) getValue(s any) aggrValue {
+	var shared map[string]increaseLastValue
+	if s == nil {
+		shared = make(map[string]increaseLastValue)
+	} else {
+		shared = s.(map[string]increaseLastValue)
+	}
+	return &increaseAggrValue{
+		shared: shared,
+	}
+}
+
+func (ac *increaseAggrConfig) getSuffix() string {
+	if ac.keepFirstSample {
+		return "increase"
+	}
+	return "increase_prometheus"
+}

lib/streamaggr/output.go

@@ -75,6 +75,9 @@ func (ao *aggrOutputs) pushSamples(samples []pushSample, deleteDeadline int64, i
 				outputs = av.blue
 			}
 			for idx, o := range outputs {
+				if o == nil {
+					o = av.blue[idx]
+				}
 				o.pushSample(ao.configs[idx], sample, inputKey, deleteDeadline)
 			}
 			av.deleteDeadline = deleteDeadline
@@ -112,6 +115,9 @@ func (ao *aggrOutputs) flushState(ctx *flushCtx) {
 			outputs = av.blue
 		}
 		for i, o := range outputs {
+			if o == nil {
+				o = av.blue[i]
+			}
 			o.flush(ao.configs[i], ctx, outputKey, ctx.isLast)
 		}
 		av.mu.Unlock()

lib/streamaggr/streamaggr.go

@@ -609,7 +609,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	outputsSeen := make(map[string]struct{}, len(cfg.Outputs))
 	for i, output := range cfg.Outputs {
 		outputMetricLabels := fmt.Sprintf(`output=%q,name=%q,path=%q,url=%q,position="%d"`, output, name, path, alias, aggrID)
-		ac, err := newOutputConfig(ms, outputMetricLabels, output, outputsSeen, useSharedState, ignoreFirstSampleInterval)
+		ac, err := newOutputConfig(ms, outputMetricLabels, output, outputsSeen, ignoreFirstSampleInterval)
 		if err != nil {
 			return nil, err
 		}
@@ -716,7 +716,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	return a, nil
 }
 
-func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen map[string]struct{}, useSharedState bool, ignoreFirstSampleInterval time.Duration) (aggrConfig, error) {
+func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen map[string]struct{}, ignoreFirstSampleInterval time.Duration) (aggrConfig, error) {
 	// check for duplicated output
 	if _, ok := outputsSeen[output]; ok {
 		return nil, fmt.Errorf("`outputs` list contains duplicate aggregation function: %s", output)
@@ -760,11 +760,11 @@ func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen m
 	case "count_series":
 		return newCountSeriesAggrConfig(), nil
 	case "histogram_bucket":
-		return newHistogramBucketAggrConfig(useSharedState), nil
+		return newHistogramBucketAggrConfig(), nil
 	case "increase":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true, true), nil
+		return newIncreaseAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true), nil
 	case "increase_prometheus":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true, false), nil
+		return newIncreaseAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false), nil
 	case "last":
 		return newLastAggrConfig(), nil
 	case "max":
@@ -782,9 +782,9 @@ func newOutputConfig(ms *metrics.Set, metricLabels, output string, outputsSeen m
 	case "sum_samples":
 		return newSumSamplesAggrConfig(), nil
 	case "total":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false, true), nil
+		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, true), nil
 	case "total_prometheus":
-		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false, false), nil
+		return newTotalAggrConfig(ms, metricLabels, ignoreFirstSampleIntervalSecs, false), nil
 	case "unique_samples":
 		return newUniqueSamplesAggrConfig(), nil
 	default:
@@ -845,6 +845,7 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipFlu
 			} else {
 				a.flush(pf, flushTime, cs, false)
 			}
+			flushTime = flushTime.Add(a.interval)
 			for time.Now().After(flushTime) {
 				flushTime = flushTime.Add(a.interval)
 			}

lib/streamaggr/streamaggr_synctest_test.go

@@ -1,5 +1,3 @@
-//go:build synctest
-
 package streamaggr
 
 import (
@@ -485,10 +483,8 @@ foo 3.3
 `, ``, ``, ``, ``}, time.Minute, `foo:1m_count_series 1
 foo:1m_count_series{bar="baz"} 1
 foo:1m_sum_samples 0
-foo:1m_sum_samples 0
 foo:1m_sum_samples 4.3
 foo:1m_sum_samples{bar="baz"} 0
-foo:1m_sum_samples{bar="baz"} 0
 foo:1m_sum_samples{bar="baz"} 2
 foo:5m_by_bar_sum_samples 4.3
 foo:5m_by_bar_sum_samples{bar="baz"} 2
@@ -694,21 +690,29 @@ foo:1m_by_cde_rate_sum{cde="1"} 0.125
 
 	// test rate_sum and rate_avg, when two aggregation intervals are empty
 	f([]string{`
-foo{abc="123", cde="1"} 2
-foo{abc="456", cde="1"} 8
-foo{abc="777", cde="1"} 9 -10
+foo{abc="123", cde="1"} 1
+foo{abc="123", cde="1"} 2 1
+foo{abc="456", cde="1"} 7
+foo{abc="456", cde="1"} 8 1
+foo{abc="777", cde="1"} 8
+foo{abc="777", cde="1"} 9 1
 `, ``, ``, `
-foo{abc="123", cde="1"} 20
+foo{abc="123", cde="1"} 19
+foo{abc="123", cde="1"} 20 1
 foo{abc="456", cde="1"} 26
-foo{abc="777", cde="1"} 27 -10
-`}, time.Minute, `foo:1m_by_cde_rate_avg{cde="1"} 0.1
-foo:1m_by_cde_rate_sum{cde="1"} 0.2
+foo{abc="456", cde="1"} 27 1
+foo{abc="777", cde="1"} 27
+foo{abc="777", cde="1"} 28 1
+`}, time.Minute, `foo:1m_by_cde_rate_avg{cde="1"} 1
+foo:1m_by_cde_rate_avg{cde="1"} 1
+foo:1m_by_cde_rate_sum{cde="1"} 3
+foo:1m_by_cde_rate_sum{cde="1"} 3
 `, `            
 - interval: 1m
   by: [cde]
   outputs: [rate_sum, rate_avg]
   enable_windows: true
-`, "111111")
+`, "111111111111")
 
 	// rate_sum and rate_avg with duplicated events
 	f([]string{`

lib/streamaggr/total.go

@@ -53,36 +53,30 @@ func (av *totalAggrValue) pushSample(c aggrConfig, sample *pushSample, key strin
 
 func (av *totalAggrValue) flush(c aggrConfig, ctx *flushCtx, key string, isLast bool) {
 	ac := c.(*totalAggrConfig)
-	suffix := ac.getSuffix()
-	// check for stale entries
 	total := av.shared.total + av.total
 	av.total = 0
-	lvs := av.shared.lastValues
-	for lk, lv := range lvs {
+	for lk, lv := range av.shared.lastValues {
 		if ctx.flushTimestamp > lv.deleteDeadline || isLast {
-			delete(lvs, lk)
+			delete(av.shared.lastValues, lk)
 		}
 	}
-	if ac.resetTotalOnFlush {
-		av.shared.total = 0
-	} else if math.Abs(total) >= (1 << 53) {
+	if math.Abs(total) >= (1 << 53) {
 		// It is time to reset the entry, since it starts losing float64 precision
 		av.shared.total = 0
 	} else {
 		av.shared.total = total
 	}
-	ctx.appendSeries(key, suffix, total)
+	ctx.appendSeries(key, ac.getSuffix(), total)
 }
 
 func (av *totalAggrValue) state() any {
 	return av.shared
 }
 
-func newTotalAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleIntervalSecs uint64, resetTotalOnFlush, keepFirstSample bool) aggrConfig {
+func newTotalAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleIntervalSecs uint64, keepFirstSample bool) aggrConfig {
 	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + ignoreFirstSampleIntervalSecs
 	cfg := &totalAggrConfig{
 		keepFirstSample:           keepFirstSample,
-		resetTotalOnFlush:         resetTotalOnFlush,
 		ignoreFirstSampleDeadline: ignoreFirstSampleDeadline,
 	}
 	cfg.counterResetsTotal = ms.NewCounter(fmt.Sprintf(`vm_streamaggr_counter_resets_total{%s}`, metricLabels))
@@ -90,8 +84,6 @@ func newTotalAggrConfig(ms *metrics.Set, metricLabels string, ignoreFirstSampleI
 }
 
 type totalAggrConfig struct {
-	resetTotalOnFlush bool
-
 	// Whether to take into account the first sample in new time series when calculating the output value.
 	keepFirstSample bool
 
@@ -117,12 +109,6 @@ func (*totalAggrConfig) getValue(s any) aggrValue {
 }
 
 func (ac *totalAggrConfig) getSuffix() string {
-	if ac.resetTotalOnFlush {
-		if ac.keepFirstSample {
-			return "increase"
-		}
-		return "increase_prometheus"
-	}
 	if ac.keepFirstSample {
 		return "total"
 	}