Merge branch 'master' into streaming-aggregation

app/victoria-metrics/main.go

@@ -117,6 +117,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 			{"expand-with-exprs", "WITH expressions' tutorial"},
 			{"api/v1/targets", "advanced information about discovered targets in JSON format"},
 			{"config", "-promscrape.config contents"},
+			{"stream-agg", "streaming aggregation status"},
 			{"metrics", "available service metrics"},
 			{"flags", "command-line flags"},
 			{"api/v1/status/tsdb", "tsdb status page"},

app/vmagent/main.go

@@ -42,6 +42,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
 )
 
 var (
@@ -228,6 +229,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 			{"metric-relabel-debug", "debug metric relabeling"},
 			{"api/v1/targets", "advanced information about discovered targets in JSON format"},
 			{"config", "-promscrape.config contents"},
+			{"stream-agg", "streaming aggregation status"},
 			{"metrics", "available service metrics"},
 			{"flags", "command-line flags"},
 			{"-/reload", "reload configuration"},
@@ -432,6 +434,9 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		procutil.SelfSIGHUP()
 		w.WriteHeader(http.StatusOK)
 		return true
+	case "/stream-agg":
+		streamaggr.WriteHumanReadableState(w, r, remotewrite.GetAggregators())
+		return true
 	case "/ready":
 		if rdy := atomic.LoadInt32(&promscrape.PendingScrapeConfigs); rdy > 0 {
 			errMsg := fmt.Sprintf("waiting for scrapes to init, left: %d", rdy)

app/vmagent/remotewrite/remotewrite.go

@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -40,6 +41,8 @@ var (
 		"Pass multiple -remoteWrite.multitenantURL flags in order to replicate data to multiple remote storage systems. See also -remoteWrite.url")
 	shardByURL = flag.Bool("remoteWrite.shardByURL", false, "Whether to shard outgoing series across all the remote storage systems enumerated via -remoteWrite.url . "+
 		"By default the data is replicated across all the -remoteWrite.url . See https://docs.victoriametrics.com/vmagent.html#sharding-among-remote-storages")
+	shardByURLLabels = flag.String("remoteWrite.shardByURL.labels", "", "Comma-separated list of label names for sharding across all the -remoteWrite.url. All labels of timeseries are used by default. "+
+		"See also -remoteWrite.shardByURL and https://docs.victoriametrics.com/vmagent.html#sharding-among-remote-storages")
 	tmpDataPath = flag.String("remoteWrite.tmpDataPath", "vmagent-remotewrite-data", "Path to directory where temporary data for remote write component is stored. "+
 		"See also -remoteWrite.maxDiskUsagePerURL")
 	keepDanglingQueues = flag.Bool("remoteWrite.keepDanglingQueues", false, "Keep persistent queues contents at -remoteWrite.tmpDataPath in case there are no matching -remoteWrite.url. "+
@@ -92,6 +95,8 @@ var (
 
 	// Data without tenant id is written to defaultAuthToken if -remoteWrite.multitenantURL is specified.
 	defaultAuthToken = &auth.Token{}
+
+	shardLabelsFilter map[string]struct{}
 )
 
 // MultitenancyEnabled returns true if -remoteWrite.multitenantURL is specified.
@@ -171,6 +176,12 @@ func Init() {
 		rwctxsDefault = newRemoteWriteCtxs(nil, *remoteWriteURLs)
 	}
 
+	if *shardByURLLabels != "" {
+		for _, label := range strings.Split(*shardByURLLabels, ",") {
+			shardLabelsFilter[strings.TrimSpace(label)] = struct{}{}
+		}
+	}
+
 	// Start config reloader.
 	configReloaderWG.Add(1)
 	go func() {
@@ -419,7 +430,7 @@ func pushBlockToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prompbmarsha
 		// Shard the data among rwctxs
 		tssByURL := make([][]prompbmarshal.TimeSeries, len(rwctxs))
 		for _, ts := range tssBlock {
-			h := getLabelsHash(ts.Labels)
+			h := getLabelsHash(ts.Labels, shardLabelsFilter)
 			idx := h % uint64(len(tssByURL))
 			tssByURL[idx] = append(tssByURL[idx], ts)
 		}
@@ -472,7 +483,7 @@ func limitSeriesCardinality(tss []prompbmarshal.TimeSeries) []prompbmarshal.Time
 	dst := make([]prompbmarshal.TimeSeries, 0, len(tss))
 	for i := range tss {
 		labels := tss[i].Labels
-		h := getLabelsHash(labels)
+		h := getLabelsHash(labels, nil)
 		if hourlySeriesLimiter != nil && !hourlySeriesLimiter.Add(h) {
 			hourlySeriesLimitRowsDropped.Add(len(tss[i].Samples))
 			logSkippedSeries(labels, "-remoteWrite.maxHourlySeries", hourlySeriesLimiter.MaxItems())
@@ -496,10 +507,16 @@ var (
 	dailySeriesLimitRowsDropped  = metrics.NewCounter(`vmagent_daily_series_limit_rows_dropped_total`)
 )
 
-func getLabelsHash(labels []prompbmarshal.Label) uint64 {
+func getLabelsHash(labels []prompbmarshal.Label, filterLabels map[string]struct{}) uint64 {
 	bb := labelsHashBufPool.Get()
 	b := bb.B[:0]
 	for _, label := range labels {
+		if len(filterLabels) > 0 {
+			_, ok := filterLabels[label.Name]
+			if !ok {
+				continue
+			}
+		}
 		b = append(b, label.Name...)
 		b = append(b, label.Value...)
 	}
@@ -802,3 +819,23 @@ func CheckStreamAggrConfigs() error {
 	}
 	return nil
 }
+
+func GetAggregators() map[string]*streamaggr.Aggregators {
+	var result = map[string]*streamaggr.Aggregators{}
+
+	if len(*remoteWriteMultitenantURLs) > 0 {
+		rwctxsMapLock.Lock()
+		for tenant, rwctxs := range rwctxsMap {
+			for rwNum, rw := range rwctxs {
+				result[fmt.Sprintf("rw %d for tenant %v:%v", rwNum, tenant.AccountID, tenant.ProjectID)] = rw.sas.Load()
+			}
+		}
+		rwctxsMapLock.Unlock()
+	} else {
+		for rwNum, rw := range rwctxsDefault {
+			result[fmt.Sprintf("remote write %d", rwNum)] = rw.sas.Load()
+		}
+	}
+
+	return result
+}

app/vminsert/common/streamaggr.go

@@ -211,3 +211,7 @@ func pushAggregateSeries(tss []prompbmarshal.TimeSeries) {
 		logger.Errorf("cannot flush aggregate series: %s", err)
 	}
 }
+
+func GetAggregators() map[string]*streamaggr.Aggregators {
+	return map[string]*streamaggr.Aggregators{"default": sasGlobal.Load()}
+}

app/vminsert/main.go

@@ -4,6 +4,7 @@ import (
 	"embed"
 	"flag"
 	"fmt"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
 	"net/http"
 	"strings"
 	"sync/atomic"
@@ -326,6 +327,9 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		procutil.SelfSIGHUP()
 		w.WriteHeader(http.StatusNoContent)
 		return true
+	case "/stream-agg":
+		streamaggr.WriteHumanReadableState(w, r, vminsertCommon.GetAggregators())
+		return true
 	case "/ready":
 		if rdy := atomic.LoadInt32(&promscrape.PendingScrapeConfigs); rdy > 0 {
 			errMsg := fmt.Sprintf("waiting for scrape config to init targets, configs left: %d", rdy)

app/vmselect/promql/parser_test.go

@@ -0,0 +1,12 @@
+package promql
+
+import "testing"
+
+func TestIsMetricSelectorWithRollup(t *testing.T) {
+	childQuery, _, _ := IsMetricSelectorWithRollup("metric_name{label='value'}[365d] or vector(0)")
+	if childQuery != "" {
+		t.Fatalf("AAAAA: %v", childQuery)
+	} else {
+		t.Fatalf("BBBBB")
+	}
+}

app/vmselect/promql/rollup_test.go

@@ -160,7 +160,7 @@ func TestDerivValues(t *testing.T) {
 	testRowsEqual(t, values, timestamps, valuesExpected, timestamps)
 }
 
-func testRollupFunc(t *testing.T, funcName string, args []interface{}, vExpected float64) {
+func testRollupFuncWithValues(t *testing.T, funcName string, args []interface{}, vInput []float64, vTimestamps []int64, vExpected float64) {
 	t.Helper()
 	nrf := getRollupFunc(funcName)
 	if nrf == nil {
@@ -172,9 +172,11 @@ func testRollupFunc(t *testing.T, funcName string, args []interface{}, vExpected
 	}
 	var rfa rollupFuncArg
 	rfa.prevValue = nan
+	rfa.realPrevValue = nan
+	rfa.realNextValue = nan
 	rfa.prevTimestamp = 0
-	rfa.values = append(rfa.values, testValues...)
+	rfa.values = append(rfa.values, vInput...)
-	rfa.timestamps = append(rfa.timestamps, testTimestamps...)
+	rfa.timestamps = append(rfa.timestamps, vTimestamps...)
 	rfa.window = rfa.timestamps[len(rfa.timestamps)-1] - rfa.timestamps[0]
 	if rollupFuncsRemoveCounterResets[funcName] {
 		removeCounterResets(rfa.values)
@@ -194,6 +196,10 @@ func testRollupFunc(t *testing.T, funcName string, args []interface{}, vExpected
 	}
 }
 
+func testRollupFunc(t *testing.T, funcName string, args []interface{}, vExpected float64) {
+	testRollupFuncWithValues(t, funcName, args, testValues, testTimestamps, vExpected)
+}
+
 func TestRollupDurationOverTime(t *testing.T) {
 	f := func(maxInterval, dExpected float64) {
 		t.Helper()
@@ -1474,3 +1480,121 @@ func TestRollupDelta(t *testing.T) {
 	f(1, nan, nan, nil, 0)
 	f(100, nan, nan, nil, 0)
 }
+
+func TestIncrease(t *testing.T) {
+	f := func(funcName string, vInput []float64, vExpected float64) {
+		t.Helper()
+		var me metricsql.MetricExpr
+		args := []interface{}{&metricsql.RollupExpr{Expr: &me}}
+		testRollupFuncWithValues(t, funcName, args, vInput, []int64{1, 2}, vExpected)
+	}
+
+	f(
+		"increase",
+		[]float64{100, 100},
+		0,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 90},
+		0,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 88},
+		0,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 87},
+
+		187,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 187},
+
+		187,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 87, 200, 287},
+
+		387,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 187, 200, 287},
+
+		287,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 87, 200, 87},
+
+		387,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 87, 200, 187},
+
+		300,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 87, 200, 177},
+
+		300,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 13},
+		113,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 9},
+		9,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 1},
+		1,
+	)
+
+	f(
+		"increase",
+		[]float64{100, 0},
+		0,
+	)
+
+	f(
+		"increase",
+		[]float64{100, -1},
+		-1,
+	)
+
+	f(
+		"increase",
+		[]float64{100, -10},
+		90,
+	)
+
+	f(
+		"increase",
+		[]float64{100, -90},
+		10,
+	)
+}

lib/streamaggr/avg.go

@@ -2,27 +2,38 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // avgAggrState calculates output=avg, e.g. the average value over input samples.
 type avgAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type avgStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	sum     float64
+	sum            float64
-	count   int64
+	count          uint64
-	deleted bool
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newAvgAggrState() *avgAggrState {
+func newAvgAggrState(interval time.Duration, stalenessInterval time.Duration) *avgAggrState {
-	return &avgAggrState{}
+	return &avgAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *avgAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -45,6 +56,7 @@ again:
 	if !deleted {
 		sv.sum += value
 		sv.count++
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -54,21 +66,71 @@ again:
 	}
 }
 
-func (as *avgAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *avgAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*avgStateValue)
+
 		sv.mu.Lock()
-		avg := sv.sum / float64(sv.count)
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "avg", currentTimeMsec, avg)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *avgAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*avgStateValue)
+		sv.mu.Lock()
+		var avg float64
+		if sv.count > 0 {
+			avg = sv.sum / float64(sv.count)
+		}
+		sv.sum = 0
+		sv.count = 0
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, avg)
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *avgAggrState) getOutputName() string {
+	return "avg"
+}
+
+func (as *avgAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*avgStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.sum / float64(value.count),
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.count,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/count_samples.go

@@ -2,26 +2,37 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // countSamplesAggrState calculates output=countSamples, e.g. the count of input samples.
 type countSamplesAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type countSamplesStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	n       uint64
+	n              uint64
-	deleted bool
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newCountSamplesAggrState() *countSamplesAggrState {
+func newCountSamplesAggrState(interval time.Duration, stalenessInterval time.Duration) *countSamplesAggrState {
-	return &countSamplesAggrState{}
+	return &countSamplesAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *countSamplesAggrState) pushSample(_, outputKey string, _ float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -42,6 +53,7 @@ again:
 	deleted := sv.deleted
 	if !deleted {
 		sv.n++
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -51,21 +63,67 @@ again:
 	}
 }
 
-func (as *countSamplesAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *countSamplesAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*countSamplesStateValue)
+
 		sv.mu.Lock()
-		n := sv.n
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "count_samples", currentTimeMsec, float64(n))
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *countSamplesAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*countSamplesStateValue)
+		sv.mu.Lock()
+		n := sv.n
+		sv.n = 0
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, float64(n))
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *countSamplesAggrState) getOutputName() string {
+	return "count_samples"
+}
+
+func (as *countSamplesAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*countSamplesStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      float64(value.n),
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.n,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/count_series.go

@@ -2,27 +2,39 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // countSeriesAggrState calculates output=count_series, e.g. the number of unique series.
 type countSeriesAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type countSeriesStateValue struct {
-	mu            sync.Mutex
+	mu             sync.Mutex
-	countedSeries map[string]struct{}
+	countedSeries  map[string]struct{}
-	n             uint64
+	n              uint64
-	deleted       bool
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newCountSeriesAggrState() *countSeriesAggrState {
+func newCountSeriesAggrState(interval time.Duration, stalenessInterval time.Duration) *countSeriesAggrState {
-	return &countSeriesAggrState{}
+	return &countSeriesAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *countSeriesAggrState) pushSample(inputKey, outputKey string, _ float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -48,7 +60,9 @@ again:
 		if _, ok := sv.countedSeries[inputKey]; !ok {
 			sv.countedSeries[inputKey] = struct{}{}
 			sv.n++
+			sv.deleteDeadline = deleteDeadline
 		}
+		sv.samplesCount++
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -58,21 +72,72 @@ again:
 	}
 }
 
-func (as *countSeriesAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *countSeriesAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*countSeriesStateValue)
+
 		sv.mu.Lock()
-		n := sv.n
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "count_series", currentTimeMsec, float64(n))
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *countSeriesAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*countSeriesStateValue)
+		sv.mu.Lock()
+		n := sv.n
+		sv.n = 0
+		// todo: use builtin function clear after switching to go 1.21
+		for csk := range sv.countedSeries {
+			delete(sv.countedSeries, csk)
+		}
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, float64(n))
+		return true
+	})
+
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *countSeriesAggrState) getOutputName() string {
+	return "count_series"
+}
+
+func (as *countSeriesAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*countSeriesStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      float64(value.n),
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/histogram_bucket.go

@@ -1,6 +1,7 @@
 package streamaggr
 
 import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"math"
 	"sync"
 	"time"
@@ -11,22 +12,24 @@ import (
 
 // histogramBucketAggrState calculates output=histogramBucket, e.g. VictoriaMetrics histogram over input samples.
 type histogramBucketAggrState struct {
-	m sync.Map
+	m                 sync.Map
-
+	intervalSecs      uint64
-	stalenessSecs uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type histogramBucketStateValue struct {
 	mu             sync.Mutex
 	h              metrics.Histogram
+	samplesCount   uint64
 	deleteDeadline uint64
 	deleted        bool
 }
 
-func newHistogramBucketAggrState(stalenessInterval time.Duration) *histogramBucketAggrState {
+func newHistogramBucketAggrState(interval time.Duration, stalenessInterval time.Duration) *histogramBucketAggrState {
-	stalenessSecs := roundDurationToSecs(stalenessInterval)
 	return &histogramBucketAggrState{
-		stalenessSecs: stalenessSecs,
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
 	}
 }
 
@@ -50,6 +53,7 @@ again:
 	deleted := sv.deleted
 	if !deleted {
 		sv.h.Update(value)
+		sv.samplesCount++
 		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
@@ -93,12 +97,44 @@ func (as *histogramBucketAggrState) appendSeriesForFlush(ctx *flushCtx) {
 		if !sv.deleted {
 			key := k.(string)
 			sv.h.VisitNonZeroBuckets(func(vmrange string, count uint64) {
-				ctx.appendSeriesWithExtraLabel(key, "histogram_bucket", currentTimeMsec, float64(count), "vmrange", vmrange)
+				ctx.appendSeriesWithExtraLabel(key, as.getOutputName(), currentTimeMsec, float64(count), "vmrange", vmrange)
 			})
 		}
 		sv.mu.Unlock()
 		return true
 	})
+
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *histogramBucketAggrState) getOutputName() string {
+	return "count_series"
+}
+
+func (as *histogramBucketAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*histogramBucketStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		value.h.VisitNonZeroBuckets(func(vmrange string, count uint64) {
+			result = append(result, aggrStateRepresentation{
+				metric: getLabelsStringFromKey(k.(string), suffix, as.getOutputName(), prompbmarshal.Label{
+					Name:  vmrange,
+					Value: vmrange,
+				}),
+				currentValue:      float64(count),
+				lastPushTimestamp: as.lastPushTimestamp,
+				nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+				samplesCount:      value.samplesCount,
+			})
+		})
+		return true
+	})
+	return result
 }
 
 func roundDurationToSecs(d time.Duration) uint64 {

lib/streamaggr/increase.go

@@ -9,16 +9,18 @@ import (
 
 // increaseAggrState calculates output=increase, e.g. the increase over input counters.
 type increaseAggrState struct {
-	m sync.Map
+	m                   sync.Map
-
+	intervalSecs        uint64
 	ignoreInputDeadline uint64
 	stalenessSecs       uint64
+	lastPushTimestamp   uint64
 }
 
 type increaseStateValue struct {
 	mu             sync.Mutex
 	lastValues     map[string]*lastValueState
 	total          float64
+	samplesCount   uint64
 	deleteDeadline uint64
 	deleted        bool
 }
@@ -28,8 +30,9 @@ func newIncreaseAggrState(interval time.Duration, stalenessInterval time.Duratio
 	intervalSecs := roundDurationToSecs(interval)
 	stalenessSecs := roundDurationToSecs(stalenessInterval)
 	return &increaseAggrState{
-		ignoreInputDeadline: currentTime + intervalSecs,
+		intervalSecs:        intervalSecs,
 		stalenessSecs:       stalenessSecs,
+		ignoreInputDeadline: currentTime + intervalSecs,
 	}
 }
 
@@ -69,6 +72,7 @@ again:
 		lv.value = value
 		lv.deleteDeadline = deleteDeadline
 		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -122,8 +126,35 @@ func (as *increaseAggrState) appendSeriesForFlush(ctx *flushCtx) {
 		sv.mu.Unlock()
 		if !deleted {
 			key := k.(string)
-			ctx.appendSeries(key, "increase", currentTimeMsec, increase)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, increase)
 		}
 		return true
 	})
+
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *increaseAggrState) getOutputName() string {
+	return "increase"
+}
+
+func (as *increaseAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*increaseStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
 }

