Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6165 (2024-04-22 10:10:25)

Automatic update operator docs from VictoriaMetrics/operator@7e2ba6f


Patch:
diff --git a/README.md b/README.md
index d9b02c568..87d5ceb80 100644
--- a/README.md
+++ b/README.md
@@ -3159,6 +3159,8 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
      An optional list of labels to drop from samples before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels
      Supports an array of values separated by comma or specified via multiple flags.
      Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -streamAggr.ignoreFirstIntervals int
+     Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -streamAggr.ignoreOldSamples
      Whether to ignore input samples with old timestamps outside the current aggregation interval. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples
   -streamAggr.keepInput
diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go
index 41f41dc5f..7cd8415cb 100644
--- a/app/vmagent/remotewrite/remotewrite.go
+++ b/app/vmagent/remotewrite/remotewrite.go
@@ -105,6 +105,8 @@ var (
 		"with -remoteWrite.streamAggr.config . See also -dedup.minScrapeInterval and https://docs.victoriametrics.com/stream-aggregation/#deduplication")
 	streamAggrIgnoreOldSamples = flagutil.NewArrayBool("remoteWrite.streamAggr.ignoreOldSamples", "Whether to ignore input samples with old timestamps outside the current aggregation interval "+
 		"for the corresponding -remoteWrite.streamAggr.config . See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
+	streamAggrIgnoreFirstIntervals = flag.Int("remoteWrite.streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving unordered delayed data from clients pushing data into the vmagent. "+
+		"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
 	streamAggrDropInputLabels = flagutil.NewArrayString("streamAggr.dropInputLabels", "An optional list of labels to drop from samples "+
 		"before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
 
@@ -857,9 +859,10 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in
 	ignoreOldSamples := streamAggrIgnoreOldSamples.GetOptionalArg(argIdx)
 	if sasFile != "" {
 		opts := &streamaggr.Options{
-			DedupInterval:    dedupInterval,
-			DropInputLabels:  *streamAggrDropInputLabels,
-			IgnoreOldSamples: ignoreOldSamples,
+			DedupInterval:        dedupInterval,
+			DropInputLabels:      *streamAggrDropInputLabels,
+			IgnoreOldSamples:     ignoreOldSamples,
+			IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
 		}
 		sas, err := streamaggr.LoadFromFile(sasFile, rwctx.pushInternalTrackDropped, opts)
 		if err != nil {
diff --git a/app/vminsert/common/streamaggr.go b/app/vminsert/common/streamaggr.go
index 680956428..7a8ae0e11 100644
--- a/app/vminsert/common/streamaggr.go
+++ b/app/vminsert/common/streamaggr.go
@@ -32,6 +32,8 @@ var (
 		"See also -streamAggr.dropInputLabels and -dedup.minScrapeInterval and https://docs.victoriametrics.com/stream-aggregation/#deduplication")
 	streamAggrDropInputLabels = flagutil.NewArrayString("streamAggr.dropInputLabels", "An optional list of labels to drop from samples "+
 		"before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
+	streamAggrIgnoreFirstIntervals = flag.Int("streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. "+
+		"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
 	streamAggrIgnoreOldSamples = flag.Bool("streamAggr.ignoreOldSamples", false, "Whether to ignore input samples with old timestamps outside the current aggregation interval. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
 )
@@ -56,9 +58,10 @@ func CheckStreamAggrConfig() error {
 	}
 	pushNoop := func(_ []prompbmarshal.TimeSeries) {}
 	opts := &streamaggr.Options{
-		DedupInterval:    *streamAggrDedupInterval,
-		DropInputLabels:  *streamAggrDropInputLabels,
-		IgnoreOldSamples: *streamAggrIgnoreOldSamples,
+		DedupInterval:        *streamAggrDedupInterval,
+		DropInputLabels:      *streamAggrDropInputLabels,
+		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
+		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
 	}
 	sas, err := streamaggr.LoadFromFile(*streamAggrConfig, pushNoop, opts)
 	if err != nil {
@@ -84,9 +87,10 @@ func InitStreamAggr() {
 	sighupCh := procutil.NewSighupChan()
 
 	opts := &streamaggr.Options{
-		DedupInterval:    *streamAggrDedupInterval,
-		DropInputLabels:  *streamAggrDropInputLabels,
-		IgnoreOldSamples: *streamAggrIgnoreOldSamples,
+		DedupInterval:        *streamAggrDedupInterval,
+		DropInputLabels:      *streamAggrDropInputLabels,
+		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
+		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
 	}
 	sas, err := streamaggr.LoadFromFile(*streamAggrConfig, pushAggregateSeries, opts)
 	if err != nil {
@@ -117,9 +121,10 @@ func reloadStreamAggrConfig() {
 	saCfgReloads.Inc()
 
 	opts := &streamaggr.Options{
-		DedupInterval:    *streamAggrDedupInterval,
-		DropInputLabels:  *streamAggrDropInputLabels,
-		IgnoreOldSamples: *streamAggrIgnoreOldSamples,
+		DedupInterval:        *streamAggrDedupInterval,
+		DropInputLabels:      *streamAggrDropInputLabels,
+		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
+		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
 	}
 	sasNew, err := streamaggr.LoadFromFile(*streamAggrConfig, pushAggregateSeries, opts)
 	if err != nil {
diff --git a/docs/README.md b/docs/README.md
index ceb58995a..dcd35ae51 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -2679,7 +2679,7 @@ Files included in each folder:
 
 Pass `-help` to VictoriaMetrics in order to see the list of supported command-line flags with their description:
 
-```sh
+```
   -bigMergeConcurrency int
      Deprecated: this flag does nothing
   -blockcache.missesBeforeCaching int
@@ -3162,6 +3162,8 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
      An optional list of labels to drop from samples before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels
      Supports an array of values separated by comma or specified via multiple flags.
      Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -streamAggr.ignoreFirstIntervals int
+     Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -streamAggr.ignoreOldSamples
      Whether to ignore input samples with old timestamps outside the current aggregation interval. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples
   -streamAggr.keepInput
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index e83be4aa3..e3329834d 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -2687,7 +2687,7 @@ Files included in each folder:
 
 Pass `-help` to VictoriaMetrics in order to see the list of supported command-line flags with their description:
 
-```sh
+```
   -bigMergeConcurrency int
      Deprecated: this flag does nothing
   -blockcache.missesBeforeCaching int
@@ -3170,6 +3170,8 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
      An optional list of labels to drop from samples before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels
      Supports an array of values separated by comma or specified via multiple flags.
      Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -streamAggr.ignoreFirstIntervals int
+     Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after restarts. It could be caused by receiving unordered delayed data from clients pushing data into the database. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -streamAggr.ignoreOldSamples
      Whether to ignore input samples with old timestamps outside the current aggregation interval. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples
   -streamAggr.keepInput
diff --git a/docs/operator/CHANGELOG.md b/docs/operator/CHANGELOG.md
index a7d0d10a6..562524157 100644
--- a/docs/operator/CHANGELOG.md
+++ b/docs/operator/CHANGELOG.md
@@ -16,6 +16,13 @@ aliases:
 
 ## Next release
 
+<a name="v0.43.2"></a>
+
+## [v0.43.2](https://github.com/VictoriaMetrics/operator/releases/tag/v0.43.2) - 22 Apr 2024
+
+- [vmagent](./api.md#vmagent): fixes bug with `ServiceAccount` not found with `ingestOnlyMode`.
+- [vmagent](./api.md#vmagent): fixes `unknown long flag '--rules-dir'` for prometheus-config-reloader.
+
 <a name="v0.43.1"></a>
 ## [v0.43.1](https://github.com/VictoriaMetrics/operator/releases/tag/v0.43.1) - 18 Apr 2024
 
diff --git a/docs/operator/vars.md b/docs/operator/vars.md
index f293cefe3..9dc57bf30 100644
--- a/docs/operator/vars.md
+++ b/docs/operator/vars.md
@@ -10,7 +10,7 @@ menu:
 
 <!-- this doc autogenerated - don't edit it manually -->
 # Auto Generated vars for package config 
- updated at Thu Apr 18 19:07:23 UTC 2024 
+ updated at Mon Apr 22 10:10:22 UTC 2024 
 
 
 | varible name | variable default value | variable required | variable description |
diff --git a/docs/stream-aggregation.md b/docs/stream-aggregation.md
index 5b7027c2f..bd418dd14 100644
--- a/docs/stream-aggregation.md
+++ b/docs/stream-aggregation.md
@@ -19,7 +19,7 @@ The aggregation is applied to all the metrics received via any [supported data i
 and/or scraped from [Prometheus-compatible targets](https://docs.victoriametrics.com/#how-to-scrape-prometheus-exporters-such-as-node-exporter)
 after applying all the configured [relabeling stages](https://docs.victoriametrics.com/vmagent/#relabeling).
 
-By default stream aggregation ignores timestamps associated with the input [samples](https://docs.victoriametrics.com/keyconcepts/#raw-samples).
+By default, stream aggregation ignores timestamps associated with the input [samples](https://docs.victoriametrics.com/keyconcepts/#raw-samples).
 It expects that the ingested samples have timestamps close to the current time. See [how to ignore old samples](#ignoring-old-samples).
 
 Stream aggregation can be configured via the following command-line flags:
@@ -82,7 +82,7 @@ The online de-duplication uses the same logic as [`-dedup.minScrapeInterval` com
 
 ## Ignoring old samples
 
-By default all the input samples are taken into account during stream aggregation. If samples with old timestamps outside the current [aggregation interval](#stream-aggregation-config)
+By default, all the input samples are taken into account during stream aggregation. If samples with old timestamps outside the current [aggregation interval](#stream-aggregation-config)
 must be ignored, then the following options can be used:
 
 - To pass `-remoteWrite.streamAggr.ignoreOldSamples` command-line flag to [vmagent](https://docs.victoriametrics.com/vmagent/)
@@ -92,9 +92,25 @@ must be ignored, then the following options can be used:
 - To set `ignore_old_samples: true` option at the particular [aggregation config](#stream-aggregation-config).
   This enables ignoring old samples for that particular aggregation config.
 
+## Ignore aggregation intervals on start
+
+Stream aggregation may yield inaccurate results if it processes incomplete data. This issue can arise when data is 
+received from clients that maintain a queue of unsent data, such as Prometheus or vmagent. If the queue isn't fully 
+cleared within the aggregation `interval`, only a portion of the time series may be processed, leading to distorted 
+calculations. To mitigate this, consider the following options:
+
+- Set `-remoteWrite.streamAggr.ignoreFirstIntervals=<intervalsCount>` command-line flag to [vmagent](https://docs.victoriametrics.com/vmagent/)
+  or `-streamAggr.ignoreFirstIntervals=<intervalsCount>` command-line flag to [single-node VictoriaMetrics](https://docs.victoriametrics.com/)
+  to skip first `<intervalsCount>` [aggregation intervals](#stream-aggregation-config)
+  from persisting to the storage. It is expected that all incomplete or queued data will be processed during 
+  specified `<intervalsCount>` and all subsequent aggregation intervals will produce correct data.
+
+- Set `ignore_first_intervals: <intervalsCount>` option individually per [aggregation config](#stream-aggregation-config).
+  This enables ignoring first `<intervalsCount>` aggregation intervals for that particular aggregation config.
+
 ## Flush time alignment
 
-By default the time for aggregated data flush is aligned by the `interval` option specified in [aggregate config](#stream-aggregation-config).
+By default, the time for aggregated data flush is aligned by the `interval` option specified in [aggregate config](#stream-aggregation-config).
 For example:
 - if `interval: 1m` is set, then the aggregated data is flushed to the storage at the end of every minute
 - if `interval: 1h` is set, then the aggregated data is flushed to the storage at the end of every hour
@@ -875,7 +891,7 @@ at [single-node VictoriaMetrics](https://docs.victoriametrics.com/single-server-
   # Samples are de-duplicated on a per-series basis. See https://docs.victoriametrics.com/keyconcepts/#time-series
   # and https://docs.victoriametrics.com/#deduplication
   # The deduplication is performed after input_relabel_configs relabeling is applied.
-  # By default the deduplication is disabled unless -remoteWrite.streamAggr.dedupInterval or -streamAggr.dedupInterval
+  # By default, the deduplication is disabled unless -remoteWrite.streamAggr.dedupInterval or -streamAggr.dedupInterval
   # command-line flags are set.
   #
   # dedup_interval: 30s
@@ -892,7 +908,7 @@ at [single-node VictoriaMetrics](https://docs.victoriametrics.com/single-server-
   # staleness_interval: 2m
   
   # no_align_flush_to_interval disables aligning of flush times for the aggregated data to multiples of interval.
-  # By default flush times for the aggregated data is aligned to multiples of interval.
+  # By default, flush times for the aggregated data is aligned to multiples of interval.
   # For example:
   # - if `interval: 1m` is set, then flushes happen at the end of every minute,
   # - if `interval: 1h` is set, then flushes happen at the end of every hour
@@ -922,16 +938,23 @@ at [single-node VictoriaMetrics](https://docs.victoriametrics.com/single-server-
 
   # keep_metric_names instructs keeping the original metric names for the aggregated samples.
   # This option can be set only if outputs list contains only a single output.
-  # By default a special suffix is added to original metric names in the aggregated samples.
+  # By default, a special suffix is added to original metric names in the aggregated samples.
   # See https://docs.victoriametrics.com/stream-aggregation/#output-metric-names
   #
   # keep_metric_names: false
 
   # ignore_old_samples instructs ignoring input samples with old timestamps outside the current aggregation interval.
+  # See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples
   # See also -streamAggr.ignoreOldSamples command-line flag.
   #
   # ignore_old_samples: false
 
+  # ignore_first_intervals instructs ignoring first N aggregation intervals after process start.
+  # See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
+  # See also -remoteWrite.streamAggr.ignoreFirstIntervals or -streamAggr.ignoreFirstIntervals
+  #
+  # ignore_first_intervals: false
+
   # drop_input_labels instructs dropping the given labels from input samples.
   # The labels' dropping is performed before input_relabel_configs are applied.
   # This also means that the labels are dropped before de-duplication ( https://docs.victoriametrics.com/stream-aggregation/#deduplication )
@@ -1004,7 +1027,7 @@ These issues can be be fixed in the following ways:
 - By increasing the `interval` option at [stream aggregation config](#stream-aggregation-config), so it covers the expected
   delays in data ingestion pipelines.
 - By specifying the `staleness_interval` option at [stream aggregation config](#stream-aggregation-config), so it covers the expected
-  delays in data ingestion pipelines. By default the `staleness_interval` equals to `2 x interval`.
+  delays in data ingestion pipelines. By default, the `staleness_interval` equals to `2 x interval`.
 
 ### High resource usage
 
diff --git a/docs/vmagent.md b/docs/vmagent.md
index ecc2f0e07..752efb903 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -1623,7 +1623,7 @@ It is safe sharing the collected profiles from security point of view, since the
 
 `vmagent` can be fine-tuned with various command-line flags. Run `./vmagent -help` in order to see the full list of these flags with their descriptions and default values:
 
-```sh
+```
 ./vmagent -help
 
 vmagent collects metrics data via popular data ingestion protocols and routes them to VictoriaMetrics.
@@ -2166,6 +2166,8 @@ See the docs at https://docs.victoriametrics.com/vmagent/ .
      Whether to drop all the input samples after the aggregation with -remoteWrite.streamAggr.config. By default, only aggregates samples are dropped, while the remaining samples are written to the corresponding -remoteWrite.url . See also -remoteWrite.streamAggr.keepInput and https://docs.victoriametrics.com/stream-aggregation/
      Supports array of values separated by comma or specified via multiple flags.
      Empty values are set to false.
+  -remoteWrite.streamAggr.ignoreFirstIntervals int
+     Number of aggregation intervals to skip after the start. Increase this value if you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving unordered delayed data from clients pushing data into the vmagent. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start
   -remoteWrite.streamAggr.ignoreOldSamples array
      Whether to ignore input samples with old timestamps outside the current aggregation interval for the corresponding -remoteWrite.streamAggr.config . See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples
      Supports array of values separated by comma or specified via multiple flags.
diff --git a/docs/vmalert.md b/docs/vmalert.md
index 1534ee835..a5c03c3f3 100644
--- a/docs/vmalert.md
+++ b/docs/vmalert.md
@@ -965,7 +965,7 @@ command-line flags with their descriptions.
 
 The shortlist of configuration flags is the following:
 
-```sh
+```
   -clusterMode
      If clusterMode is enabled, then vmalert automatically adds the tenant specified in config groups to -datasource.url, -remoteWrite.url and -remoteRead.url. See https://docs.victoriametrics.com/vmalert/#multitenancy . This flag is available only in Enterprise binaries. See https://docs.victoriametrics.com/enterprise/
   -configCheckInterval duration
diff --git a/docs/vmauth.md b/docs/vmauth.md
index 56d5367cc..b69dd8d48 100644
--- a/docs/vmauth.md
+++ b/docs/vmauth.md
@@ -1121,7 +1121,7 @@ It is safe sharing the collected profiles from security point of view, since the
 
 Pass `-help` command-line arg to `vmauth` in order to see all the configuration options:
 
-```sh
+```
 ./vmauth -help
 
 vmauth authenticates and authorizes incoming requests and proxies them to VictoriaMetrics.
diff --git a/docs/vmbackup.md b/docs/vmbackup.md
index 628e364bf..465e2f8a5 100644
--- a/docs/vmbackup.md
+++ b/docs/vmbackup.md
@@ -304,7 +304,7 @@ Refer to the respective documentation for your object storage provider for more
 
 Run `vmbackup -help` in order to see all the available options:
 
-```sh
+```
   -concurrency int
      The number of concurrent workers. Higher concurrency may reduce backup duration (default 10)
   -configFilePath string
diff --git a/docs/vmbackupmanager.md b/docs/vmbackupmanager.md
index 2bcd7dbdb..dcef2ab9c 100644
--- a/docs/vmbackupmanager.md
+++ b/docs/vmbackupmanager.md
@@ -415,7 +415,7 @@ command-line flags with their descriptions.
 
 The shortlist of configuration flags is the following:
 
-```text
+```
 vmbackupmanager performs regular backups according to the provided configs.
 
 subcommands:
diff --git a/docs/vmgateway.md b/docs/vmgateway.md
index 09445aee8..cc263e66a 100644
--- a/docs/vmgateway.md
+++ b/docs/vmgateway.md
@@ -273,7 +273,7 @@ Example usage for tokens issued by Google:
 
 Below is the list of configuration flags (it can be viewed by running `./vmgateway -help`):
 
-```sh
+```
   -auth.httpHeader string
      HTTP header name to look for JWT authorization token (default "Authorization")
   -auth.jwksEndpoints array
diff --git a/docs/vmrestore.md b/docs/vmrestore.md
index a8dec2ce1..f640539a9 100644
--- a/docs/vmrestore.md
+++ b/docs/vmrestore.md
@@ -91,7 +91,7 @@ i.e. the end result would be similar to [rsync --delete](https://askubuntu.com/q
 
 * Run `vmrestore -help` in order to see all the available options:
 
-```sh
+```
   -concurrency int
      The number of concurrent workers. Higher concurrency may reduce restore duration (default 10)
   -configFilePath string
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index 06d281f69..5cb480aad 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -111,6 +111,13 @@ type Options struct {
 	//
 	// This option can be overridden individually per each aggregation via ignore_old_samples option.
 	IgnoreOldSamples bool
+
+	// IgnoreFirstIntervals sets amount of aggregation intervals to ignore on start.
+	//
+	// By default, no intervals will be ignored.
+	//
+	// This option can be overridden individually per each aggregation via ignore_first_intervals option.
+	IgnoreFirstIntervals int
 }
 
 // Config is a configuration for a single stream aggregation.
@@ -175,6 +182,9 @@ type Config struct {
 	// IgnoreOldSamples instructs to ignore samples with old timestamps outside the current aggregation interval.
 	IgnoreOldSamples *bool `yaml:"ignore_old_samples,omitempty"`
 
+	// IgnoreFirstIntervals sets number of aggregation intervals to be ignored on start.
+	IgnoreFirstIntervals *int `yaml:"ignore_first_intervals,omitempty"`
+
 	// By is an optional list of labels for grouping input series.
 	//
 	// See also Without.
@@ -479,6 +489,12 @@ func newAggregator(cfg *Config, pushFunc PushFunc, ms *metrics.Set, opts *Option
 		ignoreOldSamples = *v
 	}
 
+	// check cfg.IgnoreFirstIntervals
+	ignoreFirstIntervals := opts.IgnoreFirstIntervals
+	if v := cfg.IgnoreFirstIntervals; v != nil {
+		ignoreFirstIntervals = *v
+	}
+
 	// initialize outputs list
 	if len(cfg.Outputs) == 0 {
 		return nil, fmt.Errorf("`outputs` list must contain at least a single entry from the list %s; "+
@@ -600,14 +616,14 @@ func newAggregator(cfg *Config, pushFunc PushFunc, ms *metrics.Set, opts *Option
 
 	a.wg.Add(1)
 	go func() {
-		a.runFlusher(pushFunc, alignFlushToInterval, skipIncompleteFlush, interval, dedupInterval)
+		a.runFlusher(pushFunc, alignFlushToInterval, skipIncompleteFlush, interval, dedupInterval, ignoreFirstIntervals)
 		a.wg.Done()
 	}()
 
 	return a, nil
 }
 
-func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipIncompleteFlush bool, interval, dedupInterval time.Duration) {
+func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipIncompleteFlush bool, interval, dedupInterval time.Duration, ignoreFirstIntervals int) {
 	alignedSleep := func(d time.Duration) {
 		if !alignFlushToInterval {
 			return
@@ -642,7 +658,12 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 		}
 
 		for tickerWait(t) {
-			a.flush(pushFunc, interval, true)
+			pf := pushFunc
+			if ignoreFirstIntervals > 0 {
+				pf = nil
+				ignoreFirstIntervals--
+			}
+			a.flush(pf, interval, true)
 
 			if alignFlushToInterval {
 				select {
@@ -663,13 +684,17 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 
 			ct := time.Now()
 			if ct.After(flushDeadline) {
+				pf := pushFunc
+				if ignoreFirstIntervals > 0 {
+					pf = nil
+					ignoreFirstIntervals--
+				}
 				// It is time to flush the aggregated state
 				if alignFlushToInterval && skipIncompleteFlush && !isSkippedFirstFlush {
-					a.flush(nil, interval, true)
+					pf = nil
 					isSkippedFirstFlush = true
-				} else {
-					a.flush(pushFunc, interval, true)
 				}
+				a.flush(pf, interval, true)
 				for ct.After(flushDeadline) {
 					flushDeadline = flushDeadline.Add(interval)
 				}
@@ -684,7 +709,7 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 		}
 	}
 
-	if !skipIncompleteFlush {
+	if !skipIncompleteFlush && ignoreFirstIntervals == 0 {
 		a.dedupFlush(dedupInterval)
 		a.flush(pushFunc, interval, true)
 	}


Test patch:
diff --git a/lib/streamaggr/streamaggr_test.go b/lib/streamaggr/streamaggr_test.go
index 5a079979b..c4731673b 100644
--- a/lib/streamaggr/streamaggr_test.go
+++ b/lib/streamaggr/streamaggr_test.go
@@ -199,6 +199,14 @@ func TestAggregatorsEqual(t *testing.T) {
   interval: 5m
   flush_on_shutdown: false
 `, false)
+	f(`
+- outputs: [total]
+  interval: 5m
+  ignore_first_intervals: 2
+`, `
+- outputs: [total]
+  interval: 5m
+  ignore_first_intervals: 4`, false)
 }
 
 func TestAggregatorsSuccess(t *testing.T) {