Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6768 (2024-08-06 21:58:37)

docs/vmanomaly: release v1.15.0
### Describe Your Changes

- updated docs on `vmanomaly` with v1.15.0
- additional chapters of FAQ and model pages

### Checklist

The following checks are **mandatory**:

- [x] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).



Patch:
diff --git a/app/vmagent/remotewrite/streamaggr.go b/app/vmagent/remotewrite/streamaggr.go
index 0ebc2e624..f20df53f0 100644
--- a/app/vmagent/remotewrite/streamaggr.go
+++ b/app/vmagent/remotewrite/streamaggr.go
@@ -50,7 +50,7 @@ var (
 	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 at the corresponding -remoteWrite.url. "+
 		"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 "+
+	streamAggrIgnoreFirstIntervals = flagutil.NewArrayInt("remoteWrite.streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start "+
 		"for the corresponding -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. Increase this value if "+
 		"you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving bufferred delayed data from clients pushing data into the vmagent. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
@@ -133,7 +133,7 @@ func initStreamAggrConfigGlobal() {
 	} else {
 		dedupInterval := streamAggrGlobalDedupInterval.Duration()
 		if dedupInterval > 0 {
-			deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, dedupInterval, *streamAggrDropInputLabels, "dedup-global")
+			deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
 		}
 	}
 }
@@ -229,7 +229,7 @@ func newStreamAggrConfigPerURL(idx int, pushFunc streamaggr.PushFunc) (*streamag
 		DedupInterval:        streamAggrDedupInterval.GetOptionalArg(idx),
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     streamAggrIgnoreOldSamples.GetOptionalArg(idx),
-		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
+		IgnoreFirstIntervals: streamAggrIgnoreFirstIntervals.GetOptionalArg(idx),
 	}
 
 	sas, err := streamaggr.LoadFromFile(path, pushFunc, opts, alias)
diff --git a/app/vmalert/rule/alerting.go b/app/vmalert/rule/alerting.go
index 632ac896e..a4ce51997 100644
--- a/app/vmalert/rule/alerting.go
+++ b/app/vmalert/rule/alerting.go
@@ -441,9 +441,6 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 			}
 			a.Value = m.Values[0]
 			a.Annotations = annotations