lib/streamaggr/increasepure.go

@@ -0,0 +1,153 @@
+package streamaggr
+
+import (
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// increasePureAggrState calculates output=increase_pure, e.g. the increasePure over input counters.
+type increasePureAggrState struct {
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
+}
+
+type increasePureStateValue struct {
+	mu             sync.Mutex
+	lastValues     map[string]*lastValueState
+	total          float64
+	samplesCount   uint64
+	deleteDeadline uint64
+	deleted        bool
+}
+
+func newIncreasePureAggrState(interval time.Duration, stalenessInterval time.Duration) *increasePureAggrState {
+	return &increasePureAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
+}
+
+func (as *increasePureAggrState) pushSample(inputKey, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &increasePureStateValue{
+			lastValues: make(map[string]*lastValueState),
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if loaded {
+			// Use the entry created by a concurrent goroutine.
+			v = vNew
+		}
+	}
+	sv := v.(*increasePureStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		lv, ok := sv.lastValues[inputKey]
+		if !ok {
+			lv = &lastValueState{}
+			sv.lastValues[inputKey] = lv
+		}
+		d := value
+		if ok && lv.value <= value {
+			d = value - lv.value
+		}
+		sv.total += d
+		lv.value = value
+		lv.deleteDeadline = deleteDeadline
+		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *increasePureAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*increasePureStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		} else {
+			// Delete outdated entries in sv.lastValues
+			m := sv.lastValues
+			for k1, v1 := range m {
+				if currentTime > v1.deleteDeadline {
+					delete(m, k1)
+				}
+			}
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *increasePureAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*increasePureStateValue)
+		sv.mu.Lock()
+		increasePure := sv.total
+		sv.total = 0
+		deleted := sv.deleted
+		sv.mu.Unlock()
+		if !deleted {
+			key := k.(string)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, increasePure)
+		}
+		return true
+	})
+
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *increasePureAggrState) getOutputName() string {
+	return "increase_pure"
+}
+
+func (as *increasePureAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*increasePureStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/last.go

@@ -2,26 +2,38 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // lastAggrState calculates output=last, e.g. the last value over input samples.
 type lastAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type lastStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	last    float64
+	last           float64
-	deleted bool
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newLastAggrState() *lastAggrState {
+func newLastAggrState(interval time.Duration, stalenessInterval time.Duration) *lastAggrState {
-	return &lastAggrState{}
+	return &lastAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *lastAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -42,6 +54,8 @@ again:
 	deleted := sv.deleted
 	if !deleted {
 		sv.last = value
+		sv.samplesCount++
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -51,21 +65,66 @@ again:
 	}
 }
 
-func (as *lastAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *lastAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*lastStateValue)
+
 		sv.mu.Lock()
-		last := sv.last
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "last", currentTimeMsec, last)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *lastAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*lastStateValue)
+		sv.mu.Lock()
+		last := sv.last
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, last)
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *lastAggrState) getOutputName() string {
+	return "last"
+}
+
+func (as *lastAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*lastStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.last,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/max.go

@@ -2,26 +2,38 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // maxAggrState calculates output=max, e.g. the maximum value over input samples.
 type maxAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type maxStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	max     float64
+	max            float64
-	deleted bool
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newMaxAggrState() *maxAggrState {
+func newMaxAggrState(interval time.Duration, stalenessInterval time.Duration) *maxAggrState {
-	return &maxAggrState{}
+	return &maxAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *maxAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -44,6 +56,8 @@ again:
 		if value > sv.max {
 			sv.max = value
 		}
+		sv.samplesCount++
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -53,21 +67,67 @@ again:
 	}
 }
 
-func (as *maxAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *maxAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*maxStateValue)
+
 		sv.mu.Lock()
-		max := sv.max
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "max", currentTimeMsec, max)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *maxAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*maxStateValue)
+		sv.mu.Lock()
+		value := sv.max
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, value)
+		return true
+	})
+
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *maxAggrState) getOutputName() string {
+	return "max"
+}
+
+func (as *maxAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*maxStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.max,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/min.go

@@ -2,26 +2,38 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // minAggrState calculates output=min, e.g. the minimum value over input samples.
 type minAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type minStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	min     float64
+	min            float64
-	deleted bool
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newMinAggrState() *minAggrState {
+func newMinAggrState(interval time.Duration, stalenessInterval time.Duration) *minAggrState {
-	return &minAggrState{}
+	return &minAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *minAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -43,6 +55,8 @@ again:
 	if !deleted {
 		if value < sv.min {
 			sv.min = value
+			sv.samplesCount++
+			sv.deleteDeadline = deleteDeadline
 		}
 	}
 	sv.mu.Unlock()
@@ -53,21 +67,66 @@ again:
 	}
 }
 
-func (as *minAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *minAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*minStateValue)
+
 		sv.mu.Lock()
-		min := sv.min
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "min", currentTimeMsec, min)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *minAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*minStateValue)
+		sv.mu.Lock()
+		m := sv.min
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, m)
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *minAggrState) getOutputName() string {
+	return "min"
+}
+
+func (as *minAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*minStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.min,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/newincrease.go

@@ -0,0 +1,176 @@
+package streamaggr
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// newincreaseAggrState calculates output=newincrease, e.g. the newincrease over input counters.
+type newincreaseAggrState struct {
+	m                   sync.Map
+	intervalSecs        uint64
+	ignoreInputDeadline uint64
+	stalenessSecs       uint64
+	lastPushTimestamp   uint64
+}
+
+type newincreaseStateValue struct {
+	mu             sync.Mutex
+	lastValues     map[string]*lastValueState
+	total          float64
+	samplesCount   uint64
+	deleteDeadline uint64
+	deleted        bool
+}
+
+func newnewincreaseAggrState(interval time.Duration, stalenessInterval time.Duration) *newincreaseAggrState {
+	currentTime := fasttime.UnixTimestamp()
+	intervalSecs := roundDurationToSecs(interval)
+	stalenessSecs := roundDurationToSecs(stalenessInterval)
+	return &newincreaseAggrState{
+		intervalSecs:        intervalSecs,
+		stalenessSecs:       stalenessSecs,
+		ignoreInputDeadline: currentTime + intervalSecs,
+	}
+}
+
+func (as *newincreaseAggrState) pushSample(inputKey, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &newincreaseStateValue{
+			lastValues: make(map[string]*lastValueState),
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if loaded {
+			// Use the entry created by a concurrent goroutine.
+			v = vNew
+		}
+	}
+	sv := v.(*newincreaseStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		lv, ok := sv.lastValues[inputKey]
+		if !ok {
+			lv = &lastValueState{}
+			lv.firstValue = value
+			lv.value = value
+			lv.correction = 0
+			sv.lastValues[inputKey] = lv
+		}
+
+		// process counter reset
+		delta := value - lv.value
+		if delta < 0 {
+			if (-delta * 8) < lv.value {
+				lv.correction += lv.value - value
+			} else {
+				lv.correction += lv.value
+			}
+		}
+
+		// process increasing counter
+		correctedValue := value + lv.correction
+		correctedDelta := correctedValue - lv.firstValue
+		if ok && math.Abs(correctedValue) < 10*(math.Abs(correctedDelta)+1) {
+			correctedDelta = correctedValue
+		}
+		if ok || currentTime > as.ignoreInputDeadline {
+			sv.total = correctedDelta
+		}
+		lv.value = value
+		lv.deleteDeadline = deleteDeadline
+		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *newincreaseAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newincreaseStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		} else {
+			// Delete outdated entries in sv.lastValues
+			m := sv.lastValues
+			for k1, v1 := range m {
+				if currentTime > v1.deleteDeadline {
+					delete(m, k1)
+				}
+			}
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *newincreaseAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newincreaseStateValue)
+		sv.mu.Lock()
+		newincrease := sv.total
+		sv.total = 0
+		deleted := sv.deleted
+		sv.mu.Unlock()
+		if !deleted {
+			key := k.(string)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, newincrease)
+		}
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *newincreaseAggrState) getOutputName() string {
+	return "newincrease"
+}
+
+func (as *newincreaseAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*newincreaseStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/newincreasepure.go

@@ -0,0 +1,176 @@
+package streamaggr
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// newincreasePureAggrState calculates output=newincrease, e.g. the newincrease over input counters.
+type newincreasePureAggrState struct {
+	m                   sync.Map
+	intervalSecs        uint64
+	ignoreInputDeadline uint64
+	stalenessSecs       uint64
+	lastPushTimestamp   uint64
+}
+
+type newincreasePureStateValue struct {
+	mu             sync.Mutex
+	lastValues     map[string]*lastValueState
+	total          float64
+	samplesCount   uint64
+	deleteDeadline uint64
+	deleted        bool
+}
+
+func newnewincreasePureAggrState(interval time.Duration, stalenessInterval time.Duration) *newincreasePureAggrState {
+	currentTime := fasttime.UnixTimestamp()
+	intervalSecs := roundDurationToSecs(interval)
+	stalenessSecs := roundDurationToSecs(stalenessInterval)
+	return &newincreasePureAggrState{
+		intervalSecs:        intervalSecs,
+		stalenessSecs:       stalenessSecs,
+		ignoreInputDeadline: currentTime + intervalSecs,
+	}
+}
+
+func (as *newincreasePureAggrState) pushSample(inputKey, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &newincreasePureStateValue{
+			lastValues: make(map[string]*lastValueState),
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if loaded {
+			// Use the entry created by a concurrent goroutine.
+			v = vNew
+		}
+	}
+	sv := v.(*newincreasePureStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		lv, ok := sv.lastValues[inputKey]
+		if !ok {
+			lv = &lastValueState{}
+			lv.firstValue = value
+			lv.value = value
+			lv.correction = 0
+			sv.lastValues[inputKey] = lv
+		}
+
+		// process counter reset
+		delta := value - lv.value
+		if delta < 0 {
+			if (-delta * 8) < lv.value {
+				lv.correction += lv.value - value
+			} else {
+				lv.correction += lv.value
+			}
+		}
+
+		// process increasing counter
+		correctedValue := value + lv.correction
+		correctedDelta := correctedValue - lv.firstValue
+		if ok && math.Abs(correctedValue) < 10*(math.Abs(correctedDelta)+1) {
+			correctedDelta = correctedValue
+		}
+		if ok || currentTime > as.ignoreInputDeadline {
+			sv.total = correctedDelta
+		}
+		lv.value = value
+		lv.deleteDeadline = deleteDeadline
+		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *newincreasePureAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newincreasePureStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		} else {
+			// Delete outdated entries in sv.lastValues
+			m := sv.lastValues
+			for k1, v1 := range m {
+				if currentTime > v1.deleteDeadline {
+					delete(m, k1)
+				}
+			}
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *newincreasePureAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newincreasePureStateValue)
+		sv.mu.Lock()
+		newincrease := sv.total
+		sv.total = 0
+		deleted := sv.deleted
+		sv.mu.Unlock()
+		if !deleted {
+			key := k.(string)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, newincrease)
+		}
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *newincreasePureAggrState) getOutputName() string {
+	return "newincrease_pure"
+}
+
+func (as *newincreasePureAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*newincreasePureStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/newtotal.go

@@ -0,0 +1,179 @@
+package streamaggr
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// newtotalAggrState calculates output=newtotal, e.g. the summary counter over input counters.
+type newtotalAggrState struct {
+	m                   sync.Map
+	intervalSecs        uint64
+	ignoreInputDeadline uint64
+	stalenessSecs       uint64
+	lastPushTimestamp   uint64
+}
+
+type newtotalStateValue struct {
+	mu             sync.Mutex
+	lastValues     map[string]*lastValueState
+	total          float64
+	samplesCount   uint64
+	deleteDeadline uint64
+	deleted        bool
+}
+
+func newnewtotalAggrState(interval time.Duration, stalenessInterval time.Duration) *newtotalAggrState {
+	currentTime := fasttime.UnixTimestamp()
+	intervalSecs := roundDurationToSecs(interval)
+	stalenessSecs := roundDurationToSecs(stalenessInterval)
+	return &newtotalAggrState{
+		intervalSecs:        intervalSecs,
+		stalenessSecs:       stalenessSecs,
+		ignoreInputDeadline: currentTime + intervalSecs,
+	}
+}
+
+func (as *newtotalAggrState) pushSample(inputKey, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &newtotalStateValue{
+			lastValues: make(map[string]*lastValueState),
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if loaded {
+			// Use the entry created by a concurrent goroutine.
+			v = vNew
+		}
+	}
+	sv := v.(*newtotalStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		lv, ok := sv.lastValues[inputKey]
+		if !ok {
+			lv = &lastValueState{}
+			lv.firstValue = value
+			lv.value = value
+			lv.correction = 0
+			sv.lastValues[inputKey] = lv
+		}
+
+		// process counter reset
+		delta := value - lv.value
+		if delta < 0 {
+			if (-delta * 8) < lv.value {
+				lv.correction += lv.value - value
+			} else {
+				lv.correction += lv.value
+			}
+		}
+
+		// process increasing counter
+		correctedValue := value + lv.correction
+		correctedDelta := correctedValue - lv.firstValue
+		if ok && math.Abs(correctedValue) < 10*(math.Abs(correctedDelta)+1) {
+			correctedDelta = correctedValue
+		}
+		if ok || currentTime > as.ignoreInputDeadline {
+			sv.total = correctedDelta
+		}
+		lv.value = value
+		lv.deleteDeadline = deleteDeadline
+		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *newtotalAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newtotalStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		} else {
+			// Delete outdated entries in sv.lastValues
+			m := sv.lastValues
+			for k1, v1 := range m {
+				if currentTime > v1.deleteDeadline {
+					delete(m, k1)
+				}
+			}
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *newtotalAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newtotalStateValue)
+		sv.mu.Lock()
+		total := sv.total
+		if math.Abs(sv.total) >= (1 << 53) {
+			// It is time to reset the entry, since it starts losing float64 precision
+			sv.total = 0
+		}
+		deleted := sv.deleted
+		sv.mu.Unlock()
+		if !deleted {
+			key := k.(string)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, total)
+		}
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *newtotalAggrState) getOutputName() string {
+	return "newtotal"
+}
+
+func (as *newtotalAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*newtotalStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/newtotalpure.go

@@ -0,0 +1,177 @@
+package streamaggr
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// newotalPureAggrState calculates output=newtotal, e.g. the summary counter over input counters.
+type newotalPureAggrState struct {
+	m                   sync.Map
+	intervalSecs        uint64
+	ignoreInputDeadline uint64
+	stalenessSecs       uint64
+	lastPushTimestamp   uint64
+}
+
+type newtotalPureStateValue struct {
+	mu             sync.Mutex
+	lastValues     map[string]*lastValueState
+	total          float64
+	samplesCount   uint64
+	deleteDeadline uint64
+	deleted        bool
+}
+
+func newnewotalPureAggrState(interval time.Duration, stalenessInterval time.Duration) *newotalPureAggrState {
+	currentTime := fasttime.UnixTimestamp()
+	intervalSecs := roundDurationToSecs(interval)
+	stalenessSecs := roundDurationToSecs(stalenessInterval)
+	return &newotalPureAggrState{
+		intervalSecs:        intervalSecs,
+		stalenessSecs:       stalenessSecs,
+		ignoreInputDeadline: currentTime + intervalSecs,
+	}
+}
+
+func (as *newotalPureAggrState) pushSample(inputKey, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &newtotalPureStateValue{
+			lastValues: make(map[string]*lastValueState),
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if loaded {
+			// Use the entry created by a concurrent goroutine.
+			v = vNew
+		}
+	}
+	sv := v.(*newtotalPureStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		lv, ok := sv.lastValues[inputKey]
+		if !ok {
+			lv = &lastValueState{}
+			lv.firstValue = value
+			lv.value = value
+			lv.correction = 0
+			sv.lastValues[inputKey] = lv
+		}
+
+		// process counter reset
+		delta := value - lv.value
+		if delta < 0 {
+			if (-delta * 8) < lv.value {
+				lv.correction += lv.value - value
+			} else {
+				lv.correction += lv.value
+			}
+		}
+
+		// process increasing counter
+		correctedValue := value + lv.correction
+		correctedDelta := correctedValue - lv.firstValue
+		if ok && math.Abs(correctedValue) < 10*(math.Abs(correctedDelta)+1) {
+			correctedDelta = correctedValue
+		}
+		sv.total = correctedDelta
+		lv.value = value
+		lv.deleteDeadline = deleteDeadline
+		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *newotalPureAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newtotalPureStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		} else {
+			// Delete outdated entries in sv.lastValues
+			m := sv.lastValues
+			for k1, v1 := range m {
+				if currentTime > v1.deleteDeadline {
+					delete(m, k1)
+				}
+			}
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *newotalPureAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*newtotalPureStateValue)
+		sv.mu.Lock()
+		total := sv.total
+		if math.Abs(sv.total) >= (1 << 53) {
+			// It is time to reset the entry, since it starts losing float64 precision
+			sv.total = 0
+		}
+		deleted := sv.deleted
+		sv.mu.Unlock()
+		if !deleted {
+			key := k.(string)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, total)
+		}
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *newotalPureAggrState) getOutputName() string {
+	return "newtotal_pure"
+}
+
+func (as *newotalPureAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*newtotalPureStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/quantiles.go

@@ -1,8 +1,10 @@
 package streamaggr
 
 import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
@@ -11,24 +13,33 @@ import (
 
 // quantilesAggrState calculates output=quantiles, e.g. the the given quantiles over the input samples.
 type quantilesAggrState struct {
-	m sync.Map
+	m                 sync.Map
-
+	phis              []float64
-	phis []float64
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type quantilesStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	h       *histogram.Fast
+	h              *histogram.Fast
-	deleted bool
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newQuantilesAggrState(phis []float64) *quantilesAggrState {
+func newQuantilesAggrState(interval time.Duration, stalenessInterval time.Duration, phis []float64) *quantilesAggrState {
 	return &quantilesAggrState{
-		phis: phis,
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+		phis:          phis,
 	}
 }
 
 func (as *quantilesAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -49,6 +60,8 @@ again:
 	deleted := sv.deleted
 	if !deleted {
 		sv.h.Update(value)
+		sv.samplesCount++
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -58,30 +71,83 @@ again:
 	}
 }
 
+func (as *quantilesAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*quantilesStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+			histogram.PutFast(sv.h)
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
 func (as *quantilesAggrState) appendSeriesForFlush(ctx *flushCtx) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
 	m := &as.m
 	phis := as.phis
 	var quantiles []float64
 	var b []byte
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*quantilesStateValue)
 		sv.mu.Lock()
 		quantiles = sv.h.Quantiles(quantiles[:0], phis)
-		histogram.PutFast(sv.h)
+		sv.h.Reset()
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-		sv.deleted = true
 		sv.mu.Unlock()
 
 		key := k.(string)
 		for i, quantile := range quantiles {
 			b = strconv.AppendFloat(b[:0], phis[i], 'g', -1, 64)
 			phiStr := bytesutil.InternBytes(b)
-			ctx.appendSeriesWithExtraLabel(key, "quantiles", currentTimeMsec, quantile, "quantile", phiStr)
+			ctx.appendSeriesWithExtraLabel(key, as.getOutputName(), currentTimeMsec, quantile, "quantile", phiStr)
 		}
 		return true
 	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *quantilesAggrState) getOutputName() string {
+	return "quantiles"
+}
+
+func (as *quantilesAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	var b []byte
+	as.m.Range(func(k, v any) bool {
+		value := v.(*quantilesStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		for i, quantile := range value.h.Quantiles(make([]float64, 0), as.phis) {
+			b = strconv.AppendFloat(b[:0], as.phis[i], 'g', -1, 64)
+			result = append(result, aggrStateRepresentation{
+				metric: getLabelsStringFromKey(k.(string), suffix, as.getOutputName(), prompbmarshal.Label{
+					Name:  "quantile",
+					Value: bytesutil.InternBytes(b),
+				}),
+				currentValue:      quantile,
+				lastPushTimestamp: as.lastPushTimestamp,
+				nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+				samplesCount:      value.samplesCount,
+			})
+		}
+		return true
+	})
+	return result
 }

lib/streamaggr/state.go

@@ -0,0 +1,74 @@
+package streamaggr
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+)
+
+// WriteHumanReadableState writes human-readable state for all aggregations.
+func WriteHumanReadableState(w http.ResponseWriter, r *http.Request, rws map[string]*Aggregators) {
+	rwActive := r.FormValue("rw")
+	if rwActive == "" {
+		for key, _ := range rws {
+			rwActive = key
+			break
+		}
+	}
+	rw, ok := rws[rwActive]
+	if !ok {
+		_, _ = fmt.Fprintf(w, "not found remoteWrite '%v'", rwActive)
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	aggParam := r.FormValue("agg")
+	if aggParam == "" {
+		WriteStreamAggHTML(w, rws, rwActive)
+		return
+	}
+
+	aggNum, err := strconv.Atoi(aggParam)
+	if err != nil {
+		_, _ = fmt.Fprintf(w, "incorrect parameter 'agg': %v", err)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	if aggNum >= len(rw.as) {
+		_, _ = fmt.Fprintf(w, "not found aggregation with num '%v'", aggNum)
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+	agg := rw.as[aggNum]
+
+	var as aggrState
+	output := r.FormValue("output")
+	for _, a := range agg.aggrStates {
+		if output == "" {
+			as = a
+			break
+		}
+		if a.getOutputName() == output {
+			as = a
+			break
+		}
+	}
+	if as == nil {
+		_, _ = fmt.Fprintf(w, "not found output '%v'", output)
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	limitNum := 1000
+	limitParam := r.FormValue("limit")
+	if limitParam != "" {
+		limitNum, err = strconv.Atoi(limitParam)
+		if err != nil {
+			_, _ = fmt.Fprintf(w, "incorrect parameter 'limit': %v", err)
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+	}
+
+	WriteStreamAggOutputStateHTML(w, rwActive, aggNum, agg, as, limitNum)
+}

lib/streamaggr/state.qtpl

@@ -0,0 +1,215 @@
+{% import (
+    "fmt"
+    "sort"
+    "strings"
+    "time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/htmlcomponents"
+) %}
+
+{% stripspace %}
+
+{% func StreamAggHTML(rws map[string]*Aggregators, rwActive string) %}
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    {%= htmlcomponents.CommonHeader() %}
+    <title>Stream aggregation</title>
+</head>
+<body>
+    {%= htmlcomponents.Navbar() %}
+    <div class="container-fluid">
+        <div class="row">
+            <main class="col-12">
+                <h1>Aggregations</h1>
+                <hr />
+                <ul class="nav nav-tabs" id="rw-tab" role="tablist">
+                    {% for rwKey, _ := range rws  %}
+                        <li class="nav-item" role="presentation">
+                            <button class="nav-link{%if rwKey==rwActive %}{% space %}active{%endif%}" type="button" role="tab"
+                              onclick="location.href='?rw={%s rwKey %}'">
+                                {%s rwKey %}
+                            </button>
+                        </li>
+                    {% endfor %}
+                </ul>
+                <div class="tab-content">
+                    <div class="tab-pane active" role="tabpanel">
+                        <div id="aggregations" class="table-responsive">
+                          <table class="table table-striped table-hover table-bordered table-sm">
+                              <thead>
+                                  <tr>
+                                      <th scope="col" style="width: 5%">Num</th>
+                                      <th scope="col" style="width: 35%">Match</th>
+                                      <th scope="col" style="width: 10%">By</th>
+                                      <th scope="col" style="width: 10%">Without</a>
+                                      <th scope="col" style="width: 40%">Outputs</a>
+                                  </tr>
+                              </thead>
+                              <tbody>
+                              {% code aggs := rws[rwActive] %}
+                              {% for an, agg := range aggs.as %}
+                                  <tr>
+                                      <td>{%d an %}</td>
+                                      <td>
+                                          <code>{%s agg.match.String() %}</code>
+                                      </td>
+                                      <td class="labels">
+                                          {% for abn, ab := range agg.by %}
+                                              {% if abn > 0 %}
+                                                <span>, </span>
+                                              {% endif %}
+                                              <span class="badge bg-secondary">
+                                                {%s ab %}
+                                              </span>
+                                          {% endfor %}
+                                      </td>
+                                      <td class="labels">
+                                          {% for awn, aw := range agg.without %}
+                                              {% if awn > 0 %}
+                                                <span>, </span>
+                                              {% endif %}
+                                              <span class="badge bg-secondary">
+                                                {%s aw %}
+                                              </span>
+                                          {% endfor %}
+                                      </td>
+                                      <td class="labels">
+                                          {% for asn, as := range agg.aggrStates %}
+                                              {% if asn > 0 %}
+                                                  <span>, </span>
+                                              {% endif %}
+                                              <a href="?rw={%s rwActive %}&agg={%d an %}&output={%s as.getOutputName() %}">
+                                                  {%s as.getOutputName() %}
+                                              </a>
+                                          {% endfor %}
+                                      </td>
+                                  </tr>
+                              {% endfor %}
+                              </tbody>
+                          </table>
+                        </div>
+                    </div>
+                </div>
+            </main>
+        </div>
+    </div>
+</body>
+</html>
+{% endfunc %}
+
+{% func StreamAggOutputStateHTML(rwActive string, aggNum int, agg *aggregator, as aggrState, limit int) %}
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    {%= htmlcomponents.CommonHeader() %}
+    <title>Stream aggregation</title>
+</head>
+<body>
+    {%= htmlcomponents.Navbar() %}
+    <div class="container-fluid">
+        <div class="row">
+            <main class="col-12">
+                <h1>Aggregation state</h1>
+                <h4> [ <a href="?rw={%s rwActive %}">back to aggregations</a> ] </h3>
+                <hr />
+                <h6>
+                    <div class="row">
+                        <div class="col-xxl-1">Remote write:</div>
+                        <code class="col w-100">{%s rwActive %}</code>
+                        <div class="w-100"></div>
+
+                        <div class="col-xxl-1">Aggregation num:</div>
+                        <code class="col w-100">{%d aggNum %}</code>
+                        <div class="w-100"></div>
+
+                        <div class="col-xxl-1">Match:</div>
+                        <code class="col w-100">{%s agg.match.String() %}</code>
+                        <div class="w-100"></div>
+
+                        {% if len(agg.by) > 0 %}
+                            <div class="col-xxl-1">By:</div>
+                            <code class="col w-100">{%s strings.Join(agg.by, ", ") %}</code>
+                            <div class="w-100"></div>
+                        {% endif %}
+                        {% if len(agg.without) > 0 %}
+                            <div class="col-xxl-1">Without:</div>
+                            <code class="col w-100">{%s strings.Join(agg.without, ", ") %}</code>
+                            <div class="w-100"></div>
+                        {% endif %}
+                    </div>
+                </h6>
+                <hr />
+                <ul class="nav nav-tabs" id="rw-tab" role="tablist">
+                    {% for _, a := range agg.aggrStates  %}
+                        <li class="nav-item" role="presentation">
+                            <button class="nav-link{%if a.getOutputName()==as.getOutputName() %}{% space %}active{%endif%}" type="button" role="tab"
+                              onclick="location.href='?rw={%s rwActive %}&agg={%d aggNum %}&output={%s a.getOutputName() %}'">
+                                {%s a.getOutputName() %}
+                            </button>
+                        </li>
+                    {% endfor %}
+                </ul>
+                <div class="tab-content">
+                    <div class="tab-pane active" role="tabpanel">
+                        <div id="aggregation-state" class="table-responsive">
+                            <table class="table table-striped table-hover table-bordered table-sm">
+                                {% code
+                                    sr := as.getStateRepresentation(agg.suffix)
+                                    sort.Slice(sr, func(i, j int) bool {
+                                        return sr[i].metric < sr[j].metric
+                                    })
+                                    if len(sr) > limit {
+                                        sr = sr[:limit]
+                                    }
+                                %}
+                                <thead>
+                                    <tr>
+                                        <th scope="col">Metric</th>
+                                        <th scope="col">Current value</th>
+                                        <th scope="col">Samples count</th>
+                                        <th scope="col">Last push time</th>
+                                        <th scope="col">Next push time</th>
+                                    </tr>
+                                </thead>
+                                <tbody>
+                                    {% for _, asr := range sr %}
+                                        <tr>
+                                            <td>
+                                                <code>{%s asr.metric %}</code>
+                                            </td>
+                                            <td class="text-end">
+                                                {%f asr.currentValue %}
+                                            </td>
+                                            <td class="text-end">
+                                                {%s fmt.Sprintf("%v", asr.samplesCount) %}
+                                            </td>
+                                            <td>
+                                                {% if asr.lastPushTimestamp == 0 %}
+                                                    {%s time.Unix(int64(agg.initialTime), 0).String() %}
+                                                {% else %}
+                                                    {%s time.Unix(int64(asr.lastPushTimestamp), 0).String() %}
+                                                {% endif %}
+                                            </td>
+                                            <td>
+                                                {% if asr.lastPushTimestamp == 0 %}
+                                                    {%s time.Unix(int64(asr.nextPushTimestamp + agg.initialTime), 0).Format(time.RFC3339) %}
+                                                {% else %}
+                                                    {%s time.Unix(int64(asr.nextPushTimestamp), 0).Format(time.RFC3339) %}
+                                                {% endif %}
+                                            </td>
+                                        </tr>
+                                    {% endfor %}
+                                </tbody>
+                            </table>
+                        </div>
+                    </div>
+                </div>
+            </main>
+        </div>
+    </div>
+</body>
+</html>
+{% endfunc %}
+
+{% endstripspace %}

lib/streamaggr/state.qtpl.go

@@ -0,0 +1,348 @@
+// Code generated by qtc from "state.qtpl". DO NOT EDIT.
+// See https://github.com/valyala/quicktemplate for details.
+
+//line lib/streamaggr/state.qtpl:1
+package streamaggr
+
+//line lib/streamaggr/state.qtpl:1
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/htmlcomponents"
+)
+
+//line lib/streamaggr/state.qtpl:12
+import (
+	qtio422016 "io"
+
+	qt422016 "github.com/valyala/quicktemplate"
+)
+
+//line lib/streamaggr/state.qtpl:12
+var (
+	_ = qtio422016.Copy
+	_ = qt422016.AcquireByteBuffer
+)
+
+//line lib/streamaggr/state.qtpl:12
+func StreamStreamAggHTML(qw422016 *qt422016.Writer, rws map[string]*Aggregators, rwActive string) {
+//line lib/streamaggr/state.qtpl:12
+	qw422016.N().S(`<!DOCTYPE html><html lang="en"><head>`)
+//line lib/streamaggr/state.qtpl:16
+	htmlcomponents.StreamCommonHeader(qw422016)
+//line lib/streamaggr/state.qtpl:16
+	qw422016.N().S(`<title>Stream aggregation</title></head><body>`)
+//line lib/streamaggr/state.qtpl:20
+	htmlcomponents.StreamNavbar(qw422016)
+//line lib/streamaggr/state.qtpl:20
+	qw422016.N().S(`<div class="container-fluid"><div class="row"><main class="col-12"><h1>Aggregations</h1><hr /><ul class="nav nav-tabs" id="rw-tab" role="tablist">`)
+//line lib/streamaggr/state.qtpl:27
+	for rwKey, _ := range rws {
+//line lib/streamaggr/state.qtpl:27
+		qw422016.N().S(`<li class="nav-item" role="presentation"><button class="nav-link`)
+//line lib/streamaggr/state.qtpl:29
+		if rwKey == rwActive {
+//line lib/streamaggr/state.qtpl:29
+			qw422016.N().S(` `)
+//line lib/streamaggr/state.qtpl:29
+			qw422016.N().S(`active`)
+//line lib/streamaggr/state.qtpl:29
+		}
+//line lib/streamaggr/state.qtpl:29
+		qw422016.N().S(`" type="button" role="tab"onclick="location.href='?rw=`)
+//line lib/streamaggr/state.qtpl:30
+		qw422016.E().S(rwKey)
+//line lib/streamaggr/state.qtpl:30
+		qw422016.N().S(`'">`)
+//line lib/streamaggr/state.qtpl:31
+		qw422016.E().S(rwKey)
+//line lib/streamaggr/state.qtpl:31
+		qw422016.N().S(`</button></li>`)
+//line lib/streamaggr/state.qtpl:34
+	}
+//line lib/streamaggr/state.qtpl:34
+	qw422016.N().S(`</ul><div class="tab-content"><div class="tab-pane active" role="tabpanel"><div id="aggregations" class="table-responsive"><table class="table table-striped table-hover table-bordered table-sm"><thead><tr><th scope="col" style="width: 5%">Num</th><th scope="col" style="width: 35%">Match</th><th scope="col" style="width: 10%">By</th><th scope="col" style="width: 10%">Without</a><th scope="col" style="width: 40%">Outputs</a></tr></thead><tbody>`)
+//line lib/streamaggr/state.qtpl:50
+	aggs := rws[rwActive]
+
+//line lib/streamaggr/state.qtpl:51
+	for an, agg := range aggs.as {
+//line lib/streamaggr/state.qtpl:51
+		qw422016.N().S(`<tr><td>`)
+//line lib/streamaggr/state.qtpl:53
+		qw422016.N().D(an)
+//line lib/streamaggr/state.qtpl:53
+		qw422016.N().S(`</td><td><code>`)
+//line lib/streamaggr/state.qtpl:55
+		qw422016.E().S(agg.match.String())
+//line lib/streamaggr/state.qtpl:55
+		qw422016.N().S(`</code></td><td class="labels">`)
+//line lib/streamaggr/state.qtpl:58
+		for abn, ab := range agg.by {
+//line lib/streamaggr/state.qtpl:59
+			if abn > 0 {
+//line lib/streamaggr/state.qtpl:59
+				qw422016.N().S(`<span>, </span>`)
+//line lib/streamaggr/state.qtpl:61
+			}
+//line lib/streamaggr/state.qtpl:61
+			qw422016.N().S(`<span class="badge bg-secondary">`)
+//line lib/streamaggr/state.qtpl:63
+			qw422016.E().S(ab)
+//line lib/streamaggr/state.qtpl:63
+			qw422016.N().S(`</span>`)
+//line lib/streamaggr/state.qtpl:65
+		}
+//line lib/streamaggr/state.qtpl:65
+		qw422016.N().S(`</td><td class="labels">`)
+//line lib/streamaggr/state.qtpl:68
+		for awn, aw := range agg.without {
+//line lib/streamaggr/state.qtpl:69
+			if awn > 0 {
+//line lib/streamaggr/state.qtpl:69
+				qw422016.N().S(`<span>, </span>`)
+//line lib/streamaggr/state.qtpl:71
+			}
+//line lib/streamaggr/state.qtpl:71
+			qw422016.N().S(`<span class="badge bg-secondary">`)
+//line lib/streamaggr/state.qtpl:73
+			qw422016.E().S(aw)
+//line lib/streamaggr/state.qtpl:73
+			qw422016.N().S(`</span>`)
+//line lib/streamaggr/state.qtpl:75
+		}
+//line lib/streamaggr/state.qtpl:75
+		qw422016.N().S(`</td><td class="labels">`)
+//line lib/streamaggr/state.qtpl:78
+		for asn, as := range agg.aggrStates {
+//line lib/streamaggr/state.qtpl:79
+			if asn > 0 {
+//line lib/streamaggr/state.qtpl:79
+				qw422016.N().S(`<span>, </span>`)
+//line lib/streamaggr/state.qtpl:81
+			}
+//line lib/streamaggr/state.qtpl:81
+			qw422016.N().S(`<a href="?rw=`)
+//line lib/streamaggr/state.qtpl:82
+			qw422016.E().S(rwActive)
+//line lib/streamaggr/state.qtpl:82
+			qw422016.N().S(`&agg=`)
+//line lib/streamaggr/state.qtpl:82
+			qw422016.N().D(an)
+//line lib/streamaggr/state.qtpl:82
+			qw422016.N().S(`&output=`)
+//line lib/streamaggr/state.qtpl:82
+			qw422016.E().S(as.getOutputName())
+//line lib/streamaggr/state.qtpl:82
+			qw422016.N().S(`">`)
+//line lib/streamaggr/state.qtpl:83
+			qw422016.E().S(as.getOutputName())
+//line lib/streamaggr/state.qtpl:83
+			qw422016.N().S(`</a>`)
+//line lib/streamaggr/state.qtpl:85
+		}
+//line lib/streamaggr/state.qtpl:85
+		qw422016.N().S(`</td></tr>`)
+//line lib/streamaggr/state.qtpl:88
+	}
+//line lib/streamaggr/state.qtpl:88
+	qw422016.N().S(`</tbody></table></div></div></div></main></div></div></body></html>`)
+//line lib/streamaggr/state.qtpl:99
+}
+
+//line lib/streamaggr/state.qtpl:99
+func WriteStreamAggHTML(qq422016 qtio422016.Writer, rws map[string]*Aggregators, rwActive string) {
+//line lib/streamaggr/state.qtpl:99
+	qw422016 := qt422016.AcquireWriter(qq422016)
+//line lib/streamaggr/state.qtpl:99
+	StreamStreamAggHTML(qw422016, rws, rwActive)
+//line lib/streamaggr/state.qtpl:99
+	qt422016.ReleaseWriter(qw422016)
+//line lib/streamaggr/state.qtpl:99
+}
+
+//line lib/streamaggr/state.qtpl:99
+func StreamAggHTML(rws map[string]*Aggregators, rwActive string) string {
+//line lib/streamaggr/state.qtpl:99
+	qb422016 := qt422016.AcquireByteBuffer()
+//line lib/streamaggr/state.qtpl:99
+	WriteStreamAggHTML(qb422016, rws, rwActive)
+//line lib/streamaggr/state.qtpl:99
+	qs422016 := string(qb422016.B)
+//line lib/streamaggr/state.qtpl:99
+	qt422016.ReleaseByteBuffer(qb422016)
+//line lib/streamaggr/state.qtpl:99
+	return qs422016
+//line lib/streamaggr/state.qtpl:99
+}
+
+//line lib/streamaggr/state.qtpl:101
+func StreamStreamAggOutputStateHTML(qw422016 *qt422016.Writer, rwActive string, aggNum int, agg *aggregator, as aggrState, limit int) {
+//line lib/streamaggr/state.qtpl:101
+	qw422016.N().S(`<!DOCTYPE html><html lang="en"><head>`)
+//line lib/streamaggr/state.qtpl:105
+	htmlcomponents.StreamCommonHeader(qw422016)
+//line lib/streamaggr/state.qtpl:105
+	qw422016.N().S(`<title>Stream aggregation</title></head><body>`)
+//line lib/streamaggr/state.qtpl:109
+	htmlcomponents.StreamNavbar(qw422016)
+//line lib/streamaggr/state.qtpl:109
+	qw422016.N().S(`<div class="container-fluid"><div class="row"><main class="col-12"><h1>Aggregation state</h1><h4> [ <a href="?rw=`)
+//line lib/streamaggr/state.qtpl:114
+	qw422016.E().S(rwActive)
+//line lib/streamaggr/state.qtpl:114
+	qw422016.N().S(`">back to aggregations</a> ] </h3><hr /><h6><div class="row"><div class="col-xxl-1">Remote write:</div><code class="col w-100">`)
+//line lib/streamaggr/state.qtpl:119
+	qw422016.E().S(rwActive)
+//line lib/streamaggr/state.qtpl:119
+	qw422016.N().S(`</code><div class="w-100"></div><div class="col-xxl-1">Aggregation num:</div><code class="col w-100">`)
+//line lib/streamaggr/state.qtpl:123
+	qw422016.N().D(aggNum)
+//line lib/streamaggr/state.qtpl:123
+	qw422016.N().S(`</code><div class="w-100"></div><div class="col-xxl-1">Match:</div><code class="col w-100">`)
+//line lib/streamaggr/state.qtpl:127
+	qw422016.E().S(agg.match.String())
+//line lib/streamaggr/state.qtpl:127
+	qw422016.N().S(`</code><div class="w-100"></div>`)
+//line lib/streamaggr/state.qtpl:130
+	if len(agg.by) > 0 {
+//line lib/streamaggr/state.qtpl:130
+		qw422016.N().S(`<div class="col-xxl-1">By:</div><code class="col w-100">`)
+//line lib/streamaggr/state.qtpl:132
+		qw422016.E().S(strings.Join(agg.by, ", "))
+//line lib/streamaggr/state.qtpl:132
+		qw422016.N().S(`</code><div class="w-100"></div>`)
+//line lib/streamaggr/state.qtpl:134
+	}
+//line lib/streamaggr/state.qtpl:135
+	if len(agg.without) > 0 {
+//line lib/streamaggr/state.qtpl:135
+		qw422016.N().S(`<div class="col-xxl-1">Without:</div><code class="col w-100">`)
+//line lib/streamaggr/state.qtpl:137
+		qw422016.E().S(strings.Join(agg.without, ", "))
+//line lib/streamaggr/state.qtpl:137
+		qw422016.N().S(`</code><div class="w-100"></div>`)
+//line lib/streamaggr/state.qtpl:139
+	}
+//line lib/streamaggr/state.qtpl:139
+	qw422016.N().S(`</div></h6><hr /><ul class="nav nav-tabs" id="rw-tab" role="tablist">`)
+//line lib/streamaggr/state.qtpl:144
+	for _, a := range agg.aggrStates {
+//line lib/streamaggr/state.qtpl:144
+		qw422016.N().S(`<li class="nav-item" role="presentation"><button class="nav-link`)
+//line lib/streamaggr/state.qtpl:146
+		if a.getOutputName() == as.getOutputName() {
+//line lib/streamaggr/state.qtpl:146
+			qw422016.N().S(` `)
+//line lib/streamaggr/state.qtpl:146
+			qw422016.N().S(`active`)
+//line lib/streamaggr/state.qtpl:146
+		}
+//line lib/streamaggr/state.qtpl:146
+		qw422016.N().S(`" type="button" role="tab"onclick="location.href='?rw=`)
+//line lib/streamaggr/state.qtpl:147
+		qw422016.E().S(rwActive)
+//line lib/streamaggr/state.qtpl:147
+		qw422016.N().S(`&agg=`)
+//line lib/streamaggr/state.qtpl:147
+		qw422016.N().D(aggNum)
+//line lib/streamaggr/state.qtpl:147
+		qw422016.N().S(`&output=`)
+//line lib/streamaggr/state.qtpl:147
+		qw422016.E().S(a.getOutputName())
+//line lib/streamaggr/state.qtpl:147
+		qw422016.N().S(`'">`)
+//line lib/streamaggr/state.qtpl:148
+		qw422016.E().S(a.getOutputName())
+//line lib/streamaggr/state.qtpl:148
+		qw422016.N().S(`</button></li>`)
+//line lib/streamaggr/state.qtpl:151
+	}
+//line lib/streamaggr/state.qtpl:151
+	qw422016.N().S(`</ul><div class="tab-content"><div class="tab-pane active" role="tabpanel"><div id="aggregation-state" class="table-responsive"><table class="table table-striped table-hover table-bordered table-sm">`)
+//line lib/streamaggr/state.qtpl:158
+	sr := as.getStateRepresentation(agg.suffix)
+	sort.Slice(sr, func(i, j int) bool {
+		return sr[i].metric < sr[j].metric
+	})
+	if len(sr) > limit {
+		sr = sr[:limit]
+	}
+
+//line lib/streamaggr/state.qtpl:165
+	qw422016.N().S(`<thead><tr><th scope="col">Metric</th><th scope="col">Current value</th><th scope="col">Samples count</th><th scope="col">Last push time</th><th scope="col">Next push time</th></tr></thead><tbody>`)
+//line lib/streamaggr/state.qtpl:176
+	for _, asr := range sr {
+//line lib/streamaggr/state.qtpl:176
+		qw422016.N().S(`<tr><td><code>`)
+//line lib/streamaggr/state.qtpl:179
+		qw422016.E().S(asr.metric)
+//line lib/streamaggr/state.qtpl:179
+		qw422016.N().S(`</code></td><td class="text-end">`)
+//line lib/streamaggr/state.qtpl:182
+		qw422016.N().F(asr.currentValue)
+//line lib/streamaggr/state.qtpl:182
+		qw422016.N().S(`</td><td class="text-end">`)
+//line lib/streamaggr/state.qtpl:185
+		qw422016.E().S(fmt.Sprintf("%v", asr.samplesCount))
+//line lib/streamaggr/state.qtpl:185
+		qw422016.N().S(`</td><td>`)
+//line lib/streamaggr/state.qtpl:188
+		if asr.lastPushTimestamp == 0 {
+//line lib/streamaggr/state.qtpl:189
+			qw422016.E().S(time.Unix(int64(agg.initialTime), 0).String())
+//line lib/streamaggr/state.qtpl:190
+		} else {
+//line lib/streamaggr/state.qtpl:191
+			qw422016.E().S(time.Unix(int64(asr.lastPushTimestamp), 0).String())
+//line lib/streamaggr/state.qtpl:192
+		}
+//line lib/streamaggr/state.qtpl:192
+		qw422016.N().S(`</td><td>`)
+//line lib/streamaggr/state.qtpl:195
+		if asr.lastPushTimestamp == 0 {
+//line lib/streamaggr/state.qtpl:196
+			qw422016.E().S(time.Unix(int64(asr.nextPushTimestamp+agg.initialTime), 0).Format(time.RFC3339))
+//line lib/streamaggr/state.qtpl:197
+		} else {
+//line lib/streamaggr/state.qtpl:198
+			qw422016.E().S(time.Unix(int64(asr.nextPushTimestamp), 0).Format(time.RFC3339))
+//line lib/streamaggr/state.qtpl:199
+		}
+//line lib/streamaggr/state.qtpl:199
+		qw422016.N().S(`</td></tr>`)
+//line lib/streamaggr/state.qtpl:202
+	}
+//line lib/streamaggr/state.qtpl:202
+	qw422016.N().S(`</tbody></table></div></div></div></main></div></div></body></html>`)
+//line lib/streamaggr/state.qtpl:213
+}
+
+//line lib/streamaggr/state.qtpl:213
+func WriteStreamAggOutputStateHTML(qq422016 qtio422016.Writer, rwActive string, aggNum int, agg *aggregator, as aggrState, limit int) {
+//line lib/streamaggr/state.qtpl:213
+	qw422016 := qt422016.AcquireWriter(qq422016)
+//line lib/streamaggr/state.qtpl:213
+	StreamStreamAggOutputStateHTML(qw422016, rwActive, aggNum, agg, as, limit)
+//line lib/streamaggr/state.qtpl:213
+	qt422016.ReleaseWriter(qw422016)
+//line lib/streamaggr/state.qtpl:213
+}
+
+//line lib/streamaggr/state.qtpl:213
+func StreamAggOutputStateHTML(rwActive string, aggNum int, agg *aggregator, as aggrState, limit int) string {
+//line lib/streamaggr/state.qtpl:213
+	qb422016 := qt422016.AcquireByteBuffer()
+//line lib/streamaggr/state.qtpl:213
+	WriteStreamAggOutputStateHTML(qb422016, rwActive, aggNum, agg, as, limit)
+//line lib/streamaggr/state.qtpl:213
+	qs422016 := string(qb422016.B)
+//line lib/streamaggr/state.qtpl:213
+	qt422016.ReleaseByteBuffer(qb422016)
+//line lib/streamaggr/state.qtpl:213
+	return qs422016
+//line lib/streamaggr/state.qtpl:213
+}

lib/streamaggr/stddev.go

@@ -3,28 +3,39 @@ package streamaggr
 import (
 	"math"
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // stddevAggrState calculates output=stddev, e.g. the average value over input samples.
 type stddevAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type stddevStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	count   float64
+	count          float64
-	avg     float64
+	avg            float64
-	q       float64
+	q              float64
-	deleted bool
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newStddevAggrState() *stddevAggrState {
+func newStddevAggrState(interval time.Duration, stalenessInterval time.Duration) *stddevAggrState {
-	return &stddevAggrState{}
+	return &stddevAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *stddevAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -45,6 +56,7 @@ again:
 		avg := sv.avg + (value-sv.avg)/sv.count
 		sv.q += (value - sv.avg) * (value - avg)
 		sv.avg = avg
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -54,21 +66,73 @@ again:
 	}
 }
 
-func (as *stddevAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *stddevAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*stddevStateValue)
+
 		sv.mu.Lock()
-		stddev := math.Sqrt(sv.q / sv.count)
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "stddev", currentTimeMsec, stddev)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *stddevAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*stddevStateValue)
+		sv.mu.Lock()
+		var stddev float64
+		if sv.count > 0 {
+			stddev = math.Sqrt(sv.q / sv.count)
+		}
+		sv.count = 0
+		sv.q = 0
+		sv.avg = 0
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, stddev)
+		return true
+	})
+
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *stddevAggrState) getOutputName() string {
+	return "stddev"
+}
+
+func (as *stddevAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*stddevStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      math.Sqrt(value.q / value.count),
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      uint64(value.count),
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/stdvar.go

@@ -2,28 +2,39 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // stdvarAggrState calculates output=stdvar, e.g. the average value over input samples.
 type stdvarAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type stdvarStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	count   float64
+	count          float64
-	avg     float64
+	avg            float64
-	q       float64
+	q              float64
-	deleted bool
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newStdvarAggrState() *stdvarAggrState {
+func newStdvarAggrState(interval time.Duration, stalenessInterval time.Duration) *stdvarAggrState {
-	return &stdvarAggrState{}
+	return &stdvarAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *stdvarAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -44,6 +55,7 @@ again:
 		avg := sv.avg + (value-sv.avg)/sv.count
 		sv.q += (value - sv.avg) * (value - avg)
 		sv.avg = avg
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -53,21 +65,72 @@ again:
 	}
 }
 
-func (as *stdvarAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *stdvarAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*stdvarStateValue)
+
 		sv.mu.Lock()
-		stdvar := sv.q / sv.count
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "stdvar", currentTimeMsec, stdvar)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *stdvarAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*stdvarStateValue)
+		sv.mu.Lock()
+		var stdvar float64
+		if sv.count > 0 {
+			stdvar = sv.q / sv.count
+		}
+		sv.q = 0
+		sv.avg = 0
+		sv.count = 0
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, stdvar)
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *stdvarAggrState) getOutputName() string {
+	return "stdvar"
+}
+
+func (as *stdvarAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*stdvarStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.q / value.count,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      uint64(value.count),
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/streamaggr.go

@@ -3,6 +3,7 @@ package streamaggr
 import (
 	"encoding/json"
 	"fmt"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"math"
 	"sort"
 	"strconv"
@@ -23,10 +24,17 @@ import (
 
 var supportedOutputs = []string{
 	"total",
+	"total_pure",
+	"newtotal",
+	"newtotal_pure",
 	"increase",
+	"increase_pure",
+	"newincrease",
+	"newincrease_pure",
 	"count_series",
 	"count_samples",
 	"sum_samples",
+	"sum_samples_total",
 	"last",
 	"min",
 	"max",
@@ -239,13 +247,24 @@ type aggregator struct {
 	// for `interval: 1m`, `by: [job]`
 	suffix string
 
-	wg     sync.WaitGroup
+	initialTime uint64
-	stopCh chan struct{}
+	wg          sync.WaitGroup
+	stopCh      chan struct{}
 }
 
 type aggrState interface {
 	pushSample(inputKey, outputKey string, value float64)
 	appendSeriesForFlush(ctx *flushCtx)
+	getOutputName() string
+	getStateRepresentation(suffix string) []aggrStateRepresentation
+}
+
+type aggrStateRepresentation struct {
+	metric            string
+	lastPushTimestamp uint64
+	nextPushTimestamp uint64
+	currentValue      float64
+	samplesCount      uint64
 }
 
 // PushFunc is called by Aggregators when it needs to push its state to metrics storage
@@ -263,9 +282,9 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse `interval: %q`: %w", cfg.Interval, err)
 	}
-	if interval <= time.Second {
+	//if interval <= time.Second {
-		return nil, fmt.Errorf("the minimum supported aggregation interval is 1s; got %s", interval)
+	//	return nil, fmt.Errorf("the minimum supported aggregation interval is 1s; got %s", interval)
-	}
+	//}
 
 	// check cfg.StalenessInterval
 	stalenessInterval := interval * 2
@@ -328,34 +347,48 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 				}
 				phis[j] = phi
 			}
-			aggrStates[i] = newQuantilesAggrState(phis)
+			aggrStates[i] = newQuantilesAggrState(interval, stalenessInterval, phis)
 			continue
 		}
 		switch output {
 		case "total":
 			aggrStates[i] = newTotalAggrState(interval, stalenessInterval)
+		case "total_pure":
+			aggrStates[i] = newTotalPureAggrState(interval, stalenessInterval)
+		case "newtotal":
+			aggrStates[i] = newnewtotalAggrState(interval, stalenessInterval)
+		case "newtotal_pure":
+			aggrStates[i] = newnewotalPureAggrState(interval, stalenessInterval)
 		case "increase":
 			aggrStates[i] = newIncreaseAggrState(interval, stalenessInterval)
+		case "increase_pure":
+			aggrStates[i] = newIncreasePureAggrState(interval, stalenessInterval)
+		case "newincrease":
+			aggrStates[i] = newnewincreaseAggrState(interval, stalenessInterval)
+		case "newincrease_pure":
+			aggrStates[i] = newnewincreasePureAggrState(interval, stalenessInterval)
 		case "count_series":
-			aggrStates[i] = newCountSeriesAggrState()
+			aggrStates[i] = newCountSeriesAggrState(interval, stalenessInterval)
 		case "count_samples":
-			aggrStates[i] = newCountSamplesAggrState()
+			aggrStates[i] = newCountSamplesAggrState(interval, stalenessInterval)
 		case "sum_samples":
-			aggrStates[i] = newSumSamplesAggrState()
+			aggrStates[i] = newSumSamplesAggrState(interval, stalenessInterval)
+		case "sum_samples_total":
+			aggrStates[i] = newSumSamplesTotalAggrState(interval, stalenessInterval)
 		case "last":
-			aggrStates[i] = newLastAggrState()
+			aggrStates[i] = newLastAggrState(interval, stalenessInterval)
 		case "min":
-			aggrStates[i] = newMinAggrState()
+			aggrStates[i] = newMinAggrState(interval, stalenessInterval)
 		case "max":
-			aggrStates[i] = newMaxAggrState()
+			aggrStates[i] = newMaxAggrState(interval, stalenessInterval)
 		case "avg":
-			aggrStates[i] = newAvgAggrState()
+			aggrStates[i] = newAvgAggrState(interval, stalenessInterval)
 		case "stddev":
-			aggrStates[i] = newStddevAggrState()
+			aggrStates[i] = newStddevAggrState(interval, stalenessInterval)
 		case "stdvar":
-			aggrStates[i] = newStdvarAggrState()
+			aggrStates[i] = newStdvarAggrState(interval, stalenessInterval)
 		case "histogram_bucket":
-			aggrStates[i] = newHistogramBucketAggrState(stalenessInterval)
+			aggrStates[i] = newHistogramBucketAggrState(interval, stalenessInterval)
 		default:
 			return nil, fmt.Errorf("unsupported output=%q; supported values: %s; "+
 				"see https://docs.victoriametrics.com/vmagent.html#stream-aggregation", output, supportedOutputs)
@@ -374,7 +407,7 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 
 	var dedupAggr *lastAggrState
 	if dedupInterval > 0 {
-		dedupAggr = newLastAggrState()
+		dedupAggr = newLastAggrState(interval, stalenessInterval)
 	}
 
 	// initialize the aggregator
@@ -394,7 +427,8 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 
 		suffix: suffix,
 
-		stopCh: make(chan struct{}),
+		initialTime: fasttime.UnixTimestamp(),
+		stopCh:      make(chan struct{}),
 	}
 
 	if dedupAggr != nil {
@@ -821,3 +855,21 @@ func sortAndRemoveDuplicates(a []string) []string {
 	}
 	return dst
 }
