Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7313 (2024-10-21 11:31:41)

feat: add IgnoreFirstSampleInterval configuration for streamaggr
### Describe Your Changes

As of right now by default aggregated output in streaming aggregation takes a staleness interval and only starts sending first samples after the staleness interval passes. We have a use case where we prefer to start sending data as soon as we have any. This adds the option to configure when we start sending first samples

https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7116

### Checklist

The following checks are **mandatory**:

- [x] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).


AndrewChubatiuk: can we reuse ignore_first_samples instead for this purpose?
iyuroch: @AndrewChubatiuk I've added the requested changes, I've also kept the same behavior as before them (equal to staleness interval). 
iyuroch: @AndrewChubatiuk @hagen1778 any chance you can review?
AndrewChubatiuk: `ignore first samples` already skips flushing in runFlusher func at `lib/streamaggr/streamaggr.go`. so we considered just removing this line https://github.com/VictoriaMetrics/VictoriaMetrics/blob/53b7288e0d5da1d53779fe7919c141f242ad5165/lib/streamaggr/total.go#L63 instead
iyuroch: @AndrewChubatiuk  I've checked noflusher and it seems this logic is redundant. I've added requested changes
iyuroch: @AndrewChubatiuk could you please share where to add changelog?
AndrewChubatiuk: https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/changelog/CHANGELOG.md
iyuroch: @AndrewChubatiuk I've added changelog
iyuroch: @AndrewChubatiuk I've pushed fixes to test cases. They change values as we didn't properly account for pushed values given the push skip deadline, given that I do have open questions about negative values in counters:
From my understanding this should not happen, but since there are already test cases they have the following flaws for <total>:
1. When the negative first value comes in we put this into lv and in total (where total becomes negative).
2. After the next increased value comes in we calculate the difference and sum this difference with total
-10
-5
becomes  -5
I propose we ignore negative values altogether as it's hard to interpret what they mean in the context of Prometheus counters and just go ahead with my proposed test changes, what do you think?
iyuroch: @AndrewChubatiuk any chance you will have time for review?
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/7313?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 60.64%. Comparing base [(`8aaa828`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/8aaa828ba3f49a43cb8de89a0bcae160f7988c14?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) to head [(`818342b`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/818342b87c589047ca6d7ef2c8d23f9d903f9063?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Report is 1830 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #7313      +/-   ##
==========================================
+ Coverage   60.37%   60.64%   +0.26%     
==========================================
  Files         411      644     +233     
  Lines       76609   108621   +32012     
==========================================
+ Hits        46253    65872   +19619     
- Misses      27794    39208   +11414     
- Partials     2562     3541     +979     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/7313?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).

----
🚨 Try these New Features:

- [Flaky Tests Detection](https://docs.codecov.com/docs/test-result-ingestion-beta) - Detect and resolve failed and flaky tests
- [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis) - Avoid shipping oversized bundles
iyuroch: @hagen1778 @AndrewChubatiuk ptal, resolved all comments
hagen1778: @AndrewChubatiuk should this change be ported to operator repo and helm charts?
AndrewChubatiuk: @hagen1778 to operator only

Patch:
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index 3383b7e96..4b53cc626 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -22,6 +22,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 * FEATURE: [vmalert](https://docs.victoriametrics.com/vmalert): revert the default value of `-remoteWrite.maxQueueSize` from `1_000_000` to `100_000`. It was bumped in [v1.104.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.104.0), which increases memory usage and is not needed for most setups. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7471).
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): add `Raw Query` tab for displaying raw data. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7024).
+* FEATURE: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): add `ignore_first_sample_interval` param to [aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config). It allows users to control the time interval when aggregation skips sending aggregated samples to avoid unexpected spikes in values. By default, this interval is set to x2 of `staleness_interval`. The new setting is applicable only to `total`, `total_prometheus`, `increase`, `increase_prometheus` and `histogram_bucket` outputs. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7116) for details. Thanks to @iyuroch for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7313).
 
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent): Properly return `200 OK` HTTP status code when importing data via [Pushgateway protocol](https://docs.victoriametrics.com/#how-to-import-data-in-prometheus-exposition-format) using [multitenant URL format](https://docs.victoriametrics.com/cluster-victoriametrics/#url-format). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3636) and [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7571).
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/), `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): properly return result for binary operation `^` aka pow at query requests for `NaN` values. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7359) for details.
diff --git a/docs/operator/api.md b/docs/operator/api.md
index 4b7e9c4d2..868c6f79d 100644
--- a/docs/operator/api.md
+++ b/docs/operator/api.md
@@ -2061,6 +2061,7 @@ _Appears in:_
 | `output_relabel_configs` | OutputRelabelConfigs is an optional relabeling rules, which are applied<br />on the aggregated output before being sent to remote storage. | _[RelabelConfig](#relabelconfig) array_ | false |
 | `outputs` | Outputs is a list of output aggregate functions to produce.<br /><br />The following names are allowed:<br /><br />- total - aggregates input counters<br />- increase - counts the increase over input counters<br />- count_series - counts the input series<br />- count_samples - counts the input samples<br />- sum_samples - sums the input samples<br />- last - the last biggest sample value<br />- min - the minimum sample value<br />- max - the maximum sample value<br />- avg - the average value across all the samples<br />- stddev - standard deviation across all the samples<br />- stdvar - standard variance across all the samples<br />- histogram_bucket - creates VictoriaMetrics histogram for input samples<br />- quantiles(phi1, ..., phiN) - quantiles' estimation for phi in the range [0..1]<br /><br />The output time series will have the following names:<br /><br />  input_name:aggr_<interval>_<output> | _string array_ | true |
 | `staleness_interval` | Staleness interval is interval after which the series state will be reset if no samples have been sent during it.<br />The parameter is only relevant for outputs: total, total_prometheus, increase, increase_prometheus and histogram_bucket. | _string_ | false |
+| `ignore_first_sample_interval` | IgnoreFirstSampleInterval specifies the interval after which the agent begins sending samples.<br />By default, it is set to the `staleness_interval`. It helps reducing the initial sample load after the agent restart.<br />This parameter is relevant only for the following outputs: total, total_prometheus, increase, increase_prometheus, and histogram_bucket. We recommend setting it to 0s unless you observe unexpected spikes in produced values. | _string_ | false |
 | `without` | Without is an optional list of labels, which must be excluded when grouping input series.<br /><br />See also By.<br /><br />If neither By nor Without are set, then the Outputs are calculated<br />individually per each input time series. | _string array_ | false |
 
 
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index 6188f8b80..f16f9a10a 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -174,6 +174,11 @@ type Config struct {
 	// The parameter is only relevant for outputs: total, total_prometheus, increase, increase_prometheus and histogram_bucket.
 	StalenessInterval string `yaml:"staleness_interval,omitempty"`
 
+	// IgnoreFirstSampleInterval specifies the interval after which the agent begins sending samples.
+	// By default, it is set to the staleness interval, and it helps reduce the initial sample load after an agent restart.
+	// This parameter is relevant only for the following outputs: total, total_prometheus, increase, increase_prometheus, and histogram_bucket.
+	IgnoreFirstSampleInterval string `yaml:"ignore_first_sample_interval,omitempty"`
+
 	// Outputs is a list of output aggregate functions to produce.
 	//
 	// The following names are allowed:
@@ -483,6 +488,16 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 		}
 	}
 
+	// check cfg.IgnoreFirstSampleInterval
+	// by default, it equals to the staleness interval to have backward compatibility, see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7116
+	ignoreFirstSampleInterval := stalenessInterval
+	if cfg.IgnoreFirstSampleInterval != "" {
+		ignoreFirstSampleInterval, err = time.ParseDuration(cfg.IgnoreFirstSampleInterval)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse `ignore_first_sample_interval: %q`: %w", cfg.IgnoreFirstSampleInterval, err)
+		}
+	}
+
 	// Check cfg.DropInputLabels
 	dropInputLabels := opts.DropInputLabels
 	if v := cfg.DropInputLabels; v != nil {
@@ -557,7 +572,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	aggrOutputs := make([]aggrOutput, len(cfg.Outputs))
 	outputsSeen := make(map[string]struct{}, len(cfg.Outputs))
 	for i, output := range cfg.Outputs {
-		as, err := newAggrState(output, outputsSeen, stalenessInterval)
+		as, err := newAggrState(output, outputsSeen, stalenessInterval, ignoreFirstSampleInterval)
 		if err != nil {
 			return nil, err
 		}
@@ -645,7 +660,7 @@ func newAggregator(cfg *Config, path string, pushFunc PushFunc, ms *metrics.Set,
 	return a, nil
 }
 
-func newAggrState(output string, outputsSeen map[string]struct{}, stalenessInterval time.Duration) (aggrState, error) {
+func newAggrState(output string, outputsSeen map[string]struct{}, stalenessInterval, ignoreFirstSampleInterval time.Duration) (aggrState, error) {
 	// check for duplicated output
 	if _, ok := outputsSeen[output]; ok {
 		return nil, fmt.Errorf("`outputs` list contains duplicate aggregation function: %s", output)
@@ -690,9 +705,9 @@ func newAggrState(output string, outputsSeen map[string]struct{}, stalenessInter
 	case "histogram_bucket":
 		return newHistogramBucketAggrState(stalenessInterval), nil
 	case "increase":
-		return newTotalAggrState(stalenessInterval, true, true), nil
+		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, true, true), nil
 	case "increase_prometheus":
-		return newTotalAggrState(stalenessInterval, true, false), nil
+		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, true, false), nil
 	case "last":
 		return newLastAggrState(), nil
 	case "max":
@@ -710,9 +725,9 @@ func newAggrState(output string, outputsSeen map[string]struct{}, stalenessInter
 	case "sum_samples":
 		return newSumSamplesAggrState(), nil
 	case "total":
-		return newTotalAggrState(stalenessInterval, false, true), nil
+		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, false, true), nil
 	case "total_prometheus":
-		return newTotalAggrState(stalenessInterval, false, false), nil
+		return newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval, false, false), nil
 	case "unique_samples":
 		return newUniqueSamplesAggrState(), nil
 	default:
diff --git a/lib/streamaggr/total.go b/lib/streamaggr/total.go
index e905e4531..53dbcdc47 100644
--- a/lib/streamaggr/total.go
+++ b/lib/streamaggr/total.go
@@ -45,9 +45,9 @@ type totalLastValueState struct {
 	deleteDeadline uint64
 }
 
-func newTotalAggrState(stalenessInterval time.Duration, resetTotalOnFlush, keepFirstSample bool) *totalAggrState {
+func newTotalAggrState(stalenessInterval, ignoreFirstSampleInterval time.Duration, resetTotalOnFlush, keepFirstSample bool) *totalAggrState {
 	stalenessSecs := roundDurationToSecs(stalenessInterval)
-	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + stalenessSecs
+	ignoreFirstSampleDeadline := fasttime.UnixTimestamp() + roundDurationToSecs(ignoreFirstSampleInterval)
 
 	return &totalAggrState{
 		resetTotalOnFlush:         resetTotalOnFlush,
@@ -60,7 +60,7 @@ func newTotalAggrState(stalenessInterval time.Duration, resetTotalOnFlush, keepF
 func (as *totalAggrState) pushSamples(samples []pushSample) {
 	currentTime := fasttime.UnixTimestamp()
 	deleteDeadline := currentTime + as.stalenessSecs
-	keepFirstSample := as.keepFirstSample && currentTime > as.ignoreFirstSampleDeadline
+	keepFirstSample := as.keepFirstSample && currentTime >= as.ignoreFirstSampleDeadline
 	for i := range samples {
 		s := &samples[i]
 		inputKey, outputKey := getInputOutputKey(s.key)


Test patch:
diff --git a/lib/streamaggr/streamaggr_test.go b/lib/streamaggr/streamaggr_test.go
index ae64a6ac7..8338be12c 100644
--- a/lib/streamaggr/streamaggr_test.go
+++ b/lib/streamaggr/streamaggr_test.go
@@ -554,6 +554,18 @@ foo 123
 bar{baz="qwe"} 4.34
 `, `bar:1m_total{baz="qwe"} 0
 foo:1m_total 0
+`, "11")
+
+	// total output for non-repeated series, ignore first sample 0s
+	f(`
+- interval: 1m
+  outputs: [total]
+  ignore_first_sample_interval: 0s
+`, `
+foo 123
+bar{baz="qwe"} 4.34
+`, `bar:1m_total{baz="qwe"} 4.34
+foo:1m_total 123
 `, "11")
 
 	// total_prometheus output for non-repeated series