-			if err != nil {
-				return nil, err
-			}
 			a.KeepFiringSince = time.Time{}
 			continue
 		}
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index c0af992d6..e16800ac1 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -32,8 +32,12 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 * FEATURE: add `/influx/health` health-check handler for Influx endpoints. This is needed as some clients use the health endpoint to determine if the server is healthy and ready for data ingestion. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6653) for the details.
 * FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl/): add `--vm-backoff-retries`, `--vm-backoff-factor`, `--vm-backoff-min-duration` and `--vm-native-backoff-retries`, `--vm-native-backoff-factor`, `--vm-native-backoff-min-duration` command-line flags. These flags allow to change backoff policy config for import requests to VictoriaMetrics. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6622).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): allow overriding the `sample_limit` option at [scrape_configs](https://docs.victoriametrics.com/sd_configs/#scrape_configs) when a label `__sample_limit__` is specified for target. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6665). Thanks to @zoglam for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6666).
 
 * BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert): respect HTTP headers defined in [notifier configuration file](https://docs.victoriametrics.com/vmalert/#notifier-configuration-file) for each request to notifiers. Previously, this param was ignored by mistake.
+* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): correctly apply `-streamAggr.dropInputLabels` when global stream deduplication is enabled without `-streamAggr.config`. Previously, `-remoteWrite.streamAggr.dropInputLabels` was used instead.
+* BUGFIX: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): fix command-line flag `-remoteWrite.streamAggr.ignoreFirstIntervals` to accept multiple values and be applied per each corresponding `-remoteWrite.url`. Previously, this flag only could have been used globally for all URLs. 
+* BUGFIX: [graphite](https://docs.victoriametrics.com/#graphite-render-api-usage): respect `-search.denyPartialResponse` cmd-line flag and `deny_partial_response` query GET param when serving requests via Graphite API. Before, partial responses were always denied. Thanks to @penguinlav for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6748).
 
 ## [v1.102.1](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.102.1)
 
diff --git a/docs/anomaly-detection/CHANGELOG.md b/docs/anomaly-detection/CHANGELOG.md
index 2061c987f..85869ed55 100644
--- a/docs/anomaly-detection/CHANGELOG.md
+++ b/docs/anomaly-detection/CHANGELOG.md
@@ -13,6 +13,19 @@ aliases:
 Please find the changelog for VictoriaMetrics Anomaly Detection below.
 
 > **Important note: Users are strongly encouraged to upgrade to `vmanomaly` [v1.9.2](https://hub.docker.com/repository/docker/victoriametrics/vmanomaly/tags?page=1&ordering=name) or newer for optimal performance and accuracy. <br><br> This recommendation is crucial for configurations with a low `infer_every` parameter [in your scheduler](./components/scheduler.md#parameters-1), and in scenarios where data exhibits significant high-order seasonality patterns (such as hourly or daily cycles). Previous versions from v1.5.1 to v1.8.0 were identified to contain a critical issue impacting model training, where models were inadvertently trained on limited data subsets, leading to suboptimal fits, affecting the accuracy of anomaly detection. <br><br> Upgrading to v1.9.2 addresses this issue, ensuring proper model training and enhanced reliability. For users utilizing Helm charts, it is recommended to upgrade to version [1.0.0](https://github.com/VictoriaMetrics/helm-charts/blob/master/charts/victoria-metrics-anomaly/CHANGELOG.md#100) or newer.**
+
+## v1.15.0
+Released: 2024-08-06
+- FEATURE: Introduced models that support [online learning](https://en.wikipedia.org/wiki/Online_machine_learning) for stream-like input. These models significantly reduce the amount of data required for the initial fit stage. For example, they enable reducing `fit_every` from **weeks to hours** and increasing `fit_every` from **hours to weeks** in the [PeriodicScheduler](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/#periodic-scheduler), significantly reducing the **peak amount** of data queried from VictoriaMetrics during `fit` stages. The next models were added:
+  - [`OnlineZscoreModel`](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-z-score) - online version of existing [Z-score](https://docs.victoriametrics.com/anomaly-detection/components/models/#z-score) implementation with the same exact behavior.
+  - [`OnlineMADModel`](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-mad) - online version of existing [MADModel](https://docs.victoriametrics.com/anomaly-detection/components/models/#mad-median-absolute-deviation) implementation with *approximate* behavior, based on [t-digests](https://www.sciencedirect.com/science/article/pii/S2665963820300403) for online quantile estimation.
+  - [`OnlineQuantileModel`](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-seasonal-quantile) - online quantile model, that supports custom ranges for seasonality estimation to cover more complex data patterns.
+  - Find out more about online models specifics in [correspondent section](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-models).
+
+- FEATURE: Introduced the `optimized_business_params` key (list of strings) to the [`AutoTuned`](https://docs.victoriametrics.com/anomaly-detection/components/models/#autotuned) `optimization_params`. This allows particular business-specific parameters such as [`detection_direction`](https://docs.victoriametrics.com/anomaly-detection/components/models/#detection-direction) and [`min_dev_from_expected`](https://docs.victoriametrics.com/anomaly-detection/components/models/#minimal-deviation-from-expected) to remain **unchanged during optimizations, retaining their default values**.
+- IMPROVEMENT: Optimized the [`AutoTuned`](https://docs.victoriametrics.com/anomaly-detection/components/models/#autotuned) model logic to minimize deviations from the expected `anomaly_percentage` specified in the configuration and the detected percentage in the data, while also reducing discrepancies between the actual values (`y`) and the predictions (`yhat`).
+- IMPROVEMENT: Allow [`ProphetModel`](https://docs.victoriametrics.com/anomaly-detection/components/models/#prophet) to fit with multiple seasonalities when used in [`AutoTuned`](https://docs.victoriametrics.com/anomaly-detection/components/models/#autotuned) mode. 
+
 ## v1.14.2
 Released: 2024-07-26
 - FIX: Patch a bug introduced in [v1.14.1](#v1141), causing `vmanomaly` to crash in `preset` [mode](/anomaly-detection/presets/).
diff --git a/docs/anomaly-detection/FAQ.md b/docs/anomaly-detection/FAQ.md
index 780d955ee..12c42bc2e 100644
--- a/docs/anomaly-detection/FAQ.md
+++ b/docs/anomaly-detection/FAQ.md
@@ -34,6 +34,8 @@ Please see example graph illustrating this logic below:
 
 ![anomaly-score-calculation-example](vmanomaly-prophet-example.webp)
 
+> p.s. please note that additional post-processing logic might be applied to produced anomaly scores, if common arguments like [`min_dev_from_expected`](https://docs.victoriametrics.com/anomaly-detection/components/models/#minimal-deviation-from-expected) or [`detection_direction`](https://docs.victoriametrics.com/anomaly-detection/components/models/#detection-direction) are enabled for a particular model. Follow the links above for the explanations.
+
 
 ## How does vmanomaly work?
 `vmanomaly` applies built-in (or custom) [anomaly detection algorithms](/anomaly-detection/components/models), specified in a config file. 
@@ -150,19 +152,85 @@ volumes:
   vmanomaly_model_dump_dir: {}
 ```
 
+> **Note**: Starting from [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/changelog#v1150) with the introduction of [online models](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-models), you can additionally reduce resource consumption (e.g., flatten `fit` stage peaks by querying less data from VictoriaMetrics at once).
+
+**Additional Benefits of Switching to Online Models**:
+
+- **Reduced Latency**: Online models update incrementally, which can lead to faster response times for anomaly detection since the model continuously adapts to new data without waiting for a batch `fit`.
+- **Scalability**: Handling smaller data chunks at a time reduces memory and computational overhead, making it easier to scale the anomaly detection system.
+- **Improved Resource Utilization**: By spreading the computational load over time and reducing peak demands, online models make more efficient use of system resources, potentially lowering operational costs.
+
+Here's an example of how we can switch from (offline) [Z-score model](https://docs.victoriametrics.com/anomaly-detection/components/models/#z-score) to [Online Z-score model](https://docs.victoriametrics.com/anomaly-detection/components/models/#online-z-score):
+
+```yaml
+schedulers:
+  periodic:
+    class: 'periodic'
+    fit_every: '1h'
+    fit_window: '2d'
+    infer_every: '1m'
+  # other schedulers ...
+models:
+  zscore_example:
+    class: 'zscore'
+    schedulers: ['periodic']
+    # other model params ...
+# other config sections ...
+```
+
+to something like
+
+```yaml
+schedulers:
+  periodic:
+    class: 'periodic'
+    fit_every: '180d'  # we need only initial fit to start
+    fit_window: '4h'  # reduced window, especially if the data doesn't have strong seasonality
+    infer_every: '1m'  # the model will be updated during each infer call
+  # other schedulers ...
+models:
+  zscore_example:
+    class: 'zscore_online'
+    min_n_samples_seen: 120  # i.e. minimal relevant seasonality or (initial) fit_window / sampling_frequency
+    schedulers: ['periodic']
+    # other model params ...
+# other config sections ...
+```
+
+As a result, switching from the offline Z-score model to the Online Z-score model results in significant data volume reduction, i.e. over one week:
+
+**Old Configuration**: 
+- `fit_window`: 2 days
+- `fit_every`: 1 hour
+
+**New Configuration**: 
+- `fit_window`: 4 hours
+- `fit_every`: 180 days ( >1 week)
+
+The old configuration would perform 168 (hours in a week) `fit` calls, each using 2 days (48 hours) of data, totaling 168 * 48 = 8064 hours of data for each timeseries returned.
+
+The new configuration performs only 1 `fit` call in 180 days, using 4 hours of data initially, totaling 4 hours of data, which is **magnitutes smaller**.
+
+P.s. `infer` data volume will remain the same for both models, so it does not affect the overall calculations.
+
+**Data Volume Reduction**:
+- Old: 8064 hours/week (fit) + 168 hours/week (infer)
+- New:    4 hours/week (fit) + 168 hours/week (infer)
+
 ## Scaling vmanomaly
-> **Note:** As of latest release we don't support cluster or auto-scaled version yet (though, it's in our roadmap for - better backends, more parallelization, etc.), so proposed workarounds should be addressed manually.
+> **Note:** As of latest release we do not support cluster or auto-scaled version yet (though, it's in our roadmap for - better backends, more parallelization, etc.), so proposed workarounds should be addressed *manually*.
 
 `vmanomaly` can be scaled horizontally by launching multiple independent instances, each with its own [MetricsQL](/MetricsQL) queries and [configurations](/anomaly-detection/components/):
 
-- By splitting **queries**, [defined in reader section](/anomaly-detection/components/reader#vm-reader) and spawn separate service around it. Also in case you have *only 1 query returning huge amount of timeseries*, you can further split it by applying MetricsQL filters, i.e. using "extra_filters" [param in reader](/anomaly-detection/components/reader?highlight=extra_filters#vm-reader)
+- By splitting **queries**, [defined in reader section](/anomaly-detection/components/reader#vm-reader) and spawn separate service around it. Also in case you have *only 1 query returning huge amount of timeseries*, you can further split it by applying MetricsQL filters, i.e. using "extra_filters" [param in reader](/anomaly-detection/components/reader?highlight=extra_filters#vm-reader). See the example below.
 
 - or **models** (in case you decide to run several models for each timeseries received i.e. for averaging anomaly scores in your alerting rules of `vmalert` or using a vote approach to reduce false positives) - see `queries` arg in [model config](/anomaly-detection/components/models#queries)
 
 - or **schedulers** (in case you want the same models to be trained under several schedules) - see `schedulers` arg [model section](/anomaly-detection/components/models#schedulers) and `scheduler` [component itself](/anomaly-detection/components/scheduler)
 
 
-Here's an example of how to split on `extra_filters`, based on `extra_filters` reader's arg
+Here's an example of how to split on `extra_filters`, based on `extra_filters` reader's arg:
+
 ```yaml
 # config file #1, for 1st vmanomaly instance
 # ...
diff --git a/docs/anomaly-detection/components/models.md b/docs/anomaly-detection/components/models.md
index 76da10fc5..aa2c0325c 100644
--- a/docs/anomaly-detection/components/models.md
+++ b/docs/anomaly-detection/components/models.md
@@ -232,10 +232,12 @@ There are **2 model types**, supported in `vmanomaly`, resulting in **4 possible
 - [Univariate models](#univariate-models)
 - [Multivariate models](#multivariate-models)
 
-Each of these models can also be
+Each of these models can be of type
 - [Rolling](#rolling-models)
 - [Non-rolling](#non-rolling-models)
 
+Moreover, starting from [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1150), there exist **[online (incremental) models](#online-models)** subclass. Please refer to the [correspondent section](#online-models) for more details.
+
 ### Univariate Models
 
 For a univariate type, **one separate model** is fit/used for inference per **each time series**, defined in its [queries](#queries) arg.
@@ -276,7 +278,7 @@ As a result, such model instances are **not stored** between consecutive re-fit
 
 Such models put **more pressure** on your reader's source, i.e. if your model should be fit on large amount of data (say, 14 days with 1-minute resolution) and at the same time you have **frequent inference** (say, once per minute) on new chunks of data - that's because such models require (fit + infer) window of data to be fit first to be used later in each inference call.
 
-> **Note**: Rolling models require `fit_every` to be set equal to `infer_every` in your [PeriodicScheduler](./scheduler.md#periodic-scheduler).
+> **Note**: Rolling models require `fit_every` either to be missing or explicitly set equal to `infer_every` in your [PeriodicScheduler](./scheduler.md#periodic-scheduler).
 
 **Examples:** [RollingQuantile](#rolling-quantile)
 
@@ -300,24 +302,65 @@ Produced model instances are **stored in-memory** between consecutive re-fit cal
 ![vmanomaly-model-type-non-rolling](model-type-non-rolling.webp)
 
 
+### Online Models
+
+Introduced in [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1150), online (incremental) models allow defining a smaller frame `fit_window` and less frequent `fit` calls to reduce the data burden from VictoriaMetrics. They make incremental updates to model parameters during each `infer_every` call, even on a single datapoint.
+If the model doesn't support online mode, it's called **offline** (its parameters are only updated during `fit` calls).
+
+Main differences between offline and online:
+
+Fit stage
+- Both types have a `fit` stage, run on the `fit_window` data frame.
+- For offline models, `fit_window` should contain enough data to train the model (e.g., 2 seasonal periods).
+- For online models, training can start gradually from smaller chunks (e.g., 1 hour).
+
+Infer stage
+- Both types have an `infer` stage, run on new datapoints (timestamps > last seen timestamp of the previous `infer` call).
+- Offline models use a pre-trained (during `fit` call) *static* model to make every `infer` call until the next `fit` call, when the model is completely re-trained.
+- Online models use a pre-trained (during `fit` call) *dynamic* model, which is gradually updated during each `infer` call with new datapoints. However, to prevent the model from accumulating outdated behavior, each `fit` call resets the model from scratch.
+
+**Strengths**:
+
+- The ability to distribute the data load evenly between the initial `fit` and subsequent `infer` calls. For example, an online model can be fit on 10 `1m` datapoints during the initial `fit` stage once per month and then be gradually updated on the same 10 `1m` datapoints during each `infer` call each 10 minutes.
+- The model can adapt to new data patterns (gradually updating itself during each `infer` call) without needing to wait for the next `fit` call and one big re-training.
+- Slightly faster training/updating times compared to similar offline models.
+
+**Limitations**:
+
+- Until the online model sees enough data (especially if the data shows strong seasonality), its predictions might be unstable, producing more [false positives](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#false-positive) (`anomaly_score > 1`) or making [false negative predictions, skipping real anomalies](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#false-negative).
+- Not all models (e.g., complex ones like [Prophet](#prophet)) have a direct online alternative, thus their applicability can be somewhat limited.
+
+Each of the ([built-in](#built-in-models) or [custom](#custom-model-guide)) online models (like [`OnlineZscoreModel`](#online-z-score)) shares the following common parameters and properties:
+- `n_samples_seen_` (int) - this model *property* refers to the number of datapoints the model was trained on and increases from 0 (before the first `fit`) with each consecutive `infer` call.
+- `min_n_samples_seen` (int), optional - this *parameter* defines the minimum number of samples to be seen before reliably computing the [anomaly score](https://docs.victoriametrics.com/anomaly-detection/faq/#what-is-anomaly-score). Otherwise, the anomaly score will be 0 until `n_samples_seen_` > `min_n_samples_seen`, as there is not enough data to trust the model's predictions. For example, if your data has hourly seasonality and '1m' frequency, setting `min_n_samples_seen_` to 288 (1440 minutes in a day / 5 minutes) should be sufficient.
+
+
+### Offline models
+
+Every other model that isn't [online](#online-models). Offline models are completely re-trained during `fit` call and isn't updated during consecutive `infer` calls.
+
+
 ## Built-in Models 
 
 ### Overview
-VM Anomaly Detection (`vmanomaly` hereinafter) models support 2 groups of parameters:
+VictoriaMetrics Anomaly Detection models support 2 groups of parameters:
 
 - **`vmanomaly`-specific** arguments - please refer to *Parameters specific for vmanomaly* and *Default model parameters* subsections for each of the models below.
 - Arguments to **inner model** (say, [Facebook's Prophet](https://facebook.github.io/prophet/docs/quick_start.html#python-api)), passed in a `args` argument as key-value pairs, that will be directly given to the model during initialization to allow granular control. Optional.
 
-**Note**: For users who may not be familiar with Python data types such as `list[dict]`, a [dictionary](https://www.w3schools.com/python/python_dictionaries.asp) in Python is a data structure that stores data values in key-value pairs. This structure allows for efficient data retrieval and management.
+> **Note**: For users who may not be familiar with Python data types such as `list[dict]`, a [dictionary](https://www.w3schools.com/python/python_dictionaries.asp) in Python is a data structure that stores data values in key-value pairs. This structure allows for efficient data retrieval and management.
 
 
 **Models**:
 * [AutoTuned](#autotuned) - designed to take the cognitive load off the user, allowing any of built-in models below to be re-tuned for best params on data seen during each `fit` phase of the algorithm. Tradeoff is between increased computational time and optimized results / simpler maintenance.
 * [Prophet](#prophet) - the most versatile one for production usage, especially for complex data ([trends](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend), [change points](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-2/#novelties), [multi-seasonality](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality))
-* [Z-score](#z-score) - useful for testing and for simpler data ([de-trended](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) data without strict [seasonality](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality) and with anomalies of similar magnitude as your "normal" data)
-* [Holt-Winters](#holt-winters) - well-suited for **data with moderate complexity**, exhibiting distinct [trends](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) and/or [seasonal patterns](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality).
-* [MAD (Median Absolute Deviation)](#mad-median-absolute-deviation) - similarly to Z-score, is effective for **identifying outliers in relatively consistent data** (useful for detecting sudden, stark deviations from the median)
+* [Z-score](#z-score) - useful for initial testing and for simpler data ([de-trended](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) data without strict [seasonality](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality) and with anomalies of similar magnitude as your "normal" data)
+* [Online Z-score](#online-z-score) - [online](#online-models) alternative to [Z-score](#z-score) model with exact same behavior and use cases.
+* [Holt-Winters](#holt-winters) - well-suited for **data with moderate complexity**, exhibiting distinct [trends](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) and/or [single seasonal pattern](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality).
+* [MAD (Median Absolute Deviation)](#mad-median-absolute-deviation) - similarly to [Z-score](#z-score), is effective for **identifying outliers in relatively consistent data** (useful for detecting sudden, stark deviations from the median).
+* [Online MAD](#online-mad) - approximate [online](#online-models) alternative to [MAD model](#mad-median-absolute-deviation), appropriate for the same use cases.
 * [Rolling Quantile](#rolling-quantile) - best for **data with evolving patterns**, as it adapts to changes over a rolling window.
+* [Online Seasonal Quantile](#online-seasonal-quantile) - best used on **[de-trended](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) data with strong (possibly multiple) [seasonalities](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality)**. Can act as a (slightly less powerful) [online](#online-models) replacement to [`ProphetModel`](#prophet).
 * [Seasonal Trend Decomposition](#seasonal-trend-decomposition) - similarly to Holt-Winters, is best for **data with pronounced [seasonal](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality) and [trend](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) components**
 * [Isolation forest (Multivariate)](#isolation-forest-multivariate) - useful for **metrics data interaction** (several queries/metrics -> single anomaly score) and **efficient in detecting anomalies in high-dimensional datasets**
 * [Custom model](#custom-model-guide) - benefit from your own models and expertise to better support your **unique use case**.
@@ -331,7 +374,8 @@ Tuning hyperparameters of a model can be tricky and often requires in-depth know
 * `class` (string) - model class name `"model.auto.AutoTunedModel"` (or `auto` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support)
 * `tuned_class_name` (string) - Built-in model class to tune, i.e. `model.zscore.ZscoreModel` (or `zscore` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support).
 * `optimization_params` (dict) - Optimization parameters for unsupervised model tuning. Control % of found anomalies, as well as a tradeoff between time spent and the accuracy. The more `timeout` and `n_trials` are, the better model configuration can be found for `tuned_class_name`, but the longer it takes and vice versa. Set `n_jobs` to `-1` to use all the CPUs available, it makes sense if only you have a big dataset to train on during `fit` calls, otherwise overhead isn't worth it.
-  - `anomaly_percentage` (float) - expected percentage of anomalies that can be seen in training data, from (0, 0.5) interval.
+  - `anomaly_percentage` (float) - Expected percentage of anomalies that can be seen in training data, from (0, 0.5) interval.
+  - `optimized_business_params` (list[string]) - Starting from [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/v1150) this argument allows particular business-specific parameters such as [`detection_direction`](https://docs.victoriametrics.com/anomaly-detection/components/models/#detection-direction) or [`min_dev_from_expected`](https://docs.victoriametrics.com/anomaly-detection/components/models/#minimal-deviation-from-expected) to remain **unchanged during optimizations, retaining their default values**. I.e. setting `optimized_business_params` to  `['detection_direction']` will allow to optimize only `detection_direction` business-specific, while `min_dev_from_expected` will retain default value (0.0). By default and if not set, will be set to `[]` (empty list), meaning no business params will be optimized. **A recommended option is to leave it empty** for more stable results and increased convergence (less iterations needed for a good result).
   - `seed` (int) - Random seed for reproducibility and deterministic nature of underlying optimizations.
   - `n_splits` (int) - How many folds to create for hyperparameter tuning out of your data. The higher, the longer it takes but the better the results can be. Defaults to 3.
   - `n_trials` (int) - How many trials to sample from hyperparameter search space. The higher, the longer it takes but the better the results can be. Defaults to 128.
@@ -353,6 +397,7 @@ models:
       n_trials: 128  # how many configurations to sample from search space during optimization
       timeout: 10  # how many seconds to spend on optimization for each trained model during `fit` phase call
       n_jobs: 1  # how many jobs in parallel to launch. Consider making it > 1 only if you have fit window containing > 10000 datapoints for each series
+      optimized_business_params: []  # business-specific params to include in optimization, if not set is empty list
   # ...
 ```
 
@@ -365,6 +410,8 @@ models:
 ### [Prophet](https://facebook.github.io/prophet/)
 Here we utilize the Facebook Prophet implementation, as detailed in their [library documentation](https://facebook.github.io/prophet/docs/quick_start.html#python-api). All parameters from this library are compatible and can be passed to the model.
 
+> **Note**: `ProphetModel` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [offline](#offline-models) model.
+
 *Parameters specific for vmanomaly*:
 
 * `class` (string) - model class name `"model.prophet.ProphetModel"` (or `prophet` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support)
@@ -405,6 +452,11 @@ models:
 Resulting metrics of the model are described [here](#vmanomaly-output)
 
 ### [Z-score](https://en.wikipedia.org/wiki/Standard_score)
+
+> **Note**: `ZScoreModel` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [offline](#offline-models) model.
+
+Model is useful for initial testing and for simpler data ([de-trended](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) data without strict [seasonality](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality) and with anomalies of similar magnitude as your "normal" data).
+
 *Parameters specific for vmanomaly*:
 
 * `class` (string) - model class name `"model.zscore.ZscoreModel"` (or `zscore` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support)
@@ -412,7 +464,6 @@ Resulting metrics of the model are described [here](#vmanomaly-output)
 
 *Config Example*
 
-
 ```yaml
 models:
   your_desired_alias_for_a_model:
@@ -420,10 +471,38 @@ models:
     z_threshold: 3.5
 ```
 
+Resulting metrics of the model are described [here](#vmanomaly-output).
+
+### Online Z-score
+
+> **Note**: `OnlineZScoreModel` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [online](#online-models) model.
+
+Online version of existing [Z-score](#z-score) implementation with the same exact behavior and implications. Introduced in [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1150)
+
+*Parameters specific for vmanomaly*:
+
+* `class` (string) - model class name `"model.online.OnlineZscoreModel"` (or `zscore_online` starting from [v1.15.0](../CHANGELOG.md#1130) with class alias support)
+* `z_threshold` (float, optional) - [standard score](https://en.wikipedia.org/wiki/Standard_score) for calculation boundaries and anomaly score. Defaults to `2.5`.
+* `min_n_samples_seen` (int, optional) - the minimum number of samples to be seen (`n_samples_seen_` property) before computing the anomaly score. Otherwise, the **anomaly score will be 0**, as there is not enough data to trust the model's predictions. Defaults to 16.
+
+*Config Example*
+
+```yaml
+models:
+  your_desired_alias_for_a_model:
+    class: "zscore_online"  # or 'model.online.OnlineZscoreModel'
+    z_threshold: 3.5
+    min_n_samples_seen: 128  # i.e. calculate it as full seasonality / data freq
+    provide_series: ['anomaly_score', 'yhat']  # common arg example
+```
 
 Resulting metrics of the model are described [here](#vmanomaly-output).
 
+
 ### [Holt-Winters](https://en.wikipedia.org/wiki/Exponential_smoothing)
+
+> **Note**: `HoltWinters` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [offline](#offline-models) model.
+
 Here we use Holt-Winters Exponential Smoothing implementation from `statsmodels` [library](https://www.statsmodels.org/dev/generated/statsmodels.tsa.holtwinters.ExponentialSmoothing.html). All parameters from this library can be passed to the model.
 
 *Parameters specific for vmanomaly*:
@@ -465,7 +544,11 @@ models:
 
 Resulting metrics of the model are described [here](#vmanomaly-output).
 
+
 ### [MAD (Median Absolute Deviation)](https://en.wikipedia.org/wiki/Median_absolute_deviation)
+
+> **Note**: `MADModel` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [offline](#offline-models) model.
+
 The MAD model is a robust method for anomaly detection that is *less sensitive* to outliers in data compared to standard deviation-based models. It considers a point as an anomaly if the absolute deviation from the median is significantly large.
 
 *Parameters specific for vmanomaly*:
@@ -483,11 +566,44 @@ models:
     threshold: 2.5
 ```
 
+Resulting metrics of the model are described [here](#vmanomaly-output).
+
+
+### Online MAD
+
+> **Note**: `OnlineMADModel` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [online](#online-models) model.
+
+The MAD model is a robust method for anomaly detection that is *less sensitive* to outliers in data compared to standard deviation-based models. It considers a point as an anomaly if the absolute deviation from the median is significantly large. This is the online approximate version, based on [t-digests](https://www.sciencedirect.com/science/article/pii/S2665963820300403) for online quantile estimation. introduced in [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1150)
+
+*Parameters specific for vmanomaly*:
+
+* `class` (string) - model class name `"model.online.OnlineMADModel"` (or `mad_online` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support)
+* `threshold` (float, optional) - The threshold multiplier for the MAD to determine anomalies. Defaults to `2.5`. Higher values will identify fewer points as anomalies.
+* `min_n_samples_seen` (int, optional) - the minimum number of samples to be seen (`n_samples_seen_` property) before computing the anomaly score. Otherwise, the **anomaly score will be 0**, as there is not enough data to trust the model's predictions. Defaults to 16.
+* `compression` (int, optional) - the compression parameter for the T-Digest. Higher values mean higher accuracy but higher memory usage. By default 100.
+
+*Config Example*
+
+
+```yaml
+models:
+  your_desired_alias_for_a_model:
+    class: "mad_online"  # or 'model.online.OnlineMADModel'
+    threshold: 2.5
+    min_n_samples_seen: 128  # i.e. calculate it as full seasonality / data freq
+    compression: 100  # higher values mean higher accuracy but higher memory usage
+    provide_series: ['anomaly_score', 'yhat']  # common arg example
+```
 
 Resulting metrics of the model are described [here](#vmanomaly-output).
 
+
 ### [Rolling Quantile](https://en.wikipedia.org/wiki/Quantile)
 
+> **Note**: `RollingQuantileModel` is [univariate](#univariate-models), [rolling](#rolling-models), [offline](#offline-models) model.
+
+This model is best used on **data with short evolving patterns** (i.e. 10-100 datapoints of particular frequency), as it adapts to changes over a rolling window.
+
 *Parameters specific for vmanomaly*:
 
 * `class` (string) - model class name `"model.rolling_quantile.RollingQuantileModel"` (or `rolling_quantile` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support)
@@ -506,7 +622,55 @@ models:
 
 Resulting metrics of the model are described [here](#vmanomaly-output).
 
+
+### Online Seasonal Quantile
+
+> **Note**: `OnlineSeasonalQuantile` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [online](#online-models) model.
+
+Online (seasonal) quantile utilizes a set of approximate distributions, based on [t-digests](https://www.sciencedirect.com/science/article/pii/S2665963820300403) for online quantile estimation. Introduced in [v1.15.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1150).
+
+Best used on **[de-trended](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#trend) data with strong (possibly multiple) [seasonalities](https://victoriametrics.com/blog/victoriametrics-anomaly-detection-handbook-chapter-1/#seasonality)**. Can act as a (slightly less powerful) replacement to [`ProphetModel`](#prophet).
+
+It uses the `quantiles` triplet to calculate `yhat_lower`, `yhat`, and `yhat_upper` [output](#vmanomaly-output), respectively, for each of the `min_subseasons` sub-intervals contained in `seasonal_interval`. For example, with '4d' + '2h' seasonality patterns (multiple), it will hold and update 24*4 / 2 = 48 consecutive estimates (each 2 hours long).
+
+*Parameters specific for vmanomaly*:
+
+* `class` (string) - model class name `"model.online.OnlineSeasonalQuantile"` (or `quantile_online` starting from [v1.13.0](../CHANGELOG.md#1130) with class alias support)
+* `quantiles` (list[float], optional) - The quantiles to estimate. `yhat_lower`, `yhat`, `yhat_upper` are the quantile order. By default (0.01, 0.5, 0.99).
+* `seasonal_interval` (string, optional) - the interval for the seasonal adjustment. If not set, the model will equal to a simple online quantile model. By default not set.
+* `min_subseason` (str, optional) - the minimum interval to estimate quantiles for. By default None. Note that the minimum interval should be a multiple of the seasonal interval, i.e. if seasonal_interval='2h', then min_subseason='15m' is valid, but '37m' is not.
+* `use_transform` (bool, optional) - whether to apply a log1p(abs(x)) * sign(x) transformation to the data to stabilize internal quantile estimation. By default False.
+* `global_smoothing` (float, optional) - the smoothing parameter for the global quantiles. i.e. the output is a weighted average of the global and seasonal quantiles (if `seasonal_interval` and `min_subseason` args are set). Should be from `[0, 1]` interval, where 0 means no smoothing and 1 means using only global quantile values.
+* `scale` (float, optional) - the scaling factor for the `yhat_lower` and `yhat_upper` quantiles. By default 1.0 (no scaling). if > 1, increases the boundaries [`yhat_lower`, `yhat_upper`] for "non-anomalous" points. Should be > 0.
+* `season_starts_from` (str, optional) - the start date for the seasonal adjustment, as a reference point to start counting the intervals. By default '1970-01-01'.
+* `min_n_samples_seen` (int, optional) - the minimum number of samples to be seen (`n_samples_seen_` property) before computing the anomaly score. Otherwise, the **anomaly score will be 0**, as there is not enough data to trust the model's predictions. Defaults to 16.
+* `compression` (int, optional) - the compression parameter for the T-Digest. Higher values mean higher accuracy but higher memory usage. By default 100.
+
+*Config Example*
+
+Suppose you have a data with strong intraday (hourly) and intraweek (daily) seasonality, and data granularity is '5m' with more than 5% expected outliers present in data. Then you can apply similar config:
+
+```yaml
+models:
+  your_desired_alias_for_a_model:
+    class: "quantile_online"  # or 'model.online.OnlineSeasonalQuantile'
+    quantiles: [0.025, 0.5, 0.975]  # lowered to exclude anomalous edges, can be compensated by `scale` param > 1
+    seasonal_interval: '7d'  # longest seasonality (week, day) = week, starting from `season_starts_from`
+    min_subseason: '1h'  # smallest seasonality (week, day, hour) = hour, will have its own quantile estimates
+    min_n_samples_seen: 288 # 1440 / 5 - at least 1 full day, ideal = 1440 / 5 * 7 - one full week (seasonal_interval)
+    scale: 1.1  # to compensate lowered quantile boundaries with wider intervals
+    season_starts_from: '2024-01-01'  # interval calculation starting point, expecially for uncommon seasonalities like '36h' or '12d'
+    compression: 100  # higher values mean higher accuracy but higher memory usage
+    provide_series: ['anomaly_score', 'yhat']  # common arg example
+```
+
+Resulting metrics of the model are described [here](#vmanomaly-output).
+
+
 ### [Seasonal Trend Decomposition](https://en.wikipedia.org/wiki/Seasonal_adjustment)
+
+> **Note**: `StdModel` is [univariate](#univariate-models), [rolling](#rolling-models), [offline](#offline-models) model.
+
 Here we use Seasonal Decompose implementation from `statsmodels` [library](https://www.statsmodels.org/dev/generated/statsmodels.tsa.seasonal.seasonal_decompose.html). Parameters from this library can be passed to the model. Some parameters are specifically predefined in `vmanomaly` and can't be changed by user(`model`='additive', `two_sided`=False).
 
 *Parameters specific for vmanomaly*:
@@ -536,6 +700,11 @@ Resulting metrics of the model are described [here](#vmanomaly-output).
 
 
 ### [Isolation forest](https://en.wikipedia.org/wiki/Isolation_forest) (Multivariate)
+
+> **Note**: `IsolationForestModel` is [univariate](#univariate-models), [non-rolling](#non-rolling-models), [offline](#offline-models) model.
+
+> **Note**: `IsolationForestMultivariateModel` is [multivariate](#multivariate-models), [non-rolling](#non-rolling-models), [offline](#offline-models) model.
+
 Detects anomalies using binary trees. The algorithm has a linear time complexity and a low memory requirement, which works well with high-volume data. It can be used on both univariate and multivariate data, but it is more effective in multivariate case.
 
 **Important**: Be aware of [the curse of dimensionality](https://en.wikipedia.org/wiki/Curse_of_dimensionality). Don't use single multivariate model if you expect your queries to return many time series of less datapoints that the number of metrics. In such case it is hard for a model to learn meaningful dependencies from too sparse data hypercube.
@@ -616,7 +785,7 @@ Here in this guide, we will
 - Define VictoriaMetrics Anomaly Detection config file to use our custom model
 - Run service
 
-**Note**: The file containing the model should be written in [Python language](https://www.python.org/) (3.11+)
+> **Note**: The file containing the model should be written in [Python language](https://www.python.org/) (3.11+)
 
 ### 1. Custom model
 
@@ -626,7 +795,7 @@ We'll create `custom_model.py` file with `CustomModel` class that will inherit f
 In the `CustomModel` class, the following methods are required: - `__init__`, `fit`, `infer`, `serialize` and `deserialize`:
 * `__init__` method should initiate parameters for the model.
 
-  **Note**: if your model relies on configs that have `arg` [key-value pair argument](./models.md#section-overview), do not forget to use Python's `**kwargs` in method's signature and to explicitly call
+  **Note**: if your model relies on configs that have `arg` [key-value pair argument, like Prophet](#prophet), do not forget to use Python's `**kwargs` in method's signature and to explicitly call
 
   ```python 
   super().__init__(**kwargs)
@@ -662,7 +831,10 @@ class CustomModel(Model):
   """
   # by default, each `Model` will be created as a univariate one
   # uncomment line below for it to be of multivariate type
-  #`is_multivariate = True`
+  # is_multivariate = True
+  # by default, each `Model` will be created as offline
+  # uncomment line below for it to be of type online
+  # is_online = True
   
   def __init__(self, percentage: float = 0.95, **kwargs):
     super().__init__(**kwargs)
diff --git a/docs/sd_configs.md b/docs/sd_configs.md
index 60f2d1db9..5772fb985 100644
--- a/docs/sd_configs.md
+++ b/docs/sd_configs.md
@@ -1768,6 +1768,9 @@ scrape_configs:
   # If more than this number of samples are present after metric relabeling
   # the entire scrape will be treated as failed.
   # By default, the limit is disabled.
+  # The sample_limit can be set on a per-target basis by specifying `__sample_limit__`
+  # label during target relabeling phase.
+  # See https://docs.victoriametrics.com/vmagent/#relabeling
   #
   # sample_limit: <int>
 
diff --git a/docs/stream-aggregation.md b/docs/stream-aggregation.md
index 042e3e52b..4530358f5 100644
--- a/docs/stream-aggregation.md
+++ b/docs/stream-aggregation.md
@@ -134,7 +134,7 @@ just after the restart of `vmagent` or single-node VictoriaMetrics. This can be
 - The `-streamAggr.ignoreFirstIntervals=N` command-line flag at `vmagent` and single-node VictoriaMetrics. This flag instructs skipping the first `N`
   [aggregation intervals](#stream-aggregation-config) just after the restart across all the [configured stream aggregation configs](#configuration).
 
-  The `-remoteWrite.streamAggr.ignoreFirstIntervals=N` command-line flag can be specified individually per each `-remoteWrite.url` at [vmagent](https://docs.victoriametrics.com/vmagent/).
+  The `-remoteWrite.streamAggr.ignoreFirstIntervals` command-line flag can be specified individually per each `-remoteWrite.url` at [vmagent](https://docs.victoriametrics.com/vmagent/).
 
 - The `ignore_first_intervals: N` option at the particular [aggregation config](#stream-aggregation-config).
 
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 517c85b2d..01f308925 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -2211,7 +2211,7 @@ See the docs at https://docs.victoriametrics.com/vmagent/ .
      An optional list of labels to drop from samples before stream de-duplication and aggregation with -remoteWrite.streamAggr.config and -remoteWrite.streamAggr.dedupInterval at the corresponding -remoteWrite.url. 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.
-  -remoteWrite.streamAggr.ignoreFirstIntervals int
+  -remoteWrite.streamAggr.ignoreFirstIntervals array
      Number of aggregation intervals to skip after the start for the corresponding -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. Increase this value if you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving bufferred 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 at the corresponding -remoteWrite.url. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index 7c94f6ae5..0b4af38d9 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -1187,6 +1187,16 @@ func (swc *scrapeWorkConfig) getScrapeWork(target string, extraLabels, metaLabel
 		}
 		seriesLimit = n
 	}
+	// Read sample_limit option from __sample_limit__ label.
+	// See https://docs.victoriametrics.com/vmagent/#automatically-generated-metrics
+	sampleLimit := swc.sampleLimit
+	if s := labels.Get("__sample_limit__"); len(s) > 0 {
+		n, err := strconv.Atoi(s)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse __sample_limit__=%q: %w", s, err)
+		}
+		sampleLimit = n
+	}
 	// Read stream_parse option from __stream_parse__ label.
 	// See https://docs.victoriametrics.com/vmagent/#stream-parsing-mode
 	streamParse := swc.streamParse
@@ -1232,7 +1242,7 @@ func (swc *scrapeWorkConfig) getScrapeWork(target string, extraLabels, metaLabel
 		AuthConfig:           swc.authConfig,
 		RelabelConfigs:       swc.relabelConfigs,
 		MetricRelabelConfigs: swc.metricRelabelConfigs,
-		SampleLimit:          swc.sampleLimit,
+		SampleLimit:          sampleLimit,
 		DisableCompression:   swc.disableCompression,
 		DisableKeepAlive:     swc.disableKeepAlive,
 		StreamParse:          streamParse,
diff --git a/lib/streamaggr/avg.go b/lib/streamaggr/avg.go
index 522c2fcb5..32ab569ac 100644
--- a/lib/streamaggr/avg.go
+++ b/lib/streamaggr/avg.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // avgAggrState calculates output=avg, e.g. the average value over input samples.
@@ -61,26 +60,21 @@ func (as *avgAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *avgAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *avgAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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)
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "avg", currentTimeMsec, avg)
+		ctx.appendSeries(key, "avg", avg)
 		return true
 	})
 }
diff --git a/lib/streamaggr/count_samples.go b/lib/streamaggr/count_samples.go
index 6a05955cc..ca2bc709e 100644
--- a/lib/streamaggr/count_samples.go
+++ b/lib/streamaggr/count_samples.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // countSamplesAggrState calculates output=count_samples, e.g. the count of input samples.
@@ -58,26 +57,21 @@ func (as *countSamplesAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *countSamplesAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *countSamplesAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "count_samples", currentTimeMsec, float64(n))
+		ctx.appendSeries(key, "count_samples", float64(n))
 		return true
 	})
 }
diff --git a/lib/streamaggr/count_series.go b/lib/streamaggr/count_series.go
index f1037c801..816581464 100644
--- a/lib/streamaggr/count_series.go
+++ b/lib/streamaggr/count_series.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/cespare/xxhash/v2"
 )
 
@@ -67,26 +66,21 @@ func (as *countSeriesAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *countSeriesAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *countSeriesAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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 := len(sv.m)
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "count_series", currentTimeMsec, float64(n))
+		ctx.appendSeries(key, "count_series", float64(n))
 		return true
 	})
 }
diff --git a/lib/streamaggr/histogram_bucket.go b/lib/streamaggr/histogram_bucket.go
index 92982dcec..fdf7b3fad 100644
--- a/lib/streamaggr/histogram_bucket.go
+++ b/lib/streamaggr/histogram_bucket.go
@@ -86,9 +86,8 @@ func (as *histogramBucketAggrState) removeOldEntries(currentTime uint64) {
 	})
 }
 
-func (as *histogramBucketAggrState) flushState(ctx *flushCtx, _ bool) {
+func (as *histogramBucketAggrState) flushState(ctx *flushCtx) {
 	currentTime := fasttime.UnixTimestamp()
-	currentTimeMsec := int64(currentTime) * 1000
 
 	as.removeOldEntries(currentTime)
 
@@ -99,7 +98,7 @@ func (as *histogramBucketAggrState) flushState(ctx *flushCtx, _ bool) {
 		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, "histogram_bucket", float64(count), "vmrange", vmrange)
 			})
 		}
 		sv.mu.Unlock()
diff --git a/lib/streamaggr/last.go b/lib/streamaggr/last.go
index eaa803e83..0ae4b9b8c 100644
--- a/lib/streamaggr/last.go
+++ b/lib/streamaggr/last.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // lastAggrState calculates output=last, e.g. the last value over input samples.
@@ -63,26 +62,21 @@ func (as *lastAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *lastAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *lastAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "last", currentTimeMsec, last)
+		ctx.appendSeries(key, "last", last)
 		return true
 	})
 }
diff --git a/lib/streamaggr/max.go b/lib/streamaggr/max.go
index bbbb3c83d..9197d3add 100644
--- a/lib/streamaggr/max.go
+++ b/lib/streamaggr/max.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // maxAggrState calculates output=max, e.g. the maximum value over input samples.
@@ -60,26 +59,21 @@ func (as *maxAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *maxAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *maxAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "max", currentTimeMsec, max)
+		ctx.appendSeries(key, "max", max)
 		return true
 	})
 }
diff --git a/lib/streamaggr/min.go b/lib/streamaggr/min.go
index 8970d41a7..308f259c7 100644
--- a/lib/streamaggr/min.go
+++ b/lib/streamaggr/min.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // minAggrState calculates output=min, e.g. the minimum value over input samples.
@@ -60,25 +59,20 @@ func (as *minAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *minAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *minAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "min", currentTimeMsec, min)
+		ctx.appendSeries(key, "min", min)
 		return true
 	})
 }
diff --git a/lib/streamaggr/quantiles.go b/lib/streamaggr/quantiles.go
index ee187d36b..00fefe877 100644
--- a/lib/streamaggr/quantiles.go
+++ b/lib/streamaggr/quantiles.go
@@ -5,7 +5,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/valyala/histogram"
 )
 
@@ -64,33 +63,28 @@ func (as *quantilesAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *quantilesAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *quantilesAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	phis := as.phis
 	var quantiles []float64
 	var b []byte
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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)
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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, "quantiles", quantile, "quantile", phiStr)
 		}
 		return true
 	})
diff --git a/lib/streamaggr/rate.go b/lib/streamaggr/rate.go
index 55bd393db..1c3f2fb5c 100644
--- a/lib/streamaggr/rate.go
+++ b/lib/streamaggr/rate.go
@@ -105,9 +105,8 @@ func (as *rateAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *rateAggrState) flushState(ctx *flushCtx, resetState bool) {
+func (as *rateAggrState) flushState(ctx *flushCtx) {
 	currentTime := fasttime.UnixTimestamp()
-	currentTimeMsec := int64(currentTime) * 1000
 
 	suffix := as.getSuffix()
 
@@ -127,11 +126,9 @@ func (as *rateAggrState) flushState(ctx *flushCtx, resetState bool) {
 				sumRate += lv.increase / d
 				countSeries++
 			}
-			if resetState {
-				lv.prevTimestamp = lv.timestamp
-				lv.increase = 0
-				lvs[k1] = lv
-			}
+			lv.prevTimestamp = lv.timestamp
+			lv.increase = 0
+			lvs[k1] = lv
 		}
 		deleted := sv.deleted
 		sv.mu.Unlock()
@@ -147,7 +144,7 @@ func (as *rateAggrState) flushState(ctx *flushCtx, resetState bool) {
 		}
 
 		key := k.(string)
-		ctx.appendSeries(key, suffix, currentTimeMsec, result)
+		ctx.appendSeries(key, suffix, result)
 		return true
 	})
 }
diff --git a/lib/streamaggr/stddev.go b/lib/streamaggr/stddev.go
index 26ea2db8c..053e0f209 100644
--- a/lib/streamaggr/stddev.go
+++ b/lib/streamaggr/stddev.go
@@ -5,7 +5,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // stddevAggrState calculates output=stddev, e.g. the average value over input samples.
@@ -61,26 +60,21 @@ func (as *stddevAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *stddevAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *stddevAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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)
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "stddev", currentTimeMsec, stddev)
+		ctx.appendSeries(key, "stddev", stddev)
 		return true
 	})
 }
diff --git a/lib/streamaggr/stdvar.go b/lib/streamaggr/stdvar.go
index 35a15097e..8170f227a 100644
--- a/lib/streamaggr/stdvar.go
+++ b/lib/streamaggr/stdvar.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // stdvarAggrState calculates output=stdvar, e.g. the average value over input samples.
@@ -60,26 +59,21 @@ func (as *stdvarAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *stdvarAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *stdvarAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "stdvar", currentTimeMsec, stdvar)
+		ctx.appendSeries(key, "stdvar", stdvar)
 		return true
 	})
 }
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index c8571b9d7..6fc674e84 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -422,13 +422,7 @@ type aggrState interface {
 	pushSamples(samples []pushSample)
 
 	// flushState must flush aggrState data to ctx.
-	//
-	// if resetState is true, then aggrState must be reset after flushing the data to ctx,
-	// otherwise the aggrState data must be kept unchanged.
-	//
-	// The resetState is set to false only in the benchmark, which measures flushState() performance
-	// over the same aggrState.
-	flushState(ctx *flushCtx, resetState bool)
+	flushState(ctx *flushCtx)
 }
 
 // PushFunc is called by Aggregators when it needs to push its state to metrics storage
@@ -733,11 +727,14 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 		}
 	}
 
+	var flushTimeMsec int64
 	tickerWait := func(t *time.Ticker) bool {
 		select {
 		case <-a.stopCh:
+			flushTimeMsec = time.Now().UnixMilli()
 			return false
-		case <-t.C:
+		case ct := <-t.C:
+			flushTimeMsec = ct.UnixMilli()
 			return true
 		}
 	}
@@ -748,16 +745,16 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 		defer t.Stop()
 
 		if alignFlushToInterval && skipIncompleteFlush {
-			a.flush(nil)
+			a.flush(nil, 0)
 			ignoreFirstIntervals--
 		}
 
 		for tickerWait(t) {
 			if ignoreFirstIntervals > 0 {
-				a.flush(nil)
+				a.flush(nil, 0)
 				ignoreFirstIntervals--
 			} else {
-				a.flush(pushFunc)
+				a.flush(pushFunc, flushTimeMsec)
 			}
 
 			if alignFlushToInterval {
@@ -781,14 +778,14 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 			if ct.After(flushDeadline) {
 				// It is time to flush the aggregated state
 				if alignFlushToInterval && skipIncompleteFlush && !isSkippedFirstFlush {
-					a.flush(nil)
+					a.flush(nil, 0)
 					ignoreFirstIntervals--
 					isSkippedFirstFlush = true
 				} else if ignoreFirstIntervals > 0 {
-					a.flush(nil)
+					a.flush(nil, 0)
 					ignoreFirstIntervals--
 				} else {
-					a.flush(pushFunc)
+					a.flush(pushFunc, flushTimeMsec)
 				}
 				for ct.After(flushDeadline) {
 					flushDeadline = flushDeadline.Add(a.interval)
@@ -806,7 +803,7 @@ func (a *aggregator) runFlusher(pushFunc PushFunc, alignFlushToInterval, skipInc
 
 	if !skipIncompleteFlush && ignoreFirstIntervals <= 0 {
 		a.dedupFlush()
-		a.flush(pushFunc)
+		a.flush(pushFunc, flushTimeMsec)
 	}
 }
 
@@ -833,17 +830,13 @@ func (a *aggregator) dedupFlush() {
 // flush flushes aggregator state to pushFunc.
 //
 // If pushFunc is nil, then the aggregator state is just reset.
-func (a *aggregator) flush(pushFunc PushFunc) {
-	a.flushInternal(pushFunc, true)
-}
-
-func (a *aggregator) flushInternal(pushFunc PushFunc, resetState bool) {
+func (a *aggregator) flush(pushFunc PushFunc, flushTimeMsec int64) {
 	startTime := time.Now()
 
 	// Update minTimestamp before flushing samples to the storage,
 	// since the flush durtion can be quite long.
 	// This should prevent from dropping samples with old timestamps when the flush takes long time.
-	a.minTimestamp.Store(startTime.UnixMilli() - 5_000)
+	a.minTimestamp.Store(flushTimeMsec - 5_000)
 
 	var wg sync.WaitGroup
 	for i := range a.aggrOutputs {
@@ -856,8 +849,8 @@ func (a *aggregator) flushInternal(pushFunc PushFunc, resetState bool) {
 				wg.Done()
 			}()
 
-			ctx := getFlushCtx(a, ao, pushFunc)
-			ao.as.flushState(ctx, resetState)
+			ctx := getFlushCtx(a, ao, pushFunc, flushTimeMsec)
+			ao.as.flushState(ctx)
 			ctx.flushSeries()
 			putFlushCtx(ctx)
 		}(ao)
@@ -1073,7 +1066,7 @@ func getInputOutputLabels(dstInput, dstOutput, labels []prompbmarshal.Label, by,
 	return dstInput, dstOutput
 }
 
-func getFlushCtx(a *aggregator, ao *aggrOutput, pushFunc PushFunc) *flushCtx {
+func getFlushCtx(a *aggregator, ao *aggrOutput, pushFunc PushFunc, flushTimestamp int64) *flushCtx {
 	v := flushCtxPool.Get()
 	if v == nil {
 		v = &flushCtx{}
@@ -1082,6 +1075,7 @@ func getFlushCtx(a *aggregator, ao *aggrOutput, pushFunc PushFunc) *flushCtx {
 	ctx.a = a
 	ctx.ao = ao
 	ctx.pushFunc = pushFunc
+	ctx.flushTimestamp = flushTimestamp
 	return ctx
 }
 
@@ -1093,9 +1087,10 @@ func putFlushCtx(ctx *flushCtx) {
 var flushCtxPool sync.Pool
 
 type flushCtx struct {
-	a        *aggregator
-	ao       *aggrOutput
-	pushFunc PushFunc
+	a              *aggregator
+	ao             *aggrOutput
+	pushFunc       PushFunc
+	flushTimestamp int64
 
 	tss     []prompbmarshal.TimeSeries
 	labels  []prompbmarshal.Label
@@ -1106,6 +1101,7 @@ func (ctx *flushCtx) reset() {
 	ctx.a = nil
 	ctx.ao = nil
 	ctx.pushFunc = nil
+	ctx.flushTimestamp = 0
 	ctx.resetSeries()
 }
 
@@ -1161,7 +1157,7 @@ func (ctx *flushCtx) flushSeries() {
 	promutils.PutLabels(auxLabels)
 }
 
-func (ctx *flushCtx) appendSeries(key, suffix string, timestamp int64, value float64) {
+func (ctx *flushCtx) appendSeries(key, suffix string, value float64) {
 	labelsLen := len(ctx.labels)
 	samplesLen := len(ctx.samples)
 	ctx.labels = decompressLabels(ctx.labels, key)
@@ -1169,7 +1165,7 @@ func (ctx *flushCtx) appendSeries(key, suffix string, timestamp int64, value flo
 		ctx.labels = addMetricSuffix(ctx.labels, labelsLen, ctx.a.suffix, suffix)
 	}
 	ctx.samples = append(ctx.samples, prompbmarshal.Sample{
-		Timestamp: timestamp,
+		Timestamp: ctx.flushTimestamp,
 		Value:     value,
 	})
 	ctx.tss = append(ctx.tss, prompbmarshal.TimeSeries{
@@ -1183,7 +1179,7 @@ func (ctx *flushCtx) appendSeries(key, suffix string, timestamp int64, value flo
 	}
 }
 
-func (ctx *flushCtx) appendSeriesWithExtraLabel(key, suffix string, timestamp int64, value float64, extraName, extraValue string) {
+func (ctx *flushCtx) appendSeriesWithExtraLabel(key, suffix string, value float64, extraName, extraValue string) {
 	labelsLen := len(ctx.labels)
 	samplesLen := len(ctx.samples)
 	ctx.labels = decompressLabels(ctx.labels, key)
@@ -1195,7 +1191,7 @@ func (ctx *flushCtx) appendSeriesWithExtraLabel(key, suffix string, timestamp in
 		Value: extraValue,
 	})
 	ctx.samples = append(ctx.samples, prompbmarshal.Sample{
-		Timestamp: timestamp,
+		Timestamp: ctx.flushTimestamp,
 		Value:     value,
 	})
 	ctx.tss = append(ctx.tss, prompbmarshal.TimeSeries{
diff --git a/lib/streamaggr/sum_samples.go b/lib/streamaggr/sum_samples.go
index 685e782c5..947239bb3 100644
--- a/lib/streamaggr/sum_samples.go
+++ b/lib/streamaggr/sum_samples.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // sumSamplesAggrState calculates output=sum_samples, e.g. the sum over input samples.
@@ -58,26 +57,21 @@ func (as *sumSamplesAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *sumSamplesAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *sumSamplesAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// 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
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "sum_samples", currentTimeMsec, sum)
+		ctx.appendSeries(key, "sum_samples", sum)
 		return true
 	})
 }
diff --git a/lib/streamaggr/total.go b/lib/streamaggr/total.go
index fda700b4c..e905e4531 100644
--- a/lib/streamaggr/total.go
+++ b/lib/streamaggr/total.go
@@ -115,9 +115,8 @@ func (as *totalAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *totalAggrState) flushState(ctx *flushCtx, resetState bool) {
+func (as *totalAggrState) flushState(ctx *flushCtx) {
 	currentTime := fasttime.UnixTimestamp()
-	currentTimeMsec := int64(currentTime) * 1000
 
 	suffix := as.getSuffix()
 
@@ -129,20 +128,18 @@ func (as *totalAggrState) flushState(ctx *flushCtx, resetState bool) {
 
 		sv.mu.Lock()
 		total := sv.total
-		if resetState {
-			if as.resetTotalOnFlush {
-				sv.total = 0
-			} else if math.Abs(sv.total) >= (1 << 53) {
-				// It is time to reset the entry, since it starts losing float64 precision
-				sv.total = 0
-			}
+		if as.resetTotalOnFlush {
+			sv.total = 0
+		} else 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, suffix, currentTimeMsec, total)
+			ctx.appendSeries(key, suffix, total)
 		}
 		return true
 	})
diff --git a/lib/streamaggr/unique_samples.go b/lib/streamaggr/unique_samples.go
index 9cb63c636..93145397c 100644
--- a/lib/streamaggr/unique_samples.go
+++ b/lib/streamaggr/unique_samples.go
@@ -4,7 +4,6 @@ import (
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 )
 
 // uniqueSamplesAggrState calculates output=unique_samples, e.g. the number of unique sample values.
@@ -62,26 +61,21 @@ func (as *uniqueSamplesAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *uniqueSamplesAggrState) flushState(ctx *flushCtx, resetState bool) {
-	currentTimeMsec := int64(fasttime.UnixTimestamp()) * 1000
+func (as *uniqueSamplesAggrState) flushState(ctx *flushCtx) {
 	m := &as.m
 	m.Range(func(k, v any) bool {
-		if resetState {
-			// Atomically delete the entry from the map, so new entry is created for the next flush.
-			m.Delete(k)
-		}
+		// Atomically delete the entry from the map, so new entry is created for the next flush.
+		m.Delete(k)
 
 		sv := v.(*uniqueSamplesStateValue)
 		sv.mu.Lock()
 		n := len(sv.m)
-		if resetState {
-			// Mark the entry as deleted, so it won't be updated anymore by concurrent pushSample() calls.
-			sv.deleted = true
-		}
+		// 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)
-		ctx.appendSeries(key, "unique_samples", currentTimeMsec, float64(n))
+		ctx.appendSeries(key, "unique_samples", float64(n))
 		return true
 	})
 }


Test patch:
diff --git a/lib/promscrape/config_test.go b/lib/promscrape/config_test.go
index 369053e18..29930d380 100644
--- a/lib/promscrape/config_test.go
+++ b/lib/promscrape/config_test.go
@@ -1150,6 +1150,8 @@ scrape_configs:
         replacement: 127.0.0.1:9116  # The SNMP exporter's real hostname:port.
       - target_label: __series_limit__
         replacement: 1234
+      - target_label: __sample_limit__
+        replacement: 5678
       - target_label: __stream_parse__
         replacement: true
 `, []*ScrapeWork{
@@ -1162,7 +1164,7 @@ scrape_configs:
 				"instance": "192.168.1.2",
 				"job":      "snmp",
 			}),
-			SampleLimit:         100,
+			SampleLimit:         5678,
 			DisableKeepAlive:    true,
 			DisableCompression:  true,
 			StreamParse:         true,
diff --git a/lib/streamaggr/streamaggr_timing_test.go b/lib/streamaggr/streamaggr_timing_test.go
index bb0d98afa..5901e56ad 100644
--- a/lib/streamaggr/streamaggr_timing_test.go
+++ b/lib/streamaggr/streamaggr_timing_test.go
@@ -39,22 +39,6 @@ func BenchmarkAggregatorsPush(b *testing.B) {
 	}
 }
 
-func BenchmarkAggregatorsFlushInternalSerial(b *testing.B) {
-	pushFunc := func(_ []prompbmarshal.TimeSeries) {}
-	a := newBenchAggregators(benchOutputs, pushFunc)
-	defer a.MustStop()
-	_ = a.Push(benchSeries, nil)
-
-	b.ResetTimer()
-	b.ReportAllocs()
-	b.SetBytes(int64(len(benchSeries) * len(benchOutputs)))
-	for i := 0; i < b.N; i++ {
-		for _, aggr := range a.as {
-			aggr.flushInternal(pushFunc, false)
-		}
-	}
-}
-
 func benchmarkAggregatorsPush(b *testing.B, output string) {
 	pushFunc := func(_ []prompbmarshal.TimeSeries) {}
 	a := newBenchAggregators([]string{output}, pushFunc)