+
+func getLabelsStringFromKey(
+	key string,
+	suffix string,
+	output string,
+	extraLabels ...prompbmarshal.Label,
+) string {
+	labels := make([]prompbmarshal.Label, 0)
+	labels, _ = unmarshalLabelsFast(labels, []byte(key))
+	labels = addMetricSuffix(labels, 0, suffix, output)
+	labels = append(labels, extraLabels...)
+	a := make([]string, len(labels))
+	for i, label := range labels {
+		a[i] = fmt.Sprintf("%s=%q", label.Name, label.Value)
+	}
+	sort.Strings(a)
+	return "{" + strings.Join(a, ",") + "}"
+}

lib/streamaggr/streamaggr_test.go

@@ -2,6 +2,7 @@ package streamaggr
 
 import (
 	"fmt"
+	"reflect"
 	"sort"
 	"strconv"
 	"strings"
@@ -814,3 +815,301 @@ func mustParsePromMetrics(s string) []prompbmarshal.TimeSeries {
 	}
 	return tss
 }
+
+func TestTotalOutput(t *testing.T) {
+	saConfig := `
+- interval: 100ms
+  outputs:
+   - total
+   - increase
+   - newtotal
+   - newincrease
+`
+	type sample struct {
+		metric    string
+		value     float64
+		timestamp int64
+	}
+
+	f := func(config string, inputMetrics, outputMetricsExpected []sample, matchIdxsStrExpected string) {
+		t.Helper()
+
+		// Initialize Aggregators
+		var tssOutput []prompbmarshal.TimeSeries
+		var tssOutputLock sync.Mutex
+		pushFunc := func(tss []prompbmarshal.TimeSeries) {
+			tssOutputLock.Lock()
+			for _, ts := range tss {
+				labelsCopy := append([]prompbmarshal.Label{}, ts.Labels...)
+				samplesCopy := append([]prompbmarshal.Sample{}, ts.Samples...)
+				tssOutput = append(tssOutput, prompbmarshal.TimeSeries{
+					Labels:  labelsCopy,
+					Samples: samplesCopy,
+				})
+			}
+			tssOutputLock.Unlock()
+		}
+
+		a, err := NewAggregatorsFromData([]byte(config), pushFunc, 0)
+		if err != nil {
+			t.Fatalf("cannot initialize aggregators: %s", err)
+		}
+
+		// Push the inputMetrics to Aggregators
+		matchIdxs := make([]byte, 0)
+		var prevTs int64 = 0
+		for _, m := range inputMetrics {
+			if (m.timestamp - prevTs) > 0 {
+				<-time.After(time.Duration(m.timestamp-prevTs) * time.Microsecond)
+			}
+			inputMetricsStr := fmt.Sprintf("%s %v\n", m.metric, m.value)
+			tssInput := mustParsePromMetrics(inputMetricsStr)
+			matchIdxs = append(matchIdxs, a.Push(tssInput, nil)...)
+			prevTs = m.timestamp
+		}
+
+		a.MustStop()
+
+		//// Verify matchIdxs equals to matchIdxsExpected
+		matchIdxsStr := ""
+		for _, v := range matchIdxs {
+			matchIdxsStr += strconv.Itoa(int(v))
+		}
+		if matchIdxsStr != matchIdxsStrExpected {
+			t.Fatalf("unexpected matchIdxs;\ngot\n%s\nwant\n%s", matchIdxsStr, matchIdxsStrExpected)
+		}
+
+		// Verify the tssOutput contains the expected metrics
+		outputMetrics := make([]sample, len(tssOutput))
+		for i, ts := range tssOutput {
+			outputMetrics[i] = sample{
+				metric: promrelabel.LabelsToString(ts.Labels),
+				value:  ts.Samples[0].Value,
+			}
+		}
+
+		if !reflect.DeepEqual(outputMetrics, outputMetricsExpected) {
+			t.Fatalf("unexpected output metrics;\ngot\n%+v\nwant\n%+v", outputMetrics, outputMetricsExpected)
+		}
+	}
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 0},
+			{metric: "foo:100ms_increase", value: 0},
+			{metric: "foo:100ms_newtotal", value: 0},
+			{metric: "foo:100ms_newincrease", value: 0},
+		},
+		"1",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 100, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 0},
+			{metric: "foo:100ms_increase", value: 0},
+			{metric: "foo:100ms_newtotal", value: 0},    // right: 0
+			{metric: "foo:100ms_newincrease", value: 0}, // right: 0
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 90, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 90},
+			{metric: "foo:100ms_increase", value: 90},
+			{metric: "foo:100ms_newtotal", value: 0},    // right: 0
+			{metric: "foo:100ms_newincrease", value: 0}, // right: 0
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 87, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 87},
+			{metric: "foo:100ms_increase", value: 87},
+			{metric: "foo:100ms_newtotal", value: 187},    // right: 187
+			{metric: "foo:100ms_newincrease", value: 187}, // right: 187
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 187, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 87},
+			{metric: "foo:100ms_increase", value: 87},
+			{metric: "foo:100ms_newtotal", value: 187},    // right: 187
+			{metric: "foo:100ms_newincrease", value: 187}, // right: 187
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 13, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 13},
+			{metric: "foo:100ms_increase", value: 13},
+			{metric: "foo:100ms_newtotal", value: 113},    // right: 113
+			{metric: "foo:100ms_newincrease", value: 113}, // right: 113
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 9, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 9},
+			{metric: "foo:100ms_increase", value: 9},
+			{metric: "foo:100ms_newtotal", value: 9},    // right: 9
+			{metric: "foo:100ms_newincrease", value: 9}, // right: 9
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 1, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 1},
+			{metric: "foo:100ms_increase", value: 1},
+			{metric: "foo:100ms_newtotal", value: 1},    // right: 1
+			{metric: "foo:100ms_newincrease", value: 1}, // right: 1
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 0, timestamp: 1},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 0},
+			{metric: "foo:100ms_increase", value: 0},
+			{metric: "foo:100ms_newtotal", value: 0},    // right: 0
+			{metric: "foo:100ms_newincrease", value: 0}, // right: 0
+		},
+		"11",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 87, timestamp: 1},
+			{metric: "foo", value: 200, timestamp: 2},
+			{metric: "foo", value: 287, timestamp: 3},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 287},
+			{metric: "foo:100ms_increase", value: 287},
+			{metric: "foo:100ms_newtotal", value: 387},    // right: 387
+			{metric: "foo:100ms_newincrease", value: 387}, // right: 387
+		},
+		"1111",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 187, timestamp: 1},
+			{metric: "foo", value: 200, timestamp: 2},
+			{metric: "foo", value: 287, timestamp: 3},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 187},
+			{metric: "foo:100ms_increase", value: 187},
+			{metric: "foo:100ms_newtotal", value: 287},    // right: 287
+			{metric: "foo:100ms_newincrease", value: 287}, // right: 287
+		},
+		"1111",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 87, timestamp: 1},
+			{metric: "foo", value: 200, timestamp: 2},
+			{metric: "foo", value: 87, timestamp: 3},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 287},
+			{metric: "foo:100ms_increase", value: 287},
+			{metric: "foo:100ms_newtotal", value: 387},    // right: 387
+			{metric: "foo:100ms_newincrease", value: 387}, // right: 387
+		},
+		"1111",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 87, timestamp: 1},
+			{metric: "foo", value: 200, timestamp: 2},
+			{metric: "foo", value: 187, timestamp: 3},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 387},
+			{metric: "foo:100ms_increase", value: 387},
+			{metric: "foo:100ms_newtotal", value: 300},    // right: 300
+			{metric: "foo:100ms_newincrease", value: 300}, // right: 300
+		},
+		"1111",
+	)
+
+	f(
+		saConfig,
+		[]sample{
+			{metric: "foo", value: 100, timestamp: 0},
+			{metric: "foo", value: 87, timestamp: 1},
+			{metric: "foo", value: 200, timestamp: 2},
+			{metric: "foo", value: 177, timestamp: 3},
+		},
+		[]sample{
+			{metric: "foo:100ms_total", value: 377},
+			{metric: "foo:100ms_increase", value: 377},
+			{metric: "foo:100ms_newtotal", value: 300},    // right: 300
+			{metric: "foo:100ms_newincrease", value: 300}, // right: 300
+		},
+		"1111",
+	)
+
+}

lib/streamaggr/sum_samples.go

@@ -2,26 +2,38 @@ package streamaggr
 
 import (
 	"sync"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // sumSamplesAggrState calculates output=sum_samples, e.g. the sum over input samples.
 type sumSamplesAggrState struct {
-	m sync.Map
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
 }
 
 type sumSamplesStateValue struct {
-	mu      sync.Mutex
+	mu             sync.Mutex
-	sum     float64
+	sum            float64
-	deleted bool
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
 }
 
-func newSumSamplesAggrState() *sumSamplesAggrState {
+func newSumSamplesAggrState(interval time.Duration, stalenessInterval time.Duration) *sumSamplesAggrState {
-	return &sumSamplesAggrState{}
+	return &sumSamplesAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
 }
 
 func (as *sumSamplesAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
 again:
 	v, ok := as.m.Load(outputKey)
 	if !ok {
@@ -42,6 +54,8 @@ again:
 	deleted := sv.deleted
 	if !deleted {
 		sv.sum += value
+		sv.samplesCount++
+		sv.deleteDeadline = deleteDeadline
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -51,21 +65,67 @@ again:
 	}
 }
 
-func (as *sumSamplesAggrState) appendSeriesForFlush(ctx *flushCtx) {
+func (as *sumSamplesAggrState) removeOldEntries(currentTime uint64) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
-		// Atomically delete the entry from the map, so new entry is created for the next flush.
-		m.Delete(k)
-
 		sv := v.(*sumSamplesStateValue)
+
 		sv.mu.Lock()
-		sum := sv.sum
+		deleted := currentTime > sv.deleteDeadline
-		// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
+		if deleted {
-		sv.deleted = true
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
 		sv.mu.Unlock()
-		key := k.(string)
+
-		ctx.appendSeries(key, "sum_samples", currentTimeMsec, sum)
+		if deleted {
+			m.Delete(k)
+		}
 		return true
 	})
 }
+
+func (as *sumSamplesAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*sumSamplesStateValue)
+		sv.mu.Lock()
+		sum := sv.sum
+		sv.sum = 0
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, sum)
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *sumSamplesAggrState) getOutputName() string {
+	return "sum_samples"
+}
+
+func (as *sumSamplesAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*sumSamplesStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.sum,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/sum_samples_total.go

@@ -0,0 +1,135 @@
+package streamaggr
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// sumSamplesTotalAggrState calculates output=sum_samples, e.g. the sum over input samples.
+type sumSamplesTotalAggrState struct {
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
+}
+
+type sumSamplesTotalStateValue struct {
+	mu             sync.Mutex
+	sum            float64
+	samplesCount   uint64
+	deleted        bool
+	deleteDeadline uint64
+}
+
+func newSumSamplesTotalAggrState(interval time.Duration, stalenessInterval time.Duration) *sumSamplesTotalAggrState {
+	return &sumSamplesTotalAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
+}
+
+func (as *sumSamplesTotalAggrState) pushSample(_, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &sumSamplesTotalStateValue{
+			sum: value,
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if !loaded {
+			// The new entry has been successfully created.
+			return
+		}
+		// Use the entry created by a concurrent goroutine.
+		v = vNew
+	}
+	sv := v.(*sumSamplesTotalStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		sv.sum += value
+		sv.samplesCount++
+		sv.deleteDeadline = deleteDeadline
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *sumSamplesTotalAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*sumSamplesTotalStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *sumSamplesTotalAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*sumSamplesTotalStateValue)
+		sv.mu.Lock()
+		sum := sv.sum
+		if math.Abs(sv.sum) >= (1 << 53) {
+			// It is time to reset the entry, since it starts losing float64 precision
+			sv.sum = 0
+		}
+		sv.mu.Unlock()
+		key := k.(string)
+		ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, sum)
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *sumSamplesTotalAggrState) getOutputName() string {
+	return "sum_samples_total"
+}
+
+func (as *sumSamplesTotalAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*sumSamplesTotalStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.sum,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}

lib/streamaggr/total.go

@@ -10,22 +10,26 @@ import (
 
 // totalAggrState calculates output=total, e.g. the summary counter over input counters.
 type totalAggrState struct {
-	m sync.Map
+	m                   sync.Map
-
+	intervalSecs        uint64
 	ignoreInputDeadline uint64
 	stalenessSecs       uint64
+	lastPushTimestamp   uint64
 }
 
 type totalStateValue struct {
 	mu             sync.Mutex
 	lastValues     map[string]*lastValueState
 	total          float64
+	samplesCount   uint64
 	deleteDeadline uint64
 	deleted        bool
 }
 
 type lastValueState struct {
 	value          float64
+	firstValue     float64
+	correction     float64
 	deleteDeadline uint64
 }
 
@@ -34,8 +38,9 @@ func newTotalAggrState(interval time.Duration, stalenessInterval time.Duration)
 	intervalSecs := roundDurationToSecs(interval)
 	stalenessSecs := roundDurationToSecs(stalenessInterval)
 	return &totalAggrState{
-		ignoreInputDeadline: currentTime + intervalSecs,
+		intervalSecs:        intervalSecs,
 		stalenessSecs:       stalenessSecs,
+		ignoreInputDeadline: currentTime + intervalSecs,
 	}
 }
 
@@ -75,6 +80,7 @@ again:
 		lv.value = value
 		lv.deleteDeadline = deleteDeadline
 		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
 	}
 	sv.mu.Unlock()
 	if deleted {
@@ -131,8 +137,34 @@ func (as *totalAggrState) appendSeriesForFlush(ctx *flushCtx) {
 		sv.mu.Unlock()
 		if !deleted {
 			key := k.(string)
-			ctx.appendSeries(key, "total", currentTimeMsec, total)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, total)
 		}
 		return true
 	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *totalAggrState) getOutputName() string {
+	return "total"
+}
+
+func (as *totalAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*totalStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
 }

lib/streamaggr/totalpure.go

@@ -0,0 +1,156 @@
+package streamaggr
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+)
+
+// totalPureAggrState calculates output=total_pure, e.g. the summary counter over input counters.
+type totalPureAggrState struct {
+	m                 sync.Map
+	intervalSecs      uint64
+	stalenessSecs     uint64
+	lastPushTimestamp uint64
+}
+
+type totalPureStateValue struct {
+	mu             sync.Mutex
+	lastValues     map[string]*lastValueState
+	total          float64
+	samplesCount   uint64
+	deleteDeadline uint64
+	deleted        bool
+}
+
+func newTotalPureAggrState(interval time.Duration, stalenessInterval time.Duration) *totalPureAggrState {
+	return &totalPureAggrState{
+		intervalSecs:  roundDurationToSecs(interval),
+		stalenessSecs: roundDurationToSecs(stalenessInterval),
+	}
+}
+
+func (as *totalPureAggrState) pushSample(inputKey, outputKey string, value float64) {
+	currentTime := fasttime.UnixTimestamp()
+	deleteDeadline := currentTime + as.stalenessSecs
+
+again:
+	v, ok := as.m.Load(outputKey)
+	if !ok {
+		// The entry is missing in the map. Try creating it.
+		v = &totalPureStateValue{
+			lastValues: make(map[string]*lastValueState),
+		}
+		vNew, loaded := as.m.LoadOrStore(outputKey, v)
+		if loaded {
+			// Use the entry created by a concurrent goroutine.
+			v = vNew
+		}
+	}
+	sv := v.(*totalPureStateValue)
+	sv.mu.Lock()
+	deleted := sv.deleted
+	if !deleted {
+		lv, ok := sv.lastValues[inputKey]
+		if !ok {
+			lv = &lastValueState{}
+			sv.lastValues[inputKey] = lv
+		}
+		d := value
+		if ok && lv.value <= value {
+			d = value - lv.value
+		}
+		sv.total += d
+		lv.value = value
+		lv.deleteDeadline = deleteDeadline
+		sv.deleteDeadline = deleteDeadline
+		sv.samplesCount++
+	}
+	sv.mu.Unlock()
+	if deleted {
+		// The entry has been deleted by the concurrent call to appendSeriesForFlush
+		// Try obtaining and updating the entry again.
+		goto again
+	}
+}
+
+func (as *totalPureAggrState) removeOldEntries(currentTime uint64) {
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*totalPureStateValue)
+
+		sv.mu.Lock()
+		deleted := currentTime > sv.deleteDeadline
+		if deleted {
+			// Mark the current entry as deleted
+			sv.deleted = deleted
+		} else {
+			// Delete outdated entries in sv.lastValues
+			m := sv.lastValues
+			for k1, v1 := range m {
+				if currentTime > v1.deleteDeadline {
+					delete(m, k1)
+				}
+			}
+		}
+		sv.mu.Unlock()
+
+		if deleted {
+			m.Delete(k)
+		}
+		return true
+	})
+}
+
+func (as *totalPureAggrState) appendSeriesForFlush(ctx *flushCtx) {
+	currentTime := fasttime.UnixTimestamp()
+	currentTimeMsec := int64(currentTime) * 1000
+
+	as.removeOldEntries(currentTime)
+
+	m := &as.m
+	m.Range(func(k, v interface{}) bool {
+		sv := v.(*totalPureStateValue)
+		sv.mu.Lock()
+		totalPure := sv.total
+		if math.Abs(sv.total) >= (1 << 53) {
+			// It is time to reset the entry, since it starts losing float64 precision
+			sv.total = 0
+		}
+		deleted := sv.deleted
+		sv.mu.Unlock()
+		if !deleted {
+			key := k.(string)
+			ctx.appendSeries(key, as.getOutputName(), currentTimeMsec, totalPure)
+		}
+		return true
+	})
+	as.lastPushTimestamp = currentTime
+}
+
+func (as *totalPureAggrState) getOutputName() string {
+	return "total_pure"
+}
+
+func (as *totalPureAggrState) getStateRepresentation(suffix string) []aggrStateRepresentation {
+	result := make([]aggrStateRepresentation, 0)
+	as.m.Range(func(k, v any) bool {
+		value := v.(*totalPureStateValue)
+		value.mu.Lock()
+		defer value.mu.Unlock()
+		if value.deleted {
+			return true
+		}
+		result = append(result, aggrStateRepresentation{
+			metric:            getLabelsStringFromKey(k.(string), suffix, as.getOutputName()),
+			currentValue:      value.total,
+			lastPushTimestamp: as.lastPushTimestamp,
+			nextPushTimestamp: as.lastPushTimestamp + as.intervalSecs,
+			samplesCount:      value.samplesCount,
+		})
+		return true
+	})
+	return result
+}