Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6546 (2024-06-28 07:42:53)

vmui: increase max query tab from 4 to 10

Loori-R: @hagen1778, could you merge it, please?

Patch:
diff --git a/Makefile b/Makefile
index db177a3a0..5bacae5f0 100644
--- a/Makefile
+++ b/Makefile
@@ -20,7 +20,6 @@ include app/*/Makefile
 include docs/Makefile
 include deployment/*/Makefile
 include dashboards/Makefile
-include snap/local/Makefile
 include package/release/Makefile
 
 all: \
diff --git a/README.md b/README.md
index bbc59b61b..853b21a94 100644
--- a/README.md
+++ b/README.md
@@ -15,8 +15,7 @@ VictoriaMetrics is a fast, cost-effective and scalable monitoring solution and t
 See [case studies for VictoriaMetrics](https://docs.victoriametrics.com/casestudies/).
 
 VictoriaMetrics is available in [binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest),
-[Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/), [Snap packages](https://snapcraft.io/victoriametrics)
-and [source code](https://github.com/VictoriaMetrics/VictoriaMetrics). 
+[Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/) and [source code](https://github.com/VictoriaMetrics/VictoriaMetrics). 
 
 Documentation for the cluster version of VictoriaMetrics is available [here](https://docs.victoriametrics.com/cluster-victoriametrics/).
 
@@ -139,7 +138,6 @@ VictoriaMetrics can also be installed via these installation methods:
 * [Ansible role for installing cluster VictoriaMetrics (by VictoriaMetrics)](https://github.com/VictoriaMetrics/ansible-playbooks).
 * [Ansible role for installing cluster VictoriaMetrics (by community)](https://github.com/Slapper/ansible-victoriametrics-cluster-role).
 * [Ansible role for installing single-node VictoriaMetrics (by community)](https://github.com/dreamteam-gg/ansible-victoriametrics-role).
-* [Snap package for VictoriaMetrics](https://snapcraft.io/victoriametrics).
 
 ### How to start VictoriaMetrics
 
@@ -182,24 +180,7 @@ Additionally, all the VictoriaMetrics components allow setting flag values via e
 
 ### Configuration with snap package
 
-Snap package for VictoriaMetrics is available [here](https://snapcraft.io/victoriametrics).
-
-Command-line flags for Snap package can be set with following command:
-
-```sh
-echo 'FLAGS="-selfScrapeInterval=10s -search.logSlowQueryDuration=20s"' > $SNAP_DATA/var/snap/victoriametrics/current/extra_flags
-snap restart victoriametrics
-```
-
-Do not change value for `-storageDataPath` flag, because snap package has limited access to host filesystem.
-
-Changing scrape configuration is possible with text editor:
-
-```sh
-vi $SNAP_DATA/var/snap/victoriametrics/current/etc/victoriametrics-scrape-config.yaml
-```
-
-After changes were made, trigger config re-read with the command `curl 127.0.0.1:8428/-/reload`.
+Snap packages for VictoriaMetrics are supported by community and are available at [https://snapcraft.io/victoriametrics](https://snapcraft.io/victoriametrics).
 
 ### Running as Windows service
 
@@ -1205,7 +1186,7 @@ More details may be found [here](https://github.com/VictoriaMetrics/VictoriaMetr
 ## Setting up service
 
 Read [instructions](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/43) on how to set up VictoriaMetrics
-as a service for your OS. A [snap package](https://snapcraft.io/victoriametrics) is available for Ubuntu.
+as a service for your OS.
 
 ## How to work with snapshots
 
diff --git a/app/vlinsert/syslog/syslog.go b/app/vlinsert/syslog/syslog.go
index 0a4f7cb03..935c6b7ee 100644
--- a/app/vlinsert/syslog/syslog.go
+++ b/app/vlinsert/syslog/syslog.go
@@ -405,6 +405,7 @@ func (slr *syslogLineReader) nextLine() bool {
 		return false
 	}
 
+again:
 	prefix, err := slr.br.ReadSlice(' ')
 	if err != nil {
 		if err != io.EOF {
@@ -420,6 +421,10 @@ func (slr *syslogLineReader) nextLine() bool {
 	for len(prefix) > 0 && prefix[0] == '\n' {
 		prefix = prefix[1:]
 	}
+	if len(prefix) == 0 {
+		// An empty prefix or a prefix with empty lines - try reading yet another prefix.
+		goto again
+	}
 
 	if prefix[0] >= '0' && prefix[0] <= '9' {
 		// This is octet-counting method. See https://www.ietf.org/archive/id/draft-gerhards-syslog-plain-tcp-07.html#msgxfer
diff --git a/app/vlselect/logsql/logsql.go b/app/vlselect/logsql/logsql.go
index b64c5755e..b4920e3b3 100644
--- a/app/vlselect/logsql/logsql.go
+++ b/app/vlselect/logsql/logsql.go
@@ -450,7 +450,7 @@ type logRow struct {
 }
 
 func sortLogRows(rows []logRow) {
-	sort.Slice(rows, func(i, j int) bool {
+	sort.SliceStable(rows, func(i, j int) bool {
 		return rows[i].timestamp < rows[j].timestamp
 	})
 }
diff --git a/app/vmagent/remotewrite/remotewrite.go b/app/vmagent/remotewrite/remotewrite.go
index a04361ffb..f79d4b42f 100644
--- a/app/vmagent/remotewrite/remotewrite.go
+++ b/app/vmagent/remotewrite/remotewrite.go
@@ -211,7 +211,7 @@ func Init() {
 		}
 		sasGlobal.Store(sas)
 	} else if sasOpts.DedupInterval > 0 {
-		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesDropFailed, sasOpts.DedupInterval, sasOpts.DropInputLabels)
+		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesDropFailed, sasOpts.DedupInterval, sasOpts.DropInputLabels, sasOpts.Alias)
 	}
 
 	if len(*remoteWriteURLs) > 0 {
@@ -840,7 +840,7 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in
 		metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_streamaggr_config_reload_successful{path=%q}`, sasFile)).Set(1)
 		metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_streamaggr_config_reload_success_timestamp_seconds{path=%q}`, sasFile)).Set(fasttime.UnixTimestamp())
 	} else if sasOpts.DedupInterval > 0 {
-		rwctx.deduplicator = streamaggr.NewDeduplicator(rwctx.pushInternalTrackDropped, sasOpts.DedupInterval, sasOpts.DropInputLabels)
+		rwctx.deduplicator = streamaggr.NewDeduplicator(rwctx.pushInternalTrackDropped, sasOpts.DedupInterval, sasOpts.DropInputLabels, sasOpts.Alias)
 	}
 
 	return rwctx
diff --git a/app/vmagent/remotewrite/streamaggr.go b/app/vmagent/remotewrite/streamaggr.go
index 454d14bf8..b8ea11a16 100644
--- a/app/vmagent/remotewrite/streamaggr.go
+++ b/app/vmagent/remotewrite/streamaggr.go
@@ -125,28 +125,35 @@ func reloadStreamAggrConfig(idx int, pushFunc streamaggr.PushFunc) {
 	metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_streamaggr_config_reload_success_timestamp_seconds{path=%q}`, path)).Set(fasttime.UnixTimestamp())
 }
 
-func getStreamAggrOpts(idx int) (string, *streamaggr.Options) {
+func getStreamAggrOpts(idx int) (string, streamaggr.Options) {
 	if idx < 0 {
-		return *streamAggrGlobalConfig, &streamaggr.Options{
+		return *streamAggrGlobalConfig, streamaggr.Options{
 			DedupInterval:        streamAggrGlobalDedupInterval.Duration(),
 			DropInputLabels:      *streamAggrGlobalDropInputLabels,
 			IgnoreOldSamples:     *streamAggrGlobalIgnoreOldSamples,
 			IgnoreFirstIntervals: *streamAggrGlobalIgnoreFirstIntervals,
+			Alias:                "global",
 		}
 	}
+	url := fmt.Sprintf("%d:secret-url", idx+1)
+	if *showRemoteWriteURL {
+		url = fmt.Sprintf("%d:%s", idx+1, remoteWriteURLs.GetOptionalArg(idx))
+	}
 	opts := streamaggr.Options{
 		DedupInterval:        streamAggrDedupInterval.GetOptionalArg(idx),
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     streamAggrIgnoreOldSamples.GetOptionalArg(idx),
 		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
+		Alias:                url,
 	}
+
 	if len(*streamAggrConfig) == 0 {
-		return "", &opts
+		return "", opts
 	}
-	return streamAggrConfig.GetOptionalArg(idx), &opts
+	return streamAggrConfig.GetOptionalArg(idx), opts
 }
 
-func newStreamAggrConfigWithOpts(pushFunc streamaggr.PushFunc, path string, opts *streamaggr.Options) (*streamaggr.Aggregators, error) {
+func newStreamAggrConfigWithOpts(pushFunc streamaggr.PushFunc, path string, opts streamaggr.Options) (*streamaggr.Aggregators, error) {
 	if len(path) == 0 {
 		// Skip empty stream aggregation config.
 		return nil, nil
diff --git a/app/vminsert/common/streamaggr.go b/app/vminsert/common/streamaggr.go
index 39b4478c9..c5ec26ce9 100644
--- a/app/vminsert/common/streamaggr.go
+++ b/app/vminsert/common/streamaggr.go
@@ -57,11 +57,12 @@ func CheckStreamAggrConfig() error {
 		return nil
 	}
 	pushNoop := func(_ []prompbmarshal.TimeSeries) {}
-	opts := &streamaggr.Options{
+	opts := streamaggr.Options{
 		DedupInterval:        *streamAggrDedupInterval,
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
+		Alias:                "global",
 	}
 	sas, err := streamaggr.LoadFromFile(*streamAggrConfig, pushNoop, opts)
 	if err != nil {
@@ -81,21 +82,23 @@ func HasStreamAggrConfigured() bool {
 // MustStopStreamAggr must be called when stream aggr is no longer needed.
 func InitStreamAggr() {
 	saCfgReloaderStopCh = make(chan struct{})
+	rwctx := "global"
 
 	if *streamAggrConfig == "" {
 		if *streamAggrDedupInterval > 0 {
-			deduplicator = streamaggr.NewDeduplicator(pushAggregateSeries, *streamAggrDedupInterval, *streamAggrDropInputLabels)
+			deduplicator = streamaggr.NewDeduplicator(pushAggregateSeries, *streamAggrDedupInterval, *streamAggrDropInputLabels, rwctx)
 		}
 		return
 	}
 
 	sighupCh := procutil.NewSighupChan()
 
-	opts := &streamaggr.Options{
+	opts := streamaggr.Options{
 		DedupInterval:        *streamAggrDedupInterval,
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
+		Alias:                rwctx,
 	}
 	sas, err := streamaggr.LoadFromFile(*streamAggrConfig, pushAggregateSeries, opts)
 	if err != nil {
@@ -125,11 +128,12 @@ func reloadStreamAggrConfig() {
 	logger.Infof("reloading -streamAggr.config=%q", *streamAggrConfig)
 	saCfgReloads.Inc()
 
-	opts := &streamaggr.Options{
+	opts := streamaggr.Options{
 		DedupInterval:        *streamAggrDedupInterval,
 		DropInputLabels:      *streamAggrDropInputLabels,
 		IgnoreOldSamples:     *streamAggrIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrIgnoreFirstIntervals,
+		Alias:                "global",
 	}
 	sasNew, err := streamaggr.LoadFromFile(*streamAggrConfig, pushAggregateSeries, opts)
 	if err != nil {
diff --git a/app/vmui/packages/vmui/src/constants/graph.ts b/app/vmui/packages/vmui/src/constants/graph.ts
index 6c7a8c46c..004026011 100644
--- a/app/vmui/packages/vmui/src/constants/graph.ts
+++ b/app/vmui/packages/vmui/src/constants/graph.ts
@@ -1,6 +1,6 @@
 import { GraphSize, SeriesItemStatsFormatted } from "../types";
 
-export const MAX_QUERY_FIELDS = 4;
+export const MAX_QUERY_FIELDS = 10;
 export const MAX_QUERIES_HISTORY = 25;
 export const DEFAULT_MAX_SERIES = {
   table: 100,
diff --git a/dashboards/vm/vmagent.json b/dashboards/vm/vmagent.json
index c10b92e03..80d5ee771 100644
--- a/dashboards/vm/vmagent.json
+++ b/dashboards/vm/vmagent.json
@@ -4575,7 +4575,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -4591,7 +4592,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 14
+            "y": 6
           },
           "id": 131,
           "options": {
@@ -4679,7 +4680,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -4695,7 +4697,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 22
+            "y": 14
           },
           "id": 130,
           "options": {
@@ -4796,7 +4798,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -4812,7 +4815,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 22
+            "y": 14
           },
           "id": 77,
           "options": {
@@ -4887,7 +4890,7 @@
             "type": "victoriametrics-datasource",
             "uid": "$ds"
           },
-          "description": "The 99th percentile of avg flush duration for the aggregated data. \n\nSmaller is better.\n\nAggregation can produce incorrect results ff flush duration exceeds configured deduplication interval. See \"Flush Timeouts\" panel.",
+          "description": "Shows the number of matched samples by the aggregation rule. \n\nThe more samples is matched, the more work this aggregation rule does. The matching rule is specified via `match` param.\n\nSee more details in [stream aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config). ",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -4939,8 +4942,7 @@
                     "value": 80
                   }
                 ]
-              },
-              "unit": "s"
+              }
             },
             "overrides": []
           },
@@ -4950,13 +4952,19 @@
             "x": 0,
             "y": 7
           },
-          "id": 137,
+          "id": 146,
           "options": {
             "legend": {
-              "calcs": [],
-              "displayMode": "list",
+              "calcs": [
+                "min",
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
               "placement": "bottom",
-              "showLegend": true
+              "showLegend": true,
+              "sortBy": "Last *",
+              "sortDesc": true
             },
             "tooltip": {
               "mode": "single",
@@ -4970,14 +4978,14 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "histogram_quantile(0.99, vm_streamaggr_dedup_flush_duration_seconds_bucket{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval])",
+              "expr": "rate(vm_streamaggr_matched_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Dedup flush duration (0.99)",
+          "title": "Matched samples ($instance)",
           "type": "timeseries"
         },
         {
@@ -4985,7 +4993,7 @@
             "type": "victoriametrics-datasource",
             "uid": "$ds"
           },
-          "description": "Amount of ignored samples during aggregation. \nStream aggregation will drop samples with NaN values, or samples with too old timestamps. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples ",
+          "description": "The rate of ignored samples during aggregation. \nStream aggregation will drop samples with NaN values, or samples with too old timestamps. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples ",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5067,14 +5075,117 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "increase(vm_streamaggr_ignored_samples_total[5m]) > 0",
+              "expr": "rate(vm_streamaggr_ignored_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
+              "instant": false,
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
+              "range": true,
+              "refId": "A"
+            }
+          ],
+          "title": "Ignored samples ($instance)",
+          "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "type": "victoriametrics-datasource",
+            "uid": "$ds"
+          },
+          "description": "Shows the number of produced samples by the aggregation rule. \n\nNumber of produced samples depend on params like `by`, `without`, `interval`, etc.\n\nSee more details in [stream aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config). ",
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "axisSoftMin": 0,
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 0,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 5,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "auto",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  },
+                  {
+                    "color": "red",
+                    "value": 80
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 15
+          },
+          "id": 147,
+          "options": {
+            "legend": {
+              "calcs": [
+                "min",
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true,
+              "sortBy": "Last *",
+              "sortDesc": true
+            },
+            "tooltip": {
+              "mode": "single",
+              "sort": "none"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "victoriametrics-datasource",
+                "uid": "$ds"
+              },
+              "editorMode": "code",
+              "expr": "rate(vm_streamaggr_flushed_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
               "instant": false,
-              "legendFormat": "{{reason}}: {{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; output={{output}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Ignored samples",
+          "title": "Produced samples ($instance)",
           "type": "timeseries"
         },
         {
@@ -5142,7 +5253,7 @@
           "gridPos": {
             "h": 8,
             "w": 12,
-            "x": 0,
+            "x": 12,
             "y": 15
           },
           "id": 139,
@@ -5165,9 +5276,9 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "increase(vm_streamaggr_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval]) > 0",
+              "expr": "increase(vm_streamaggr_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval])",
               "instant": false,
-              "legendFormat": "aggregate: {{instance}} ({{job}})",
+              "legendFormat": "aggregation: {{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             },
@@ -5177,15 +5288,15 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "increase(vm_streamaggr_dedup_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval]) > 0",
+              "expr": "increase(vm_streamaggr_dedup_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval])",
               "hide": false,
               "instant": false,
-              "legendFormat": "dedup: {{instance}} ({{job}})",
+              "legendFormat": "deduplication: {{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "B"
             }
           ],
-          "title": "Flush timeouts",
+          "title": "Flush timeouts ($instance)",
           "type": "timeseries"
         },
         {
@@ -5193,7 +5304,7 @@
             "type": "victoriametrics-datasource",
             "uid": "$ds"
           },
-          "description": "Shows the lag between average samples timestamp and aggregation interval.\n\nLower is better.\n\nToo high lag or lag exceeding the interval might be a sign of data delay before aggregation or resource insufficiency on aggregator. Samples with high lag may affect accuracy of aggregation.\n\nSee https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples",
+          "description": "Shows the max lag between samples timestamps within one batch passed to the aggregation input.\n\nLower is better.\n\nToo high lag or lag exceeding the interval might be a sign that data was delayed before aggregation or resource insufficiency on aggregator. Samples with high lag may affect accuracy of aggregation.\n\nSee https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5254,8 +5365,8 @@
           "gridPos": {
             "h": 8,
             "w": 12,
-            "x": 12,
-            "y": 15
+            "x": 0,
+            "y": 23
           },
           "id": 142,
           "options": {
@@ -5277,14 +5388,14 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "histogram_quantile(0.99, vm_streamaggr_samples_lag_seconds_bucket{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval])",
+              "expr": "histogram_quantile(0.99, rate(vm_streamaggr_samples_lag_seconds_bucket{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]))",
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Aggregated samples lag",
+          "title": "Samples lag 0.99 quantile ($instance)",
           "type": "timeseries"
         },
         {
@@ -5292,7 +5403,7 @@
             "type": "victoriametrics-datasource",
             "uid": "$ds"
           },
-          "description": "Shows the size of Label Compressor in bytes.\n\nLabels compressor encodes label-value pairs during aggregation to optimise memory usage. It is expected for its size to grow with time and to reset on vmagent restarts.\n\nRapid spikes in Label compressor size might be a sign of significant changes in labels of received samples.",
+          "description": "The 99th percentile of avg flush duration for the aggregated data. \n\nSmaller is better.\n\nAggregation can produce incorrect results ff flush duration exceeds configured deduplication interval. See \"Flush Timeouts\" panel.",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5345,17 +5456,17 @@
                   }
                 ]
               },
-              "unit": "decbytes"
+              "unit": "s"
             },
             "overrides": []
           },
           "gridPos": {
             "h": 8,
             "w": 12,
-            "x": 0,
+            "x": 12,
             "y": 23
           },
-          "id": 140,
+          "id": 137,
           "options": {
             "legend": {
               "calcs": [],
@@ -5375,14 +5486,14 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "vm_streamaggr_labels_compressor_size_bytes{job=~\"$job\",instance=~\"$instance\"} > 0",
+              "expr": "histogram_quantile(0.99, rate(vm_streamaggr_dedup_flush_duration_seconds_bucket{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]))",
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Labels compressor bytes",
+          "title": "Dedup flush duration 0.99 quantile ($instance)",
           "type": "timeseries"
         },
         {
@@ -5390,7 +5501,7 @@
             "type": "victoriametrics-datasource",
             "uid": "$ds"
           },
-          "description": "Shows the size of Label Compressor in number of entries.\n\nLabels compressor encodes label-value pairs during aggregation to optimise memory usage. It is expected for its size to grow with time and to reset on vmagent restarts.\n\nRapid spikes in Label compressor size might be a sign of significant changes in labels of received samples.",
+          "description": "Shows the eviction rate of time series because of staleness.\n\nThere are two stages where series can be marked as stale.\n1. Input. Aggregator keeps in memory each received unique time series. The time series becomes stale and gets removed if no samples were received during [staleness interval](https://docs.victoriametrics.com/stream-aggregation/#staleness) for this series. \n\n2. Output. The output key is a resulting time series produced by aggregating many input series. The time series becomes stale and gets removed if no samples were received during [staleness interval](https://docs.victoriametrics.com/stream-aggregation/#staleness) for any of input series for this aggregation.\n\nIncrease in `input` keys shows that series previously matched by the aggregation rule now became stale.\n\nIncrease in `output` keys shows that series previously produced by the aggregation rule now became stale.",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5446,11 +5557,125 @@
             },
             "overrides": []
           },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 31
+          },
+          "id": 144,
+          "options": {
+            "legend": {
+              "calcs": [],
+              "displayMode": "list",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "single",
+              "sort": "none"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "victoriametrics-datasource",
+                "uid": "$ds"
+              },
+              "editorMode": "code",
+              "expr": "increase(vm_streamaggr_stale_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
+              "instant": false,
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; key={{key}}",
+              "range": true,
+              "refId": "A"
+            }
+          ],
+          "title": "Staleness rate ($instance)",
+          "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "type": "victoriametrics-datasource",
+            "uid": "$ds"
+          },
+          "description": "Shows the size of Label Compressor in number of entries.\n\nLabels compressor encodes label-value pairs during aggregation to optimise memory usage. It is expected for its size to grow with time and to reset on vmagent restarts.\n\nRapid spikes in Label compressor size might be a sign of significant changes in labels of received samples.",
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "axisSoftMin": 0,
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 0,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 5,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "auto",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  },
+                  {
+                    "color": "red",
+                    "value": 80
+                  }
+                ]
+              }
+            },
+            "overrides": [
+              {
+                "matcher": {
+                  "id": "byRegexp",
+                  "options": "/bytes.*/"
+                },
+                "properties": [
+                  {
+                    "id": "custom.axisPlacement",
+                    "value": "right"
+                  },
+                  {
+                    "id": "unit",
+                    "value": "bytes"
+                  }
+                ]
+              }
+            ]
+          },
           "gridPos": {
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 23
+            "y": 31
           },
           "id": 141,
           "options": {
@@ -5472,14 +5697,28 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "vm_streamaggr_labels_compressor_items_count{job=~\"$job\",instance=~\"$instance\"} > 0",
+              "expr": "max(vm_streamaggr_labels_compressor_items_count{job=~\"$job\",instance=~\"$instance\"}) by(job, instance)",
+              "hide": false,
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "items: {{instance}} ({{job}})",
               "range": true,
               "refId": "A"
+            },
+            {
+              "datasource": {
+                "type": "victoriametrics-datasource",
+                "uid": "$ds"
+              },
+              "editorMode": "code",
+              "expr": "max(vm_streamaggr_labels_compressor_size_bytes{job=~\"$job\", instance=~\"$instance\"}) by(job, instance)",
+              "hide": false,
+              "instant": false,
+              "legendFormat": "bytes: {{instance}} ({{job}})",
+              "range": true,
+              "refId": "B"
             }
           ],
-          "title": "Labels compressor items count",
+          "title": "Labels compressor ($instance)",
           "type": "timeseries"
         }
       ],
@@ -5551,7 +5790,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5567,7 +5807,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 16
+            "y": 8
           },
           "id": 60,
           "options": {
@@ -5655,7 +5895,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5671,7 +5912,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 16
+            "y": 8
           },
           "id": 66,
           "options": {
@@ -5759,7 +6000,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5775,7 +6017,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 24
+            "y": 16
           },
           "id": 61,
           "options": {
@@ -5863,7 +6105,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5879,7 +6122,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 24
+            "y": 16
           },
           "id": 65,
           "options": {
@@ -5929,6 +6172,7 @@
                 "mode": "palette-classic"
               },
               "custom": {
+                "axisBorderShow": false,
                 "axisCenteredZero": false,
                 "axisColorMode": "text",
                 "axisLabel": "",
@@ -5942,6 +6186,7 @@
                   "tooltip": false,
                   "viz": false
                 },
+                "insertNulls": false,
                 "lineInterpolation": "linear",
                 "lineWidth": 1,
                 "pointSize": 5,
@@ -5964,7 +6209,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "transparent"
+                    "color": "transparent",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5980,7 +6226,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 32
+            "y": 24
           },
           "id": 88,
           "options": {
@@ -6026,6 +6272,7 @@
                 "mode": "palette-classic"
               },
               "custom": {
+                "axisBorderShow": false,
                 "axisCenteredZero": false,
                 "axisColorMode": "text",
                 "axisLabel": "",
@@ -6078,7 +6325,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 32
+            "y": 24
           },
           "id": 84,
           "options": {
@@ -6128,6 +6375,7 @@
                 "mode": "palette-classic"
               },
               "custom": {
+                "axisBorderShow": false,
                 "axisCenteredZero": false,
                 "axisColorMode": "text",
                 "axisLabel": "",
@@ -6141,6 +6389,7 @@
                   "tooltip": false,
                   "viz": false
                 },
+                "insertNulls": false,
                 "lineInterpolation": "linear",
                 "lineWidth": 1,
                 "pointSize": 5,
@@ -6163,7 +6412,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "transparent"
+                    "color": "transparent",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -6179,7 +6429,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 40
+            "y": 32
           },
           "id": 90,
           "options": {
@@ -6984,4 +7234,4 @@
   "uid": "G7Z9GzMGz_vm",
   "version": 1,
   "weekStart": ""
-}
\ No newline at end of file
+}
diff --git a/dashboards/vmagent.json b/dashboards/vmagent.json
index 5b02b43cc..fffa75163 100644
--- a/dashboards/vmagent.json
+++ b/dashboards/vmagent.json
@@ -4574,7 +4574,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -4590,7 +4591,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 14
+            "y": 6
           },
           "id": 131,
           "options": {
@@ -4678,7 +4679,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -4694,7 +4696,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 22
+            "y": 14
           },
           "id": 130,
           "options": {
@@ -4795,7 +4797,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -4811,7 +4814,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 22
+            "y": 14
           },
           "id": 77,
           "options": {
@@ -4886,7 +4889,7 @@
             "type": "prometheus",
             "uid": "$ds"
           },
-          "description": "The 99th percentile of avg flush duration for the aggregated data. \n\nSmaller is better.\n\nAggregation can produce incorrect results ff flush duration exceeds configured deduplication interval. See \"Flush Timeouts\" panel.",
+          "description": "Shows the number of matched samples by the aggregation rule. \n\nThe more samples is matched, the more work this aggregation rule does. The matching rule is specified via `match` param.\n\nSee more details in [stream aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config). ",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -4938,8 +4941,7 @@
                     "value": 80
                   }
                 ]
-              },
-              "unit": "s"
+              }
             },
             "overrides": []
           },
@@ -4949,13 +4951,19 @@
             "x": 0,
             "y": 7
           },
-          "id": 137,
+          "id": 146,
           "options": {
             "legend": {
-              "calcs": [],
-              "displayMode": "list",
+              "calcs": [
+                "min",
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
               "placement": "bottom",
-              "showLegend": true
+              "showLegend": true,
+              "sortBy": "Last *",
+              "sortDesc": true
             },
             "tooltip": {
               "mode": "single",
@@ -4969,14 +4977,14 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "histogram_quantile(0.99, vm_streamaggr_dedup_flush_duration_seconds_bucket{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval])",
+              "expr": "rate(vm_streamaggr_matched_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Dedup flush duration (0.99)",
+          "title": "Matched samples ($instance)",
           "type": "timeseries"
         },
         {
@@ -4984,7 +4992,7 @@
             "type": "prometheus",
             "uid": "$ds"
           },
-          "description": "Amount of ignored samples during aggregation. \nStream aggregation will drop samples with NaN values, or samples with too old timestamps. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples ",
+          "description": "The rate of ignored samples during aggregation. \nStream aggregation will drop samples with NaN values, or samples with too old timestamps. See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples ",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5066,14 +5074,117 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "increase(vm_streamaggr_ignored_samples_total[5m]) > 0",
+              "expr": "rate(vm_streamaggr_ignored_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
+              "instant": false,
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
+              "range": true,
+              "refId": "A"
+            }
+          ],
+          "title": "Ignored samples ($instance)",
+          "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "$ds"
+          },
+          "description": "Shows the number of produced samples by the aggregation rule. \n\nNumber of produced samples depend on params like `by`, `without`, `interval`, etc.\n\nSee more details in [stream aggregation config](https://docs.victoriametrics.com/stream-aggregation/#stream-aggregation-config). ",
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "axisSoftMin": 0,
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 0,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 5,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "auto",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  },
+                  {
+                    "color": "red",
+                    "value": 80
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 15
+          },
+          "id": 147,
+          "options": {
+            "legend": {
+              "calcs": [
+                "min",
+                "lastNotNull",
+                "max"
+              ],
+              "displayMode": "table",
+              "placement": "bottom",
+              "showLegend": true,
+              "sortBy": "Last *",
+              "sortDesc": true
+            },
+            "tooltip": {
+              "mode": "single",
+              "sort": "none"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "$ds"
+              },
+              "editorMode": "code",
+              "expr": "rate(vm_streamaggr_flushed_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
               "instant": false,
-              "legendFormat": "{{reason}}: {{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; output={{output}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Ignored samples",
+          "title": "Produced samples ($instance)",
           "type": "timeseries"
         },
         {
@@ -5141,7 +5252,7 @@
           "gridPos": {
             "h": 8,
             "w": 12,
-            "x": 0,
+            "x": 12,
             "y": 15
           },
           "id": 139,
@@ -5164,9 +5275,9 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "increase(vm_streamaggr_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval]) > 0",
+              "expr": "increase(vm_streamaggr_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval])",
               "instant": false,
-              "legendFormat": "aggregate: {{instance}} ({{job}})",
+              "legendFormat": "aggregation: {{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             },
@@ -5176,15 +5287,15 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "increase(vm_streamaggr_dedup_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval]) > 0",
+              "expr": "increase(vm_streamaggr_dedup_flush_timeouts_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval])",
               "hide": false,
               "instant": false,
-              "legendFormat": "dedup: {{instance}} ({{job}})",
+              "legendFormat": "deduplication: {{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "B"
             }
           ],
-          "title": "Flush timeouts",
+          "title": "Flush timeouts ($instance)",
           "type": "timeseries"
         },
         {
@@ -5192,7 +5303,7 @@
             "type": "prometheus",
             "uid": "$ds"
           },
-          "description": "Shows the lag between average samples timestamp and aggregation interval.\n\nLower is better.\n\nToo high lag or lag exceeding the interval might be a sign of data delay before aggregation or resource insufficiency on aggregator. Samples with high lag may affect accuracy of aggregation.\n\nSee https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples",
+          "description": "Shows the max lag between samples timestamps within one batch passed to the aggregation input.\n\nLower is better.\n\nToo high lag or lag exceeding the interval might be a sign that data was delayed before aggregation or resource insufficiency on aggregator. Samples with high lag may affect accuracy of aggregation.\n\nSee https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5253,8 +5364,8 @@
           "gridPos": {
             "h": 8,
             "w": 12,
-            "x": 12,
-            "y": 15
+            "x": 0,
+            "y": 23
           },
           "id": 142,
           "options": {
@@ -5276,14 +5387,14 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "histogram_quantile(0.99, vm_streamaggr_samples_lag_seconds_bucket{job=~\"$job\",instance=~\"$instance\"}[$__rate_interval])",
+              "expr": "histogram_quantile(0.99, rate(vm_streamaggr_samples_lag_seconds_bucket{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]))",
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Aggregated samples lag",
+          "title": "Samples lag 0.99 quantile ($instance)",
           "type": "timeseries"
         },
         {
@@ -5291,7 +5402,7 @@
             "type": "prometheus",
             "uid": "$ds"
           },
-          "description": "Shows the size of Label Compressor in bytes.\n\nLabels compressor encodes label-value pairs during aggregation to optimise memory usage. It is expected for its size to grow with time and to reset on vmagent restarts.\n\nRapid spikes in Label compressor size might be a sign of significant changes in labels of received samples.",
+          "description": "The 99th percentile of avg flush duration for the aggregated data. \n\nSmaller is better.\n\nAggregation can produce incorrect results ff flush duration exceeds configured deduplication interval. See \"Flush Timeouts\" panel.",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5344,17 +5455,17 @@
                   }
                 ]
               },
-              "unit": "decbytes"
+              "unit": "s"
             },
             "overrides": []
           },
           "gridPos": {
             "h": 8,
             "w": 12,
-            "x": 0,
+            "x": 12,
             "y": 23
           },
-          "id": 140,
+          "id": 137,
           "options": {
             "legend": {
               "calcs": [],
@@ -5374,14 +5485,14 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "vm_streamaggr_labels_compressor_size_bytes{job=~\"$job\",instance=~\"$instance\"} > 0",
+              "expr": "histogram_quantile(0.99, rate(vm_streamaggr_dedup_flush_duration_seconds_bucket{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]))",
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; outputs={{outputs}}",
               "range": true,
               "refId": "A"
             }
           ],
-          "title": "Labels compressor bytes",
+          "title": "Dedup flush duration 0.99 quantile ($instance)",
           "type": "timeseries"
         },
         {
@@ -5389,7 +5500,7 @@
             "type": "prometheus",
             "uid": "$ds"
           },
-          "description": "Shows the size of Label Compressor in number of entries.\n\nLabels compressor encodes label-value pairs during aggregation to optimise memory usage. It is expected for its size to grow with time and to reset on vmagent restarts.\n\nRapid spikes in Label compressor size might be a sign of significant changes in labels of received samples.",
+          "description": "Shows the eviction rate of time series because of staleness.\n\nThere are two stages where series can be marked as stale.\n1. Input. Aggregator keeps in memory each received unique time series. The time series becomes stale and gets removed if no samples were received during [staleness interval](https://docs.victoriametrics.com/stream-aggregation/#staleness) for this series. \n\n2. Output. The output key is a resulting time series produced by aggregating many input series. The time series becomes stale and gets removed if no samples were received during [staleness interval](https://docs.victoriametrics.com/stream-aggregation/#staleness) for any of input series for this aggregation.\n\nIncrease in `input` keys shows that series previously matched by the aggregation rule now became stale.\n\nIncrease in `output` keys shows that series previously produced by the aggregation rule now became stale.",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -5445,11 +5556,125 @@
             },
             "overrides": []
           },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 31
+          },
+          "id": 144,
+          "options": {
+            "legend": {
+              "calcs": [],
+              "displayMode": "list",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "mode": "single",
+              "sort": "none"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "$ds"
+              },
+              "editorMode": "code",
+              "expr": "increase(vm_streamaggr_stale_samples_total{job=~\"$job\",instance=~\"$instance\", url=~\"$url\"}[$__rate_interval]) > 0",
+              "instant": false,
+              "legendFormat": "{{url}} ({{job}}): match={{match}}; key={{key}}",
+              "range": true,
+              "refId": "A"
+            }
+          ],
+          "title": "Staleness rate ($instance)",
+          "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "$ds"
+          },
+          "description": "Shows the size of Label Compressor in number of entries.\n\nLabels compressor encodes label-value pairs during aggregation to optimise memory usage. It is expected for its size to grow with time and to reset on vmagent restarts.\n\nRapid spikes in Label compressor size might be a sign of significant changes in labels of received samples.",
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisBorderShow": false,
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "axisSoftMin": 0,
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 0,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "insertNulls": false,
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 5,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "auto",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green",
+                    "value": null
+                  },
+                  {
+                    "color": "red",
+                    "value": 80
+                  }
+                ]
+              }
+            },
+            "overrides": [
+              {
+                "matcher": {
+                  "id": "byRegexp",
+                  "options": "/bytes.*/"
+                },
+                "properties": [
+                  {
+                    "id": "custom.axisPlacement",
+                    "value": "right"
+                  },
+                  {
+                    "id": "unit",
+                    "value": "bytes"
+                  }
+                ]
+              }
+            ]
+          },
           "gridPos": {
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 23
+            "y": 31
           },
           "id": 141,
           "options": {
@@ -5471,14 +5696,28 @@
                 "uid": "$ds"
               },
               "editorMode": "code",
-              "expr": "vm_streamaggr_labels_compressor_items_count{job=~\"$job\",instance=~\"$instance\"} > 0",
+              "expr": "max(vm_streamaggr_labels_compressor_items_count{job=~\"$job\",instance=~\"$instance\"}) by(job, instance)",
+              "hide": false,
               "instant": false,
-              "legendFormat": "{{instance}} ({{job}})",
+              "legendFormat": "items: {{instance}} ({{job}})",
               "range": true,
               "refId": "A"
+            },
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "$ds"
+              },
+              "editorMode": "code",
+              "expr": "max(vm_streamaggr_labels_compressor_size_bytes{job=~\"$job\", instance=~\"$instance\"}) by(job, instance)",
+              "hide": false,
+              "instant": false,
+              "legendFormat": "bytes: {{instance}} ({{job}})",
+              "range": true,
+              "refId": "B"
             }
           ],
-          "title": "Labels compressor items count",
+          "title": "Labels compressor ($instance)",
           "type": "timeseries"
         }
       ],
@@ -5550,7 +5789,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5566,7 +5806,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 16
+            "y": 8
           },
           "id": 60,
           "options": {
@@ -5654,7 +5894,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5670,7 +5911,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 16
+            "y": 8
           },
           "id": 66,
           "options": {
@@ -5758,7 +5999,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5774,7 +6016,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 24
+            "y": 16
           },
           "id": 61,
           "options": {
@@ -5862,7 +6104,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "green"
+                    "color": "green",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5878,7 +6121,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 24
+            "y": 16
           },
           "id": 65,
           "options": {
@@ -5928,6 +6171,7 @@
                 "mode": "palette-classic"
               },
               "custom": {
+                "axisBorderShow": false,
                 "axisCenteredZero": false,
                 "axisColorMode": "text",
                 "axisLabel": "",
@@ -5941,6 +6185,7 @@
                   "tooltip": false,
                   "viz": false
                 },
+                "insertNulls": false,
                 "lineInterpolation": "linear",
                 "lineWidth": 1,
                 "pointSize": 5,
@@ -5963,7 +6208,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "transparent"
+                    "color": "transparent",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -5979,7 +6225,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 32
+            "y": 24
           },
           "id": 88,
           "options": {
@@ -6025,6 +6271,7 @@
                 "mode": "palette-classic"
               },
               "custom": {
+                "axisBorderShow": false,
                 "axisCenteredZero": false,
                 "axisColorMode": "text",
                 "axisLabel": "",
@@ -6077,7 +6324,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 32
+            "y": 24
           },
           "id": 84,
           "options": {
@@ -6127,6 +6374,7 @@
                 "mode": "palette-classic"
               },
               "custom": {
+                "axisBorderShow": false,
                 "axisCenteredZero": false,
                 "axisColorMode": "text",
                 "axisLabel": "",
@@ -6140,6 +6388,7 @@
                   "tooltip": false,
                   "viz": false
                 },
+                "insertNulls": false,
                 "lineInterpolation": "linear",
                 "lineWidth": 1,
                 "pointSize": 5,
@@ -6162,7 +6411,8 @@
                 "mode": "absolute",
                 "steps": [
                   {
-                    "color": "transparent"
+                    "color": "transparent",
+                    "value": null
                   },
                   {
                     "color": "red",
@@ -6178,7 +6428,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 40
+            "y": 32
           },
           "id": 90,
           "options": {
@@ -6983,4 +7233,4 @@
   "uid": "G7Z9GzMGz",
   "version": 1,
   "weekStart": ""
-}
\ No newline at end of file
+}
diff --git a/deployment/docker/docker-compose-victorialogs.yml b/deployment/docker/docker-compose-victorialogs.yml
index 0ac373fa3..a60a0c56b 100644
--- a/deployment/docker/docker-compose-victorialogs.yml
+++ b/deployment/docker/docker-compose-victorialogs.yml
@@ -40,7 +40,7 @@ services:
 #  storing logs and serving read queries.
   victorialogs:
     container_name: victorialogs
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     command:
       - "--storageDataPath=/vlogs"
       - "--httpListenAddr=:9428"
diff --git a/deployment/docker/victorialogs/filebeat-docker/docker-compose.yml b/deployment/docker/victorialogs/filebeat-docker/docker-compose.yml
index 423c09187..5103aa4c2 100644
--- a/deployment/docker/victorialogs/filebeat-docker/docker-compose.yml
+++ b/deployment/docker/victorialogs/filebeat-docker/docker-compose.yml
@@ -22,7 +22,7 @@ services:
       - -beat.uri=http://filebeat-victorialogs:5066
 
   victorialogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - victorialogs-filebeat-docker-vl:/vlogs
     ports:
diff --git a/deployment/docker/victorialogs/filebeat-syslog/docker-compose.yml b/deployment/docker/victorialogs/filebeat-syslog/docker-compose.yml
index 39b30c2ed..a07e8a9e9 100644
--- a/deployment/docker/victorialogs/filebeat-syslog/docker-compose.yml
+++ b/deployment/docker/victorialogs/filebeat-syslog/docker-compose.yml
@@ -13,7 +13,7 @@ services:
       - "5140:5140"
 
   victorialogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - victorialogs-filebeat-syslog-vl:/vlogs
     ports:
diff --git a/deployment/docker/victorialogs/fluentbit-docker/docker-compose.yml b/deployment/docker/victorialogs/fluentbit-docker/docker-compose.yml
index 272e00e11..261ee6b08 100644
--- a/deployment/docker/victorialogs/fluentbit-docker/docker-compose.yml
+++ b/deployment/docker/victorialogs/fluentbit-docker/docker-compose.yml
@@ -11,7 +11,7 @@ services:
       - "5140:5140"
 
   victorialogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - victorialogs-fluentbit-vl:/vlogs
     ports:
diff --git a/deployment/docker/victorialogs/logstash/docker-compose.yml b/deployment/docker/victorialogs/logstash/docker-compose.yml
index a78879005..286f6b189 100644
--- a/deployment/docker/victorialogs/logstash/docker-compose.yml
+++ b/deployment/docker/victorialogs/logstash/docker-compose.yml
@@ -14,7 +14,7 @@ services:
       - "5140:5140"
 
   victorialogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - victorialogs-logstash-vl:/vlogs
     ports:
diff --git a/deployment/docker/victorialogs/promtail/docker-compose.yml b/deployment/docker/victorialogs/promtail/docker-compose.yml
index 24e8bf488..adc8bbc28 100644
--- a/deployment/docker/victorialogs/promtail/docker-compose.yml
+++ b/deployment/docker/victorialogs/promtail/docker-compose.yml
@@ -12,7 +12,7 @@ services:
       - "5140:5140"
 
   vlogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - victorialogs-promtail-docker:/vlogs
     ports:
diff --git a/deployment/docker/victorialogs/vector-docker/docker-compose.yml b/deployment/docker/victorialogs/vector-docker/docker-compose.yml
index fc1c270e4..1debfb2e4 100644
--- a/deployment/docker/victorialogs/vector-docker/docker-compose.yml
+++ b/deployment/docker/victorialogs/vector-docker/docker-compose.yml
@@ -22,7 +22,7 @@ services:
         condition: service_healthy
 
   victorialogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - victorialogs-vector-docker-vl:/vlogs
     ports:
diff --git a/deployment/logs-benchmark/docker-compose.yml b/deployment/logs-benchmark/docker-compose.yml
index 580bc6615..3f26b2f25 100644
--- a/deployment/logs-benchmark/docker-compose.yml
+++ b/deployment/logs-benchmark/docker-compose.yml
@@ -3,7 +3,7 @@ version: '3'
 services:
   # Run `make package-victoria-logs` to build victoria-logs image
   vlogs:
-    image: docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+    image: docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
     volumes:
       - vlogs:/vlogs
     ports:
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 3565afe96..37af6f63f 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -29,9 +29,18 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 ## tip
 
+**Update note 1: support for snap packages was removed due to lack of interest from community. See this [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6543) for details. Please read about supported package types [here](https://docs.victoriametrics.com/#install).**
+
 * FEATURE: [vmauth](https://docs.victoriametrics.com/vmauth/): allow overriding `Host` header with a target host before sending to a downstream. See this [issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6453)
 * FEATURE: [dashboards](https://grafana.com/orgs/victoriametrics): add [Grafana dashboard](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/dashboards/vmauth.json) and [alerting rules](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/alerts-vmauth.yml) for [vmauth](https://docs.victoriametrics.com/vmauth/) dashboard. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4313) for details.
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): added `yandexcloud_sd` AWS API IMDSv2 support.
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): expose metrics related to [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/):
+  * `vm_streamaggr_matched_samples_total` - shows the number of samples matched by the aggregation rule; 
+  * `vm_streamaggr_flushed_samples_total` - shows the number of samples produced by the aggregation rule; 
+  * `vm_streamaggr_samples_lag_seconds` - shows the max lag between samples timestamps within one batch received by the aggregation; 
+  * `vm_streamaggr_stale_samples_total` - shows the number of time series that became [stale](https://docs.victoriametrics.com/stream-aggregation/#staleness) during aggregation;
+  * metrics related to stream aggregation got additional labels `match` (matching param), `group` (`by` or `without` param), `url` (address of `remoteWrite.url` where aggregation is applied), `position` (the position of the aggregation rule in config file).
+  * These and other metrics were reflected on the [vmagent dashboard](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/dashboards/vmagent.json) in `stream aggregation` section.
 * BUGFIX: [docker-compose](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker#docker-compose-environment-for-victoriametrics): fix incorrect link to vmui from [VictoriaMetrics plugin in Grafana](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker#grafana).
 
 * BUGFIX: [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) and `vmstorage` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): Fix the dateMetricIDCache consistency issue that leads to duplicate per-day index entries when new time series are inserted concurrently. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6534) for details.
diff --git a/docs/Quick-Start.md b/docs/Quick-Start.md
index f2c64174e..f9ea00dd0 100644
--- a/docs/Quick-Start.md
+++ b/docs/Quick-Start.md
@@ -25,7 +25,6 @@ Single-server-VictoriaMetrics VictoriaMetrics is available as:
 
 * [Managed VictoriaMetrics at AWS](https://aws.amazon.com/marketplace/pp/prodview-4tbfq5icmbmyc)
 * [Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/)
-* [Snap packages](https://snapcraft.io/victoriametrics)
 * [Helm Charts](https://github.com/VictoriaMetrics/helm-charts#list-of-charts)
 * [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest)
 * [Source code](https://github.com/VictoriaMetrics/VictoriaMetrics).
diff --git a/docs/README.md b/docs/README.md
index 750465f83..ac8bec4d0 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -18,8 +18,7 @@ VictoriaMetrics is a fast, cost-effective and scalable monitoring solution and t
 See [case studies for VictoriaMetrics](https://docs.victoriametrics.com/casestudies/).
 
 VictoriaMetrics is available in [binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest),
-[Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/), [Snap packages](https://snapcraft.io/victoriametrics)
-and [source code](https://github.com/VictoriaMetrics/VictoriaMetrics). 
+[Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/) and [source code](https://github.com/VictoriaMetrics/VictoriaMetrics). 
 
 Documentation for the cluster version of VictoriaMetrics is available [here](https://docs.victoriametrics.com/cluster-victoriametrics/).
 
@@ -142,7 +141,6 @@ VictoriaMetrics can also be installed via these installation methods:
 * [Ansible role for installing cluster VictoriaMetrics (by VictoriaMetrics)](https://github.com/VictoriaMetrics/ansible-playbooks).
 * [Ansible role for installing cluster VictoriaMetrics (by community)](https://github.com/Slapper/ansible-victoriametrics-cluster-role).
 * [Ansible role for installing single-node VictoriaMetrics (by community)](https://github.com/dreamteam-gg/ansible-victoriametrics-role).
-* [Snap package for VictoriaMetrics](https://snapcraft.io/victoriametrics).
 
 ### How to start VictoriaMetrics
 
@@ -185,24 +183,7 @@ Additionally, all the VictoriaMetrics components allow setting flag values via e
 
 ### Configuration with snap package
 
-Snap package for VictoriaMetrics is available [here](https://snapcraft.io/victoriametrics).
-
-Command-line flags for Snap package can be set with following command:
-
-```sh
-echo 'FLAGS="-selfScrapeInterval=10s -search.logSlowQueryDuration=20s"' > $SNAP_DATA/var/snap/victoriametrics/current/extra_flags
-snap restart victoriametrics
-```
-
-Do not change value for `-storageDataPath` flag, because snap package has limited access to host filesystem.
-
-Changing scrape configuration is possible with text editor:
-
-```sh
-vi $SNAP_DATA/var/snap/victoriametrics/current/etc/victoriametrics-scrape-config.yaml
-```
-
-After changes were made, trigger config re-read with the command `curl 127.0.0.1:8428/-/reload`.
+Snap packages for VictoriaMetrics are supported by community and are available at [https://snapcraft.io/victoriametrics](https://snapcraft.io/victoriametrics).
 
 ### Running as Windows service
 
@@ -1208,7 +1189,7 @@ More details may be found [here](https://github.com/VictoriaMetrics/VictoriaMetr
 ## Setting up service
 
 Read [instructions](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/43) on how to set up VictoriaMetrics
-as a service for your OS. A [snap package](https://snapcraft.io/victoriametrics) is available for Ubuntu.
+as a service for your OS.
 
 ## How to work with snapshots
 
diff --git a/docs/Release-Guide.md b/docs/Release-Guide.md
index 0a4fe5f66..f201a1a27 100644
--- a/docs/Release-Guide.md
+++ b/docs/Release-Guide.md
@@ -99,20 +99,6 @@ Bumping the limits may significantly improve build speed.
 1. Bump VictoriaMetrics version at `deployment/docker/docker-compose.yml` and at `deployment/docker/docker-compose-cluster.yml`.
 1. Follow the instructions in [release follow-up](https://github.com/VictoriaMetrics/VictoriaMetrics-enterprise/blob/master/Release-Guide.md).
 
-## Building snap package
-
- pre-requirements:
-
-* snapcraft binary, can be installed with commands:
-   for MacOS `brew install snapcraft` and [install multipass](https://discourse.ubuntu.com/t/installing-multipass-on-macos/8329),
-   for Ubuntu - `sudo snap install snapcraft --classic`
-* exported snapcraft login to `~/.snap/login.json` with `snapcraft export-login login.json && mkdir -p ~/.snap && mv login.json ~/.snap/`
-* already created release at github (it operates `git describe` version, so git tag must be annotated).
-
-1. checkout to the latest git tag for single-node version.
-1. execute `make release-snap` - it must build and upload snap package.
-1. promote release to current, if needed manually at release page [snapcraft-releases](https://snapcraft.io/victoriametrics/releases)
-
 ### Public Announcement
 
 * Publish message in Slack  at <https://victoriametrics.slack.com>
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 034486ab2..97b8108a1 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -26,8 +26,7 @@ VictoriaMetrics is a fast, cost-effective and scalable monitoring solution and t
 See [case studies for VictoriaMetrics](https://docs.victoriametrics.com/casestudies/).
 
 VictoriaMetrics is available in [binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest),
-[Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/), [Snap packages](https://snapcraft.io/victoriametrics)
-and [source code](https://github.com/VictoriaMetrics/VictoriaMetrics). 
+[Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/) and [source code](https://github.com/VictoriaMetrics/VictoriaMetrics). 
 
 Documentation for the cluster version of VictoriaMetrics is available [here](https://docs.victoriametrics.com/cluster-victoriametrics/).
 
@@ -150,7 +149,6 @@ VictoriaMetrics can also be installed via these installation methods:
 * [Ansible role for installing cluster VictoriaMetrics (by VictoriaMetrics)](https://github.com/VictoriaMetrics/ansible-playbooks).
 * [Ansible role for installing cluster VictoriaMetrics (by community)](https://github.com/Slapper/ansible-victoriametrics-cluster-role).
 * [Ansible role for installing single-node VictoriaMetrics (by community)](https://github.com/dreamteam-gg/ansible-victoriametrics-role).
-* [Snap package for VictoriaMetrics](https://snapcraft.io/victoriametrics).
 
 ### How to start VictoriaMetrics
 
@@ -193,24 +191,7 @@ Additionally, all the VictoriaMetrics components allow setting flag values via e
 
 ### Configuration with snap package
 
-Snap package for VictoriaMetrics is available [here](https://snapcraft.io/victoriametrics).
-
-Command-line flags for Snap package can be set with following command:
-
-```sh
-echo 'FLAGS="-selfScrapeInterval=10s -search.logSlowQueryDuration=20s"' > $SNAP_DATA/var/snap/victoriametrics/current/extra_flags
-snap restart victoriametrics
-```
-
-Do not change value for `-storageDataPath` flag, because snap package has limited access to host filesystem.
-
-Changing scrape configuration is possible with text editor:
-
-```sh
-vi $SNAP_DATA/var/snap/victoriametrics/current/etc/victoriametrics-scrape-config.yaml
-```
-
-After changes were made, trigger config re-read with the command `curl 127.0.0.1:8428/-/reload`.
+Snap packages for VictoriaMetrics are supported by community and are available at [https://snapcraft.io/victoriametrics](https://snapcraft.io/victoriametrics).
 
 ### Running as Windows service
 
@@ -1216,7 +1197,7 @@ More details may be found [here](https://github.com/VictoriaMetrics/VictoriaMetr
 ## Setting up service
 
 Read [instructions](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/43) on how to set up VictoriaMetrics
-as a service for your OS. A [snap package](https://snapcraft.io/victoriametrics) is available for Ubuntu.
+as a service for your OS.
 
 ## How to work with snapshots
 
diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index eefa66c36..8e6bccb25 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -19,8 +19,29 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
+## [v0.26.1](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.26.1-victorialogs)
+
+Released at 2024-07-01
+
+* BUGFIX: return the proper surrounding logs for [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe) when additional [pipes](https://docs.victoriametrics.com/victorialogs/logsql/#pipes) are put after the `stream_context` pipe. This has been broken in [v0.26.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.26.0-victorialogs).
+
+## [v0.26.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.26.0-victorialogs)
+
+Released at 2024-07-01
+
+* FEATURE: add ability to return log position (aka rank) after sorting logs with [`sort` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#sort-pipe). This can be done by adding `rank as <fieldName>` to the end of `| sort ...` pipe. For example, `_time:5m | sort by (_time) rank as position` instructs storing position of every sorted log line into `position` [field name](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model).
+* FEATURE: add delimiter log with `---` message between log chunks returned by [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe). This should simplify investigation of the returned logs.
+* FEATURE: reduce memory usage when big number of context logs are requested from [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe).
+
+## [v0.25.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.25.0-victorialogs)
+
+Released at 2024-06-28
+
+* FEATURE: add ability to select surrounding logs in front and after the selected logs via [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe). This functionality may be useful for investigating stacktraces, panics or some correlated log messages. This functionality is similar to `grep -A` and `grep -B`.
 * FEATURE: add ability to return top `N` `"fields"` groups from  [`/select/logsql/hits` HTTP endpoint](https://docs.victoriametrics.com/victorialogs/querying/#querying-hits-stats), by specifying `fields_limit=N` query arg. This query arg is going to be used in [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6545).
 
+* BUGFIX: fix `runtime error: index out of range [0] with length 0` panic when empty lines are ingested via [Syslog format](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/) by Cisco controllers. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6548).
+
 ## [v0.24.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.24.0-victorialogs)
 
 Released at 2024-06-27
diff --git a/docs/VictoriaLogs/LogsQL.md b/docs/VictoriaLogs/LogsQL.md
index 3a2a5f499..c09424c0a 100644
--- a/docs/VictoriaLogs/LogsQL.md
+++ b/docs/VictoriaLogs/LogsQL.md
@@ -1305,6 +1305,8 @@ LogsQL supports the following pipes:
 - [`replace_regexp`](#replace_regexp-pipe) updates [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) with regular expressions.
 - [`sort`](#sort-pipe) sorts logs by the given [fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model).
 - [`stats`](#stats-pipe) calculates various stats over the selected logs.
+- [`stream_context`](#stream_context-pipe) allows selecting surrounding logs in front and after the matching logs
+  per each [log stream](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields).
 - [`top`](#top-pipe) returns top `N` field sets with the maximum number of matching logs.
 - [`uniq`](#uniq-pipe) returns unique log entires.
 - [`unpack_json`](#unpack_json-pipe) unpacks JSON messages from [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model).
@@ -2113,6 +2115,14 @@ and then returns the next 20 sorted logs for the last 5 minutes:
 _time:1h | sort by (request_duration desc) offset 10 limit 20
 ```
 
+It is possible returning a rank (sort order number) for every sorted log by adding `rank as <fieldName>` to the end of `| sort ...` pipe.
+For example, the following query stores rank for sorted by [`_time`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field) logs
+into `position` [field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model):
+
+```logsql
+_time:5m | sort by (_time) rank as position
+```
+
 Note that sorting of big number of logs can be slow and can consume a lot of additional memory.
 It is recommended limiting the number of logs before sorting with the following approaches:
 
@@ -2300,6 +2310,33 @@ _time:5m | stats
   count() total
 ```
 
+### stream_context pipe
+
+`| stream_context ...` [pipe](#pipes) allows selecting surrounding logs for the matching logs in [logs stream](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields)
+in the way similar to `grep -A` / `grep -B`. The returned log chunks are delimited with `---` [log message](https://docs.victoriametrics.com/victorialogs/keyconcepts/#message-field)
+for easier investigation.
+
+For example, the following query returns up to 10 additional logs after every log message with the `panic` [word](#word) across all the logs for the last 5 minutes:
+
+```logsql
+_time:5m panic | stream_context after 10
+```
+
+The following query returns up to 5 additional logs in front of every log message with the `stacktrace` [word](#word) across all the logs for the last 5 minutes:
+
+```logsql
+_time:5m stacktrace | stream_context before 5
+```
+
+The following query returns up to 2 logs in frount of the log message with the `error` [word](#word) and up to 5 logs after this log message
+across all the logs for the last 5 minutes:
+
+```logsql
+_time:5m error | stream_context before 2 after 5
+```
+
+The `| stream_context` [pipe](#pipes) must go first just after the [filters](#filters).
+
 ### top pipe
 
 `| top N by (field1, ..., fieldN)` [pipe](#pipes) returns top `N` sets for `(field1, ..., fieldN)` [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model)
@@ -3042,10 +3079,7 @@ See also:
 
 ## Stream context
 
-LogsQL will support the ability to select the given number of surrounding log lines for the selected log lines
-on a [per-stream](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) basis.
-
-See the [Roadmap](https://docs.victoriametrics.com/victorialogs/roadmap/) for details.
+See [`stream_context` pipe](#stream_context-pipe).
 
 ## Transformations
 
diff --git a/docs/VictoriaLogs/QuickStart.md b/docs/VictoriaLogs/QuickStart.md
index 328ecf466..0ec043dc9 100644
--- a/docs/VictoriaLogs/QuickStart.md
+++ b/docs/VictoriaLogs/QuickStart.md
@@ -36,8 +36,8 @@ Just download archive for the needed Operating system and architecture, unpack i
 For example, the following commands download VictoriaLogs archive for Linux/amd64, unpack and run it:
 
 ```sh
-curl -L -O https://github.com/VictoriaMetrics/VictoriaMetrics/releases/download/v0.24.0-victorialogs/victoria-logs-linux-amd64-v0.24.0-victorialogs.tar.gz
-tar xzf victoria-logs-linux-amd64-v0.24.0-victorialogs.tar.gz
+curl -L -O https://github.com/VictoriaMetrics/VictoriaMetrics/releases/download/v0.26.1-victorialogs/victoria-logs-linux-amd64-v0.26.1-victorialogs.tar.gz
+tar xzf victoria-logs-linux-amd64-v0.26.1-victorialogs.tar.gz
 ./victoria-logs-prod
 ```
 
@@ -61,7 +61,7 @@ Here is the command to run VictoriaLogs in a Docker container:
 
 ```sh
 docker run --rm -it -p 9428:9428 -v ./victoria-logs-data:/victoria-logs-data \
-  docker.io/victoriametrics/victoria-logs:v0.24.0-victorialogs
+  docker.io/victoriametrics/victoria-logs:v0.26.1-victorialogs
 ```
 
 See also:
diff --git a/docs/VictoriaLogs/README.md b/docs/VictoriaLogs/README.md
index b876af621..23190f985 100644
--- a/docs/VictoriaLogs/README.md
+++ b/docs/VictoriaLogs/README.md
@@ -29,12 +29,9 @@ VictoriaLogs provides the following features:
 - VictoriaLogs supports multitenancy - see [these docs](#multitenancy).
 - VictoriaLogs supports out-of-order logs' ingestion aka backfilling.
 - VictoriaLogs supports live tailing for newly ingested logs. See [these docs](https://docs.victoriametrics.com/victorialogs/querying/#live-tailing).
+- VictoriaLogs supports selecting surrounding logs in front and after the selected logs. See [these docs](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe).
 - VictoriaLogs provides web UI for querying logs - see [these docs](https://docs.victoriametrics.com/victorialogs/querying/#web-ui).
 
-VictoriaLogs is at the Preview stage now. It is ready for evaluation in production and verifying the claims given above.
-It isn't recommended to migrate from existing logging solutions to VictoriaLogs Preview in general cases yet.
-See the [Roadmap](https://docs.victoriametrics.com/victorialogs/roadmap/) for details.
-
 If you have questions about VictoriaLogs, then read [this FAQ](https://docs.victoriametrics.com/victorialogs/faq/).
 Also feel free asking any questions at [VictoriaMetrics community Slack chat](https://victoriametrics.slack.com/), 
 you can join it via [Slack Inviter](https://slack.victoriametrics.com/).
diff --git a/docs/VictoriaLogs/Roadmap.md b/docs/VictoriaLogs/Roadmap.md
index 98921042b..d5c6a0079 100644
--- a/docs/VictoriaLogs/Roadmap.md
+++ b/docs/VictoriaLogs/Roadmap.md
@@ -14,12 +14,7 @@ aliases:
 
 # VictoriaLogs roadmap
 
-The [VictoriaLogs](https://docs.victoriametrics.com/victorialogs/) Preview is ready for evaluation in production.
-It is recommended running it alongside the existing solutions such as Elasticsearch and Grafana Loki
-and comparing their resource usage and usability.
-It isn't recommended migrating from existing solutions to VictoriaLogs Preview yet.
-
-The following functionality is available in VictoriaLogs Preview:
+The following functionality is available in [VictoriaLogs](https://docs.victoriametrics.com/victorialogs/):
 
 - [Data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/).
 - [Querying](https://docs.victoriametrics.com/victorialogs/querying/).
@@ -34,8 +29,6 @@ The following functionality is planned in the future versions of VictoriaLogs:
   - Fluentd
   - Journald (systemd)
   - Datadog protocol for logs
-- Add missing functionality to [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/):
-  - [Stream context](https://docs.victoriametrics.com/victorialogs/logsql/#stream-context).
 - Integration with Grafana ([partially done](https://github.com/VictoriaMetrics/victorialogs-datasource)).
 - Ability to make instant snapshots and backups in the way [similar to VictoriaMetrics](https://docs.victoriametrics.com/#how-to-work-with-snapshots).
 - Cluster version of VictoriaLogs.
diff --git a/docs/VictoriaLogs/keyConcepts.md b/docs/VictoriaLogs/keyConcepts.md
index 6fa0f7efe..a5160568b 100644
--- a/docs/VictoriaLogs/keyConcepts.md
+++ b/docs/VictoriaLogs/keyConcepts.md
@@ -247,7 +247,7 @@ This can help narrowing down and eliminating high-cardinality fields from [log s
 ### Other fields
 
 Every ingested log entry may contain arbitrary number of [fields](#data-model) additionally to [`_msg`](#message-field) and [`_time`](#time-field).
-For example, `level`, `ip`, `user_id`, `trace_id`, etc. Such fields can be used for simplifying and optimizing [search queries](#https://docs.victoriametrics.com/victorialogs/logsql/).
+For example, `level`, `ip`, `user_id`, `trace_id`, etc. Such fields can be used for simplifying and optimizing [search queries](https://docs.victoriametrics.com/victorialogs/logsql/).
 It is usually faster to search over a dedicated `trace_id` field instead of searching for the `trace_id` inside long [log message](#message-field).
 E.g. the `trace_id:="XXXX-YYYY-ZZZZ"` query usually works faster than the `_msg:"trace_id=XXXX-YYYY-ZZZZ"` query.
 
diff --git a/docs/VictoriaLogs/logsql-examples.md b/docs/VictoriaLogs/logsql-examples.md
index 971d05646..a28eec5f5 100644
--- a/docs/VictoriaLogs/logsql-examples.md
+++ b/docs/VictoriaLogs/logsql-examples.md
@@ -472,3 +472,13 @@ returns logs with the `cannot open file` phrase over the last 5 minutes:
 ```logsql
 _time:5m "cannot open file"
 ```
+
+## How to select all the logs for a particular stacktrace or panic?
+
+Use [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe) for selecting surrounding logs for the given log.
+For example, the following query selects up to 10 logs in front of every log message containing the `stacktrace` [word](https://docs.victoriametrics.com/victorialogs/logsql/#word),
+plus up to 100 logs after the given log message:
+
+```logsql
+_time:5m stacktrace | stream_context before 10 after 100
+```
diff --git a/docs/operator/vars.md b/docs/operator/vars.md
index 439daa66b..c968d9f03 100644
--- a/docs/operator/vars.md
+++ b/docs/operator/vars.md
@@ -6,6 +6,8 @@ menu:
   docs:
     parent: "operator"
     weight: 11
+aliases:
+  - /operator/vars.html
 ---
 
 <!-- this doc autogenerated - don't edit it manually -->
diff --git a/docs/stream-aggregation.md b/docs/stream-aggregation.md
index 96d64fe6d..b8adb18ed 100644
--- a/docs/stream-aggregation.md
+++ b/docs/stream-aggregation.md
@@ -973,7 +973,7 @@ specified individually per each `-remoteWrite.url`:
   # without is an optional list of labels, which must be removed from the output aggregation.
   # See https://docs.victoriametrics.com/stream-aggregation/#aggregating-by-labels
   #
-  without: [instance]
+  # without: [instance]
 
   # by is an optional list of labels, which must be preserved in the output aggregation.
   # See https://docs.victoriametrics.com/stream-aggregation/#aggregating-by-labels
diff --git a/lib/logstorage/parser.go b/lib/logstorage/parser.go
index 3ff75b5e2..7fffea336 100644
--- a/lib/logstorage/parser.go
+++ b/lib/logstorage/parser.go
@@ -350,7 +350,26 @@ func (q *Query) CanReturnLastNResults() bool {
 
 // GetFilterTimeRange returns filter time range for the given q.
 func (q *Query) GetFilterTimeRange() (int64, int64) {
-	return getFilterTimeRange(q.f)
+	switch t := q.f.(type) {
+	case *filterAnd:
+		minTimestamp := int64(math.MinInt64)
+		maxTimestamp := int64(math.MaxInt64)
+		for _, filter := range t.filters {
+			ft, ok := filter.(*filterTime)
+			if ok {
+				if ft.minTimestamp > minTimestamp {
+					minTimestamp = ft.minTimestamp
+				}
+				if ft.maxTimestamp < maxTimestamp {
+					maxTimestamp = ft.maxTimestamp
+				}
+			}
+		}
+		return minTimestamp, maxTimestamp
+	case *filterTime:
+		return t.minTimestamp, t.maxTimestamp
+	}
+	return math.MinInt64, math.MaxInt64
 }
 
 // AddTimeFilter adds global filter _time:[start ... end] to q.
diff --git a/lib/logstorage/pipe.go b/lib/logstorage/pipe.go
index f07b4b008..d119fd89f 100644
--- a/lib/logstorage/pipe.go
+++ b/lib/logstorage/pipe.go
@@ -219,6 +219,12 @@ func parsePipe(lex *lexer) (pipe, error) {
 			return nil, fmt.Errorf("cannot parse 'stats' pipe: %w", err)
 		}
 		return ps, nil
+	case lex.isKeyword("stream_context"):
+		pc, err := parsePipeStreamContext(lex)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse 'stream_context' pipe: %w", err)
+		}
+		return pc, nil
 	case lex.isKeyword("top"):
 		pt, err := parsePipeTop(lex)
 		if err != nil {
@@ -298,6 +304,7 @@ var pipeNames = func() map[string]struct{} {
 		"replace_regexp",
 		"sort",
 		"stats",
+		"stream_context",
 		"top",
 		"uniq",
 		"unpack_json",
diff --git a/lib/logstorage/pipe_sort.go b/lib/logstorage/pipe_sort.go
index df0d8f850..849a721eb 100644
--- a/lib/logstorage/pipe_sort.go
+++ b/lib/logstorage/pipe_sort.go
@@ -11,6 +11,7 @@ import (
 	"sync/atomic"
 	"unsafe"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
 )
@@ -32,6 +33,9 @@ type pipeSort struct {
 	//
 	// if zero, then all the results are returned
 	limit uint64
+
+	// The name of the field to store the row rank.
+	rankName string
 }
 
 func (ps *pipeSort) String() string {
@@ -52,6 +56,9 @@ func (ps *pipeSort) String() string {
 	if ps.limit > 0 {
 		s += fmt.Sprintf(" limit %d", ps.limit)
 	}
+	if ps.rankName != "" {
+		s += " rank as " + quoteTokenIfNeeded(ps.rankName)
+	}
 	return s
 }
 
@@ -64,6 +71,13 @@ func (ps *pipeSort) updateNeededFields(neededFields, unneededFields fieldsSet) {
 		return
 	}
 
+	if ps.rankName != "" {
+		neededFields.remove(ps.rankName)
+		if neededFields.contains("*") {
+			unneededFields.add(ps.rankName)
+		}
+	}
+
 	if len(ps.byFields) == 0 {
 		neededFields.add("*")
 		unneededFields.reset()
@@ -505,6 +519,9 @@ type pipeSortWriteContext struct {
 	rcs []resultColumn
 	br  blockResult
 
+	// buf is a temporary buffer for non-flushed block.
+	buf []byte
+
 	// rowsWritten is the total number of rows passed to writeNextRow.
 	rowsWritten uint64
 
@@ -517,6 +534,11 @@ type pipeSortWriteContext struct {
 
 func (wctx *pipeSortWriteContext) writeNextRow(shard *pipeSortProcessorShard) {
 	ps := shard.ps
+	rankName := ps.rankName
+	rankFields := 0
+	if rankName != "" {
+		rankFields = 1
+	}
 
 	rowIdx := shard.rowRefNext
 	shard.rowRefNext++
@@ -532,10 +554,10 @@ func (wctx *pipeSortWriteContext) writeNextRow(shard *pipeSortProcessorShard) {
 	byFields := ps.byFields
 	rcs := wctx.rcs
 
-	areEqualColumns := len(rcs) == len(byFields)+len(b.otherColumns)
+	areEqualColumns := len(rcs) == rankFields+len(byFields)+len(b.otherColumns)
 	if areEqualColumns {
 		for i, c := range b.otherColumns {
-			if rcs[len(byFields)+i].name != c.name {
+			if rcs[rankFields+len(byFields)+i].name != c.name {
 				areEqualColumns = false
 				break
 			}
@@ -546,6 +568,9 @@ func (wctx *pipeSortWriteContext) writeNextRow(shard *pipeSortProcessorShard) {
 		wctx.flush()
 
 		rcs = wctx.rcs[:0]
+		if rankName != "" {
+			rcs = appendResultColumnWithName(rcs, rankName)
+		}
 		for _, bf := range byFields {
 			rcs = appendResultColumnWithName(rcs, bf.name)
 		}
@@ -555,17 +580,24 @@ func (wctx *pipeSortWriteContext) writeNextRow(shard *pipeSortProcessorShard) {
 		wctx.rcs = rcs
 	}
 
+	if rankName != "" {
+		bufLen := len(wctx.buf)
+		wctx.buf = marshalUint64String(wctx.buf, wctx.rowsWritten)
+		v := bytesutil.ToUnsafeString(wctx.buf[bufLen:])
+		rcs[0].addValue(v)
+	}
+
 	br := b.br
 	byColumns := b.byColumns
 	for i := range byFields {
 		v := byColumns[i].c.getValueAtRow(br, rr.rowIdx)
-		rcs[i].addValue(v)
+		rcs[rankFields+i].addValue(v)
 		wctx.valuesLen += len(v)
 	}
 
 	for i, c := range b.otherColumns {
 		v := c.getValueAtRow(br, rr.rowIdx)
-		rcs[len(byFields)+i].addValue(v)
+		rcs[rankFields+len(byFields)+i].addValue(v)
 		wctx.valuesLen += len(v)
 	}
 
@@ -589,6 +621,7 @@ func (wctx *pipeSortWriteContext) flush() {
 	for i := range rcs {
 		rcs[i].resetValues()
 	}
+	wctx.buf = wctx.buf[:0]
 }
 
 type pipeSortProcessorShardsHeap []*pipeSortProcessorShard
@@ -763,6 +796,16 @@ func parsePipeSort(lex *lexer) (*pipeSort, error) {
 				return nil, fmt.Errorf("duplicate 'limit'; the previous one is %d; the new one is %s", ps.limit, s)
 			}
 			ps.limit = n
+		case lex.isKeyword("rank"):
+			lex.nextToken()
+			if lex.isKeyword("as") {
+				lex.nextToken()
+			}
+			rankName, err := getCompoundToken(lex)
+			if err != nil {
+				return nil, fmt.Errorf("cannot read rank field name: %s", err)
+			}
+			ps.rankName = rankName
 		default:
 			return &ps, nil
 		}
diff --git a/lib/logstorage/pipe_sort_topk.go b/lib/logstorage/pipe_sort_topk.go
index 57938b2fe..6d9d26c20 100644
--- a/lib/logstorage/pipe_sort_topk.go
+++ b/lib/logstorage/pipe_sort_topk.go
@@ -425,6 +425,9 @@ type pipeTopkWriteContext struct {
 	rcs []resultColumn
 	br  blockResult
 
+	// buf is a temporary buffer for non-flushed block.
+	buf []byte
+
 	// rowsWritten is the total number of rows passed to writeNextRow.
 	rowsWritten uint64
 
@@ -437,6 +440,11 @@ type pipeTopkWriteContext struct {
 
 func (wctx *pipeTopkWriteContext) writeNextRow(shard *pipeTopkProcessorShard) bool {
 	ps := shard.ps
+	rankName := ps.rankName
+	rankFields := 0
+	if rankName != "" {
+		rankFields = 1
+	}
 
 	rowIdx := shard.rowNext
 	shard.rowNext++
@@ -454,10 +462,10 @@ func (wctx *pipeTopkWriteContext) writeNextRow(shard *pipeTopkProcessorShard) bo
 	byFields := ps.byFields
 	rcs := wctx.rcs
 
-	areEqualColumns := len(rcs) == len(byFields)+len(r.otherColumns)
+	areEqualColumns := len(rcs) == rankFields+len(byFields)+len(r.otherColumns)
 	if areEqualColumns {
 		for i, c := range r.otherColumns {
-			if rcs[len(byFields)+i].name != c.Name {
+			if rcs[rankFields+len(byFields)+i].name != c.Name {
 				areEqualColumns = false
 				break
 			}
@@ -468,6 +476,9 @@ func (wctx *pipeTopkWriteContext) writeNextRow(shard *pipeTopkProcessorShard) bo
 		wctx.flush()
 
 		rcs = wctx.rcs[:0]
+		if rankName != "" {
+			rcs = appendResultColumnWithName(rcs, rankName)
+		}
 		for _, bf := range byFields {
 			rcs = appendResultColumnWithName(rcs, bf.name)
 		}
@@ -477,6 +488,13 @@ func (wctx *pipeTopkWriteContext) writeNextRow(shard *pipeTopkProcessorShard) bo
 		wctx.rcs = rcs
 	}
 
+	if rankName != "" {
+		bufLen := len(wctx.buf)
+		wctx.buf = marshalUint64String(wctx.buf, wctx.rowsWritten)
+		v := bytesutil.ToUnsafeString(wctx.buf[bufLen:])
+		rcs[0].addValue(v)
+	}
+
 	byColumns := r.byColumns
 	byColumnsIsTime := r.byColumnsIsTime
 	for i := range byFields {
@@ -484,13 +502,13 @@ func (wctx *pipeTopkWriteContext) writeNextRow(shard *pipeTopkProcessorShard) bo
 		if byColumnsIsTime[i] {
 			v = string(marshalTimestampRFC3339NanoString(nil, r.timestamp))
 		}
-		rcs[i].addValue(v)
+		rcs[rankFields+i].addValue(v)
 		wctx.valuesLen += len(v)
 	}
 
 	for i, c := range r.otherColumns {
 		v := c.Value
-		rcs[len(byFields)+i].addValue(v)
+		rcs[rankFields+len(byFields)+i].addValue(v)
 		wctx.valuesLen += len(v)
 	}
 
@@ -516,6 +534,7 @@ func (wctx *pipeTopkWriteContext) flush() {
 	for i := range rcs {
 		rcs[i].resetValues()
 	}
+	wctx.buf = wctx.buf[:0]
 }
 
 type pipeTopkProcessorShardsHeap []*pipeTopkProcessorShard
diff --git a/lib/logstorage/pipe_stream_context.go b/lib/logstorage/pipe_stream_context.go
new file mode 100644
index 000000000..51add0f3a
--- /dev/null
+++ b/lib/logstorage/pipe_stream_context.go
@@ -0,0 +1,545 @@
+package logstorage
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"unsafe"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
+)
+
+// pipeStreamContext processes '| stream_context ...' queries.
+//
+// See https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe
+type pipeStreamContext struct {
+	// linesBefore is the number of lines to return before the matching line
+	linesBefore int
+
+	// linesAfter is the number of lines to return after the matching line
+	linesAfter int
+}
+
+func (pc *pipeStreamContext) String() string {
+	s := "stream_context"
+	if pc.linesBefore > 0 {
+		s += fmt.Sprintf(" before %d", pc.linesBefore)
+	}
+	if pc.linesAfter > 0 {
+		s += fmt.Sprintf(" after %d", pc.linesAfter)
+	}
+	return s
+}
+
+func (pc *pipeStreamContext) canLiveTail() bool {
+	return false
+}
+
+var neededFieldsForStreamContext = []string{
+	"_time",
+	"_stream_id",
+}
+
+func (pc *pipeStreamContext) updateNeededFields(neededFields, unneededFields fieldsSet) {
+	neededFields.addFields(neededFieldsForStreamContext)
+	unneededFields.removeFields(neededFieldsForStreamContext)
+}
+
+func (pc *pipeStreamContext) optimize() {
+	// nothing to do
+}
+
+func (pc *pipeStreamContext) hasFilterInWithQuery() bool {
+	return false
+}
+
+func (pc *pipeStreamContext) initFilterInValues(_ map[string][]string, _ getFieldValuesFunc) (pipe, error) {
+	return pc, nil
+}
+
+func (pc *pipeStreamContext) newPipeProcessor(workersCount int, stopCh <-chan struct{}, cancel func(), ppNext pipeProcessor) pipeProcessor {
+	maxStateSize := int64(float64(memory.Allowed()) * 0.2)
+
+	shards := make([]pipeStreamContextProcessorShard, workersCount)
+	for i := range shards {
+		shards[i] = pipeStreamContextProcessorShard{
+			pipeStreamContextProcessorShardNopad: pipeStreamContextProcessorShardNopad{
+				pc:              pc,
+				stateSizeBudget: stateSizeBudgetChunk,
+			},
+		}
+		maxStateSize -= stateSizeBudgetChunk
+	}
+
+	pcp := &pipeStreamContextProcessor{
+		pc:     pc,
+		stopCh: stopCh,
+		cancel: cancel,
+		ppNext: ppNext,
+
+		shards: shards,
+
+		maxStateSize: maxStateSize,
+	}
+	pcp.stateSizeBudget.Store(maxStateSize)
+
+	return pcp
+}
+
+type pipeStreamContextProcessor struct {
+	pc     *pipeStreamContext
+	stopCh <-chan struct{}
+	cancel func()
+	ppNext pipeProcessor
+
+	shards []pipeStreamContextProcessorShard
+
+	getStreamRows func(streamID string, stateSizeBudget int) ([]streamContextRow, error)
+
+	maxStateSize    int64
+	stateSizeBudget atomic.Int64
+}
+
+func (pcp *pipeStreamContextProcessor) init(ctx context.Context, s *Storage, minTimestamp, maxTimestamp int64) {
+	pcp.getStreamRows = func(streamID string, stateSizeBudget int) ([]streamContextRow, error) {
+		return getStreamRows(ctx, s, streamID, minTimestamp, maxTimestamp, stateSizeBudget)
+	}
+}
+
+func getStreamRows(ctx context.Context, s *Storage, streamID string, minTimestamp, maxTimestamp int64, stateSizeBudget int) ([]streamContextRow, error) {
+	tenantID, ok := getTenantIDFromStreamIDString(streamID)
+	if !ok {
+		logger.Panicf("BUG: cannot obtain tenantID from streamID %q", streamID)
+	}
+
+	qStr := "_stream_id:" + streamID
+	q, err := ParseQuery(qStr)
+	if err != nil {
+		logger.Panicf("BUG: cannot parse query [%s]: %s", qStr, err)
+	}
+	q.AddTimeFilter(minTimestamp, maxTimestamp)
+
+	ctxWithCancel, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	var mu sync.Mutex
+	var rows []streamContextRow
+	stateSize := 0
+	writeBlock := func(_ uint, br *blockResult) {
+		mu.Lock()
+		defer mu.Unlock()
+
+		if stateSize > stateSizeBudget {
+			cancel()
+		}
+
+		cs := br.getColumns()
+		for i, timestamp := range br.timestamps {
+			fields := make([]Field, len(cs))
+			stateSize += int(unsafe.Sizeof(fields[0])) * len(fields)
+
+			for j, c := range cs {
+				v := c.getValueAtRow(br, i)
+				fields[j] = Field{
+					Name:  strings.Clone(c.name),
+					Value: strings.Clone(v),
+				}
+				stateSize += len(c.name) + len(v)
+			}
+
+			row := streamContextRow{
+				timestamp: timestamp,
+				fields:    fields,
+			}
+			stateSize += int(unsafe.Sizeof(row))
+			rows = append(rows, row)
+		}
+	}
+
+	if err := s.runQuery(ctxWithCancel, []TenantID{tenantID}, q, writeBlock); err != nil {
+		return nil, err
+	}
+	if stateSize > stateSizeBudget {
+		return nil, fmt.Errorf("more than %dMB of memory is needed for query [%s]", stateSizeBudget/(1<<20), q)
+	}
+
+	return rows, nil
+}
+
+func getTenantIDFromStreamIDString(s string) (TenantID, bool) {
+	var sid streamID
+	if !sid.tryUnmarshalFromString(s) {
+		return TenantID{}, false
+	}
+	return sid.tenantID, true
+}
+
+type pipeStreamContextProcessorShard struct {
+	pipeStreamContextProcessorShardNopad
+
+	// The padding prevents false sharing on widespread platforms with 128 mod (cache line size) = 0 .
+	_ [128 - unsafe.Sizeof(pipeStreamContextProcessorShardNopad{})%128]byte
+}
+
+type streamContextRow struct {
+	timestamp int64
+	fields    []Field
+}
+
+type pipeStreamContextProcessorShardNopad struct {
+	// pc points to the parent pipeStreamContext.
+	pc *pipeStreamContext
+
+	// m holds per-stream matching rows
+	m map[string][]streamContextRow
+
+	// stateSizeBudget is the remaining budget for the whole state size for the shard.
+	// The per-shard budget is provided in chunks from the parent pipeStreamContextProcessor.
+	stateSizeBudget int
+}
+
+// writeBlock writes br to shard.
+func (shard *pipeStreamContextProcessorShard) writeBlock(br *blockResult) {
+	m := shard.getM()
+
+	cs := br.getColumns()
+	cStreamID := br.getColumnByName("_stream_id")
+	stateSize := 0
+	for i, timestamp := range br.timestamps {
+		fields := make([]Field, len(cs))
+		stateSize += int(unsafe.Sizeof(fields[0])) * len(fields)
+
+		for j, c := range cs {
+			v := c.getValueAtRow(br, i)
+			fields[j] = Field{
+				Name:  strings.Clone(c.name),
+				Value: strings.Clone(v),
+			}
+			stateSize += len(c.name) + len(v)
+		}
+
+		row := streamContextRow{
+			timestamp: timestamp,
+			fields:    fields,
+		}
+		stateSize += int(unsafe.Sizeof(row))
+
+		streamID := cStreamID.getValueAtRow(br, i)
+		rows, ok := m[streamID]
+		if !ok {
+			stateSize += len(streamID)
+		}
+		rows = append(rows, row)
+		streamID = strings.Clone(streamID)
+		m[streamID] = rows
+	}
+
+	shard.stateSizeBudget -= stateSize
+}
+
+func (shard *pipeStreamContextProcessorShard) getM() map[string][]streamContextRow {
+	if shard.m == nil {
+		shard.m = make(map[string][]streamContextRow)
+	}
+	return shard.m
+}
+
+func (pcp *pipeStreamContextProcessor) writeBlock(workerID uint, br *blockResult) {
+	if len(br.timestamps) == 0 {
+		return
+	}
+	if pcp.pc.linesBefore <= 0 && pcp.pc.linesAfter <= 0 {
+		// Fast path - there is no need to fetch stream context.
+		pcp.ppNext.writeBlock(workerID, br)
+		return
+	}
+
+	shard := &pcp.shards[workerID]
+
+	for shard.stateSizeBudget < 0 {
+		// steal some budget for the state size from the global budget.
+		remaining := pcp.stateSizeBudget.Add(-stateSizeBudgetChunk)
+		if remaining < 0 {
+			// The state size is too big. Stop processing data in order to avoid OOM crash.
+			if remaining+stateSizeBudgetChunk >= 0 {
+				// Notify worker goroutines to stop calling writeBlock() in order to save CPU time.
+				pcp.cancel()
+			}
+			return
+		}
+		shard.stateSizeBudget += stateSizeBudgetChunk
+	}
+
+	shard.writeBlock(br)
+}
+
+func (pcp *pipeStreamContextProcessor) flush() error {
+	if pcp.pc.linesBefore <= 0 && pcp.pc.linesAfter <= 0 {
+		// Fast path - nothing to do.
+		return nil
+	}
+
+	n := pcp.stateSizeBudget.Load()
+	if n <= 0 {
+		return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pcp.pc.String(), pcp.maxStateSize/(1<<20))
+	}
+	if n > math.MaxInt {
+		logger.Panicf("BUG: stateSizeBudget shouldn't exceed math.MaxInt=%v; got %d", math.MaxInt, n)
+	}
+	stateSizeBudget := int(n)
+
+	// merge state across shards
+	shards := pcp.shards
+	m := shards[0].getM()
+	shards = shards[1:]
+	for i := range shards {
+		if needStop(pcp.stopCh) {
+			return nil
+		}
+
+		for streamID, rowsSrc := range shards[i].getM() {
+			rows, ok := m[streamID]
+			if !ok {
+				m[streamID] = rowsSrc
+			} else {
+				m[streamID] = append(rows, rowsSrc...)
+			}
+		}
+	}
+
+	// write result
+	wctx := &pipeStreamContextWriteContext{
+		pcp: pcp,
+	}
+
+	for streamID, rows := range m {
+		streamRows, err := pcp.getStreamRows(streamID, stateSizeBudget)
+		if err != nil {
+			return fmt.Errorf("cannot read rows for _stream_id=%q: %w", streamID, err)
+		}
+		if needStop(pcp.stopCh) {
+			return nil
+		}
+		if err := wctx.writeStreamContextRows(streamID, streamRows, rows, pcp.pc.linesBefore, pcp.pc.linesAfter); err != nil {
+			return fmt.Errorf("cannot obtain context rows for _stream_id=%q: %w", streamID, err)
+		}
+	}
+
+	wctx.flush()
+
+	return nil
+}
+
+func (wctx *pipeStreamContextWriteContext) writeStreamContextRows(streamID string, streamRows, rows []streamContextRow, linesBefore, linesAfter int) error {
+	sortStreamContextRows(streamRows)
+	sortStreamContextRows(rows)
+
+	idxNext := 0
+	for i := range rows {
+		r := &rows[i]
+		idx := getStreamContextRowIdx(streamRows, r)
+		if idx < 0 {
+			// This error may happen when streamRows became out of sync with rows.
+			// For example, when some streamRows were deleted after obtaining rows.
+			return fmt.Errorf("missing row for timestamp=%d; len(streamRows)=%d, len(rows)=%d; re-execute the query", r.timestamp, len(streamRows), len(rows))
+		}
+
+		idxStart := idx - linesBefore
+		if idxStart < idxNext {
+			idxStart = idxNext
+		} else if idxNext > 0 && idxStart > idxNext {
+			// Write delimiter row between multiple contexts in the same stream.
+			// This simplifies investigation of the returned logs.
+			fields := []Field{
+				{
+					Name:  "_time",
+					Value: string(marshalTimestampRFC3339NanoString(nil, r.timestamp+1)),
+				},
+				{
+					Name:  "_stream_id",
+					Value: streamID,
+				},
+				{
+					Name:  "_stream",
+					Value: getFieldValue(r.fields, "_stream"),
+				},
+				{
+					Name:  "_msg",
+					Value: "---",
+				},
+			}
+			wctx.writeRow(fields)
+		}
+		for idxStart < idx {
+			wctx.writeRow(streamRows[idxStart].fields)
+			idxStart++
+		}
+
+		if idx >= idxNext {
+			wctx.writeRow(streamRows[idx].fields)
+			idxNext = idx + 1
+		}
+
+		idxEnd := idx + 1 + linesAfter
+		for idxNext < idxEnd && idxNext < len(streamRows) {
+			wctx.writeRow(streamRows[idxNext].fields)
+			idxNext++
+		}
+
+		if idxNext >= len(streamRows) {
+			break
+		}
+	}
+
+	return nil
+}
+
+func getStreamContextRowIdx(rows []streamContextRow, r *streamContextRow) int {
+	n := sort.Search(len(rows), func(i int) bool {
+		return rows[i].timestamp >= r.timestamp
+	})
+	if n == len(rows) {
+		return -1
+	}
+
+	equalFields := func(fields []Field) bool {
+		for _, f := range r.fields {
+			if f.Value != getFieldValue(fields, f.Name) {
+				return false
+			}
+		}
+		return true
+	}
+
+	for rows[n].timestamp == r.timestamp && !equalFields(rows[n].fields) {
+		n++
+		if n >= len(rows) {
+			return -1
+		}
+	}
+	if rows[n].timestamp != r.timestamp {
+		return -1
+	}
+	return n
+}
+
+func sortStreamContextRows(rows []streamContextRow) {
+	sort.Slice(rows, func(i, j int) bool {
+		return rows[i].timestamp < rows[j].timestamp
+	})
+}
+
+type pipeStreamContextWriteContext struct {
+	pcp *pipeStreamContextProcessor
+	rcs []resultColumn
+	br  blockResult
+
+	// rowsCount is the number of rows in the current block
+	rowsCount int
+
+	// valuesLen is the total length of values in the current block
+	valuesLen int
+}
+
+func (wctx *pipeStreamContextWriteContext) writeRow(rowFields []Field) {
+	rcs := wctx.rcs
+
+	areEqualColumns := len(rcs) == len(rowFields)
+	if areEqualColumns {
+		for i, f := range rowFields {
+			if rcs[i].name != f.Name {
+				areEqualColumns = false
+				break
+			}
+		}
+	}
+	if !areEqualColumns {
+		// send the current block to ppNext and construct a block with new set of columns
+		wctx.flush()
+
+		rcs = wctx.rcs[:0]
+		for _, f := range rowFields {
+			rcs = appendResultColumnWithName(rcs, f.Name)
+		}
+		wctx.rcs = rcs
+	}
+
+	for i, f := range rowFields {
+		v := f.Value
+		rcs[i].addValue(v)
+		wctx.valuesLen += len(v)
+	}
+
+	wctx.rowsCount++
+	if wctx.valuesLen >= 1_000_000 {
+		wctx.flush()
+	}
+}
+
+func (wctx *pipeStreamContextWriteContext) flush() {
+	rcs := wctx.rcs
+	br := &wctx.br
+
+	wctx.valuesLen = 0
+
+	// Flush rcs to ppNext
+	br.setResultColumns(rcs, wctx.rowsCount)
+	wctx.rowsCount = 0
+	wctx.pcp.ppNext.writeBlock(0, br)
+	br.reset()
+	for i := range rcs {
+		rcs[i].resetValues()
+	}
+}
+
+func parsePipeStreamContext(lex *lexer) (*pipeStreamContext, error) {
+	if !lex.isKeyword("stream_context") {
+		return nil, fmt.Errorf("expecting 'stream_context'; got %q", lex.token)
+	}
+	lex.nextToken()
+
+	linesBefore := 0
+	beforeSet := false
+	if lex.isKeyword("before") {
+		lex.nextToken()
+		f, s, err := parseNumber(lex)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse 'before' value in 'stream_context': %w", err)
+		}
+		if f < 0 {
+			return nil, fmt.Errorf("'before' value cannot be smaller than 0; got %q", s)
+		}
+		linesBefore = int(f)
+		beforeSet = true
+	}
+
+	linesAfter := 0
+	afterSet := false
+	if lex.isKeyword("after") {
+		lex.nextToken()
+		f, s, err := parseNumber(lex)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse 'after' value in 'stream_context': %w", err)
+		}
+		if f < 0 {
+			return nil, fmt.Errorf("'after' value cannot be smaller than 0; got %q", s)
+		}
+		linesAfter = int(f)
+		afterSet = true
+	}
+
+	if !beforeSet && !afterSet {
+		return nil, fmt.Errorf("missing 'before N' or 'after N' in 'stream_context'")
+	}
+
+	pc := &pipeStreamContext{
+		linesBefore: linesBefore,
+		linesAfter:  linesAfter,
+	}
+	return pc, nil
+}
diff --git a/lib/logstorage/rows.go b/lib/logstorage/rows.go
index c165641a3..8b444a768 100644
--- a/lib/logstorage/rows.go
+++ b/lib/logstorage/rows.go
@@ -79,6 +79,15 @@ func (f *Field) marshalToLogfmt(dst []byte) []byte {
 	return dst
 }
 
+func getFieldValue(fields []Field, name string) string {
+	for _, f := range fields {
+		if f.Name == name {
+			return f.Value
+		}
+	}
+	return ""
+}
+
 func needLogfmtQuoting(s string) bool {
 	for _, c := range s {
 		if !isTokenRune(c) {
diff --git a/lib/logstorage/storage_search.go b/lib/logstorage/storage_search.go
index 4dcaff65e..ba82ccd0d 100644
--- a/lib/logstorage/storage_search.go
+++ b/lib/logstorage/storage_search.go
@@ -22,6 +22,12 @@ type genericSearchOptions struct {
 	// If it is empty, then the search is performed by tenantIDs
 	streamIDs []streamID
 
+	// minTimestamp is the minimum timestamp for the search
+	minTimestamp int64
+
+	// maxTimestamp is the maximum timestamp for the search
+	maxTimestamp int64
+
 	// filter is the filter to use for the search
 	filter filter
 
@@ -110,10 +116,14 @@ func (s *Storage) runQuery(ctx context.Context, tenantIDs []TenantID, q *Query,
 		return streamIDs[i].less(&streamIDs[j])
 	})
 
+	minTimestamp, maxTimestamp := q.GetFilterTimeRange()
+
 	neededColumnNames, unneededColumnNames := q.getNeededColumns()
 	so := &genericSearchOptions{
 		tenantIDs:           tenantIDs,
 		streamIDs:           streamIDs,
+		minTimestamp:        minTimestamp,
+		maxTimestamp:        maxTimestamp,
 		filter:              q.f,
 		neededColumnNames:   neededColumnNames,
 		unneededColumnNames: unneededColumnNames,
@@ -127,10 +137,21 @@ func (s *Storage) runQuery(ctx context.Context, tenantIDs []TenantID, q *Query,
 	stopCh := ctx.Done()
 	cancels := make([]func(), len(q.pipes))
 	pps := make([]pipeProcessor, len(q.pipes))
+
+	var errPipe error
 	for i := len(q.pipes) - 1; i >= 0; i-- {
 		p := q.pipes[i]
 		ctxChild, cancel := context.WithCancel(ctx)
 		pp = p.newPipeProcessor(workersCount, stopCh, cancel, pp)
+
+		pcp, ok := pp.(*pipeStreamContextProcessor)
+		if ok {
+			pcp.init(ctx, s, minTimestamp, maxTimestamp)
+			if i > 0 {
+				errPipe = fmt.Errorf("[%s] pipe must go after [%s] filter; now it goes after the [%s] pipe", p, q.f, q.pipes[i-1])
+			}
+		}
+
 		stopCh = ctxChild.Done()
 		ctx = ctxChild
 
@@ -138,7 +159,9 @@ func (s *Storage) runQuery(ctx context.Context, tenantIDs []TenantID, q *Query,
 		pps[i] = pp
 	}
 
-	s.search(workersCount, so, stopCh, pp.writeBlock)
+	if errPipe == nil {
+		s.search(workersCount, so, stopCh, pp.writeBlock)
+	}
 
 	var errFlush error
 	for i, pp := range pps {
@@ -151,6 +174,11 @@ func (s *Storage) runQuery(ctx context.Context, tenantIDs []TenantID, q *Query,
 	if err := ppMain.flush(); err != nil && errFlush == nil {
 		errFlush = err
 	}
+
+	if errPipe != nil {
+		return errPipe
+	}
+
 	return errFlush
 }
 
@@ -642,19 +670,15 @@ func (s *Storage) search(workersCount int, so *genericSearchOptions, stopCh <-ch
 		}(uint(i))
 	}
 
-	// Obtain time range from so.filter
-	f := so.filter
-	minTimestamp, maxTimestamp := getFilterTimeRange(f)
-
 	// Select partitions according to the selected time range
 	s.partitionsLock.Lock()
 	ptws := s.partitions
-	minDay := minTimestamp / nsecPerDay
+	minDay := so.minTimestamp / nsecPerDay
 	n := sort.Search(len(ptws), func(i int) bool {
 		return ptws[i].day >= minDay
 	})
 	ptws = ptws[n:]
-	maxDay := maxTimestamp / nsecPerDay
+	maxDay := so.maxTimestamp / nsecPerDay
 	n = sort.Search(len(ptws), func(i int) bool {
 		return ptws[i].day > maxDay
 	})
@@ -665,8 +689,7 @@ func (s *Storage) search(workersCount int, so *genericSearchOptions, stopCh <-ch
 	s.partitionsLock.Unlock()
 
 	// Obtain common filterStream from f
-	var sf *StreamFilter
-	sf, f = getCommonStreamFilter(f)
+	sf, f := getCommonStreamFilter(so.filter)
 
 	// Schedule concurrent search across matching partitions.
 	psfs := make([]partitionSearchFinalizer, len(ptws))
@@ -675,7 +698,7 @@ func (s *Storage) search(workersCount int, so *genericSearchOptions, stopCh <-ch
 		partitionSearchConcurrencyLimitCh <- struct{}{}
 		wgSearchers.Add(1)
 		go func(idx int, pt *partition) {
-			psfs[idx] = pt.search(minTimestamp, maxTimestamp, sf, f, so, workCh, stopCh)
+			psfs[idx] = pt.search(sf, f, so, workCh, stopCh)
 			wgSearchers.Done()
 			<-partitionSearchConcurrencyLimitCh
 		}(i, ptw.pt)
@@ -704,7 +727,7 @@ var partitionSearchConcurrencyLimitCh = make(chan struct{}, cgroup.AvailableCPUs
 
 type partitionSearchFinalizer func()
 
-func (pt *partition) search(minTimestamp, maxTimestamp int64, sf *StreamFilter, f filter, so *genericSearchOptions, workCh chan<- *blockSearchWorkBatch, stopCh <-chan struct{}) partitionSearchFinalizer {
+func (pt *partition) search(sf *StreamFilter, f filter, so *genericSearchOptions, workCh chan<- *blockSearchWorkBatch, stopCh <-chan struct{}) partitionSearchFinalizer {
 	if needStop(stopCh) {
 		// Do not spend CPU time on search, since it is already stopped.
 		return func() {}
@@ -728,8 +751,8 @@ func (pt *partition) search(minTimestamp, maxTimestamp int64, sf *StreamFilter,
 	soInternal := &searchOptions{
 		tenantIDs:           tenantIDs,
 		streamIDs:           streamIDs,
-		minTimestamp:        minTimestamp,
-		maxTimestamp:        maxTimestamp,
+		minTimestamp:        so.minTimestamp,
+		maxTimestamp:        so.maxTimestamp,
 		filter:              f,
 		neededColumnNames:   so.neededColumnNames,
 		unneededColumnNames: so.unneededColumnNames,
@@ -1095,29 +1118,6 @@ func getCommonStreamFilter(f filter) (*StreamFilter, filter) {
 	return nil, f
 }
 
-func getFilterTimeRange(f filter) (int64, int64) {
-	switch t := f.(type) {
-	case *filterAnd:
-		minTimestamp := int64(math.MinInt64)
-		maxTimestamp := int64(math.MaxInt64)
-		for _, filter := range t.filters {
-			ft, ok := filter.(*filterTime)
-			if ok {
-				if ft.minTimestamp > minTimestamp {
-					minTimestamp = ft.minTimestamp
-				}
-				if ft.maxTimestamp < maxTimestamp {
-					maxTimestamp = ft.maxTimestamp
-				}
-			}
-		}
-		return minTimestamp, maxTimestamp
-	case *filterTime:
-		return t.minTimestamp, t.maxTimestamp
-	}
-	return math.MinInt64, math.MaxInt64
-}
-
 func forEachStreamField(streams []ValueWithHits, f func(f Field, hits uint64)) {
 	var fields []Field
 	for i := range streams {
diff --git a/lib/promrelabel/if_expression.go b/lib/promrelabel/if_expression.go
index 7d102a1f6..90790c244 100644
--- a/lib/promrelabel/if_expression.go
+++ b/lib/promrelabel/if_expression.go
@@ -1,6 +1,7 @@
 package promrelabel
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 
@@ -143,7 +144,13 @@ func (ie *IfExpression) String() string {
 	if len(ie.ies) == 1 {
 		return ie.ies[0].String()
 	}
-	return fmt.Sprintf("%s", ie.ies)
+	var buf bytes.Buffer
+	buf.WriteString(ie.ies[0].String())
+	for _, e := range ie.ies[1:] {
+		buf.WriteString(",")
+		buf.WriteString(e.String())
+	}
+	return buf.String()
 }
 
 type ifExpression struct {
diff --git a/lib/streamaggr/deduplicator.go b/lib/streamaggr/deduplicator.go
index 0212bfd9e..c49580dcc 100644
--- a/lib/streamaggr/deduplicator.go
+++ b/lib/streamaggr/deduplicator.go
@@ -1,6 +1,7 @@
 package streamaggr
 
 import (
+	"fmt"
 	"slices"
 	"sync"
 	"time"
@@ -35,7 +36,7 @@ type Deduplicator struct {
 // Common case is to drop `replica`-like labels from samples received from HA datasources.
 //
 // MustStop must be called on the returned deduplicator in order to free up occupied resources.
-func NewDeduplicator(pushFunc PushFunc, dedupInterval time.Duration, dropLabels []string) *Deduplicator {
+func NewDeduplicator(pushFunc PushFunc, dedupInterval time.Duration, dropLabels []string, alias string) *Deduplicator {
 	d := &Deduplicator{
 		da:         newDedupAggr(),
 		dropLabels: dropLabels,
@@ -45,15 +46,17 @@ func NewDeduplicator(pushFunc PushFunc, dedupInterval time.Duration, dropLabels
 	}
 
 	ms := d.ms
-	_ = ms.NewGauge(`vm_streamaggr_dedup_state_size_bytes`, func() float64 {
+
+	metricLabels := fmt.Sprintf(`url=%q`, alias)
+	_ = ms.NewGauge(fmt.Sprintf(`vm_streamaggr_dedup_state_size_bytes{%s}`, metricLabels), func() float64 {
 		return float64(d.da.sizeBytes())
 	})
-	_ = ms.NewGauge(`vm_streamaggr_dedup_state_items_count`, func() float64 {
+	_ = ms.NewGauge(fmt.Sprintf(`vm_streamaggr_dedup_state_items_count{%s}`, metricLabels), func() float64 {
 		return float64(d.da.itemsCount())
 	})
 
-	d.dedupFlushDuration = ms.GetOrCreateHistogram(`vm_streamaggr_dedup_flush_duration_seconds`)
-	d.dedupFlushTimeouts = ms.GetOrCreateCounter(`vm_streamaggr_dedup_flush_timeouts_total`)
+	d.dedupFlushDuration = ms.GetOrCreateHistogram(fmt.Sprintf(`vm_streamaggr_dedup_flush_duration_seconds{%s}`, metricLabels))
+	d.dedupFlushTimeouts = ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_dedup_flush_timeouts_total{%s}`, metricLabels))
 
 	metrics.RegisterSet(ms)
 
diff --git a/lib/streamaggr/histogram_bucket.go b/lib/streamaggr/histogram_bucket.go
index 4cce32c2d..afb7ac8b7 100644
--- a/lib/streamaggr/histogram_bucket.go
+++ b/lib/streamaggr/histogram_bucket.go
@@ -66,8 +66,9 @@ func (as *histogramBucketAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *histogramBucketAggrState) removeOldEntries(currentTime uint64) {
+func (as *histogramBucketAggrState) removeOldEntries(ctx *flushCtx, currentTime uint64) {
 	m := &as.m
+	var staleOutputSamples int
 	m.Range(func(k, v interface{}) bool {
 		sv := v.(*histogramBucketStateValue)
 
@@ -76,6 +77,7 @@ func (as *histogramBucketAggrState) removeOldEntries(currentTime uint64) {
 		if deleted {
 			// Mark the current entry as deleted
 			sv.deleted = deleted
+			staleOutputSamples++
 		}
 		sv.mu.Unlock()
 
@@ -84,14 +86,14 @@ func (as *histogramBucketAggrState) removeOldEntries(currentTime uint64) {
 		}
 		return true
 	})
+	ctx.a.staleOutputSamples["histogram_bucket"].Add(staleOutputSamples)
 }
 
-func (as *histogramBucketAggrState) flushState(ctx *flushCtx, resetState bool) {
-	_ = resetState // it isn't used here
+func (as *histogramBucketAggrState) flushState(ctx *flushCtx, _ bool) {
 	currentTime := fasttime.UnixTimestamp()
 	currentTimeMsec := int64(currentTime) * 1000
 
-	as.removeOldEntries(currentTime)
+	as.removeOldEntries(ctx, currentTime)
 
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
diff --git a/lib/streamaggr/rate.go b/lib/streamaggr/rate.go
index 33d81bc2f..07fd2783b 100644
--- a/lib/streamaggr/rate.go
+++ b/lib/streamaggr/rate.go
@@ -108,6 +108,7 @@ func (as *rateAggrState) pushSamples(samples []pushSample) {
 func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
 	currentTime := fasttime.UnixTimestamp()
 	currentTimeMsec := int64(currentTime) * 1000
+	var staleOutputSamples, staleInputSamples int
 
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
@@ -120,6 +121,7 @@ func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
 			// Mark the current entry as deleted
 			sv.deleted = deleted
 			sv.mu.Unlock()
+			staleOutputSamples++
 			m.Delete(k)
 			return true
 		}
@@ -130,6 +132,7 @@ func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
 		for k1, v1 := range lvs {
 			if currentTime > v1.deleteDeadline {
 				delete(lvs, k1)
+				staleInputSamples++
 				continue
 			}
 			rateInterval := v1.timestamp - v1.prevTimestamp
@@ -153,4 +156,6 @@ func (as *rateAggrState) flushState(ctx *flushCtx, _ bool) {
 		ctx.appendSeries(key, as.suffix, currentTimeMsec, rate)
 		return true
 	})
+	ctx.a.staleOutputSamples[as.suffix].Add(staleOutputSamples)
+	ctx.a.staleInputSamples[as.suffix].Add(staleInputSamples)
 }
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index 34ad118cf..831cd500f 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -47,6 +47,9 @@ var supportedOutputs = []string{
 	"quantiles(phi1, ..., phiN)",
 }
 
+// maxLabelValueLen is maximum match expression label value length in stream aggregation metrics
+const maxLabelValueLen = 64
+
 var (
 	// lc contains information about all compressed labels for streaming aggregation
 	lc promutils.LabelsCompressor
@@ -65,7 +68,7 @@ var (
 // opts can contain additional options. If opts is nil, then default options are used.
 //
 // The returned Aggregators must be stopped with MustStop() when no longer needed.
-func LoadFromFile(path string, pushFunc PushFunc, opts *Options) (*Aggregators, error) {
+func LoadFromFile(path string, pushFunc PushFunc, opts Options) (*Aggregators, error) {
 	data, err := fscore.ReadFileOrHTTP(path)
 	if err != nil {
 		return nil, fmt.Errorf("cannot load aggregators: %w", err)
@@ -133,6 +136,12 @@ type Options struct {
 	//
 	// This option can be overridden individually per each aggregation via ignore_first_intervals option.
 	IgnoreFirstIntervals int
+
+	// Alias is name or url of remote write context
+	Alias string
+
+	// aggrID is aggregators id number starting from 1, which is used in metrics labels
+	aggrID int
 }
 
 // Config is a configuration for a single stream aggregation.
@@ -243,7 +252,7 @@ type Aggregators struct {
 	ms *metrics.Set
 }
 
-func newAggregatorsFromData(data []byte, pushFunc PushFunc, opts *Options) (*Aggregators, error) {
+func newAggregatorsFromData(data []byte, pushFunc PushFunc, opts Options) (*Aggregators, error) {
 	var cfgs []*Config
 	if err := yaml.UnmarshalStrict(data, &cfgs); err != nil {
 		return nil, fmt.Errorf("cannot parse stream aggregation config: %w", err)
@@ -252,6 +261,7 @@ func newAggregatorsFromData(data []byte, pushFunc PushFunc, opts *Options) (*Agg
 	ms := metrics.NewSet()
 	as := make([]*aggregator, len(cfgs))
 	for i, cfg := range cfgs {
+		opts.aggrID = i + 1
 		a, err := newAggregator(cfg, pushFunc, ms, opts)
 		if err != nil {
 			// Stop already initialized aggregators before returning the error.
@@ -267,7 +277,8 @@ func newAggregatorsFromData(data []byte, pushFunc PushFunc, opts *Options) (*Agg
 		logger.Panicf("BUG: cannot marshal the provided configs: %s", err)
 	}
 
-	_ = ms.NewGauge(`vm_streamaggr_dedup_state_size_bytes`, func() float64 {
+	metricLabels := fmt.Sprintf("url=%q", opts.Alias)
+	_ = ms.NewGauge(fmt.Sprintf(`vm_streamaggr_dedup_state_size_bytes{%s}`, metricLabels), func() float64 {
 		n := uint64(0)
 		for _, aggr := range as {
 			if aggr.da != nil {
@@ -276,7 +287,7 @@ func newAggregatorsFromData(data []byte, pushFunc PushFunc, opts *Options) (*Agg
 		}
 		return float64(n)
 	})
-	_ = ms.NewGauge(`vm_streamaggr_dedup_state_items_count`, func() float64 {
+	_ = ms.NewGauge(fmt.Sprintf(`vm_streamaggr_dedup_state_items_count{%s}`, metricLabels), func() float64 {
 		n := uint64(0)
 		for _, aggr := range as {
 			if aggr.da != nil {
@@ -395,6 +406,10 @@ type aggregator struct {
 	dedupFlushTimeouts *metrics.Counter
 	ignoredOldSamples  *metrics.Counter
 	ignoredNanSamples  *metrics.Counter
+	matchedSamples     *metrics.Counter
+	staleInputSamples  map[string]*metrics.Counter
+	staleOutputSamples map[string]*metrics.Counter
+	flushedSamples     map[string]*metrics.Counter
 }
 
 type aggrState interface {
@@ -414,11 +429,7 @@ type PushFunc func(tss []prompbmarshal.TimeSeries)
 // opts can contain additional options. If opts is nil, then default options are used.
 //
 // The returned aggregator must be stopped when no longer needed by calling MustStop().
-func newAggregator(cfg *Config, pushFunc PushFunc, ms *metrics.Set, opts *Options) (*aggregator, error) {
-	if opts == nil {
-		opts = &Options{}
-	}
-
+func newAggregator(cfg *Config, pushFunc PushFunc, ms *metrics.Set, opts Options) (*aggregator, error) {
 	// check cfg.Interval
 	if cfg.Interval == "" {
 		return nil, fmt.Errorf("missing `interval` option")
@@ -586,14 +597,26 @@ func newAggregator(cfg *Config, pushFunc PushFunc, ms *metrics.Set, opts *Option
 
 	// initialize suffix to add to metric names after aggregation
 	suffix := ":" + cfg.Interval
+	group := "none"
 	if labels := removeUnderscoreName(by); len(labels) > 0 {
+		group = fmt.Sprintf("by: %s", strings.Join(labels, ","))
 		suffix += fmt.Sprintf("_by_%s", strings.Join(labels, "_"))
 	}
 	if labels := removeUnderscoreName(without); len(labels) > 0 {
+		group = fmt.Sprintf("without: %s", strings.Join(labels, ","))
 		suffix += fmt.Sprintf("_without_%s", strings.Join(labels, "_"))
 	}
 	suffix += "_"
 
+	outputs := strings.Join(cfg.Outputs, ",")
+
+	matchExpr := cfg.Match.String()
+	if len(matchExpr) > maxLabelValueLen {
+		matchExpr = matchExpr[:maxLabelValueLen-3] + "..."
+	}
+
+	metricLabels := fmt.Sprintf(`match=%q, group=%q, url=%q, position="%d"`, matchExpr, group, opts.Alias, opts.aggrID)
+
 	// initialize the aggregator
 	a := &aggregator{
 		match: cfg.Match,
@@ -615,14 +638,27 @@ func newAggregator(cfg *Config, pushFunc PushFunc, ms *metrics.Set, opts *Option
 
 		stopCh: make(chan struct{}),
 
-		flushDuration:      ms.GetOrCreateHistogram(`vm_streamaggr_flush_duration_seconds`),
-		dedupFlushDuration: ms.GetOrCreateHistogram(`vm_streamaggr_dedup_flush_duration_seconds`),
-		samplesLag:         ms.GetOrCreateHistogram(`vm_streamaggr_samples_lag_seconds`),
-
-		flushTimeouts:      ms.GetOrCreateCounter(`vm_streamaggr_flush_timeouts_total`),
-		dedupFlushTimeouts: ms.GetOrCreateCounter(`vm_streamaggr_dedup_flush_timeouts_total`),
-		ignoredNanSamples:  ms.GetOrCreateCounter(`vm_streamaggr_ignored_samples_total{reason="nan"}`),
-		ignoredOldSamples:  ms.GetOrCreateCounter(`vm_streamaggr_ignored_samples_total{reason="too_old"}`),
+		flushDuration:      ms.GetOrCreateHistogram(fmt.Sprintf(`vm_streamaggr_flush_duration_seconds{outputs=%q, %s}`, outputs, metricLabels)),
+		dedupFlushDuration: ms.GetOrCreateHistogram(fmt.Sprintf(`vm_streamaggr_dedup_flush_duration_seconds{outputs=%q, %s}`, outputs, metricLabels)),
+		samplesLag:         ms.GetOrCreateHistogram(fmt.Sprintf(`vm_streamaggr_samples_lag_seconds{outputs=%q, %s}`, outputs, metricLabels)),
+
+		matchedSamples:     ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_matched_samples_total{outputs=%q, %s}`, outputs, metricLabels)),
+		flushTimeouts:      ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_flush_timeouts_total{outputs=%q, %s}`, outputs, metricLabels)),
+		dedupFlushTimeouts: ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_dedup_flush_timeouts_total{outputs=%q, %s}`, outputs, metricLabels)),
+		ignoredNanSamples:  ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_ignored_samples_total{reason="nan", outputs=%q, %s}`, outputs, metricLabels)),
+		ignoredOldSamples:  ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_ignored_samples_total{reason="too_old", outputs=%q, %s}`, outputs, metricLabels)),
+		staleInputSamples:  make(map[string]*metrics.Counter, len(cfg.Outputs)),
+		staleOutputSamples: make(map[string]*metrics.Counter, len(cfg.Outputs)),
+		flushedSamples:     make(map[string]*metrics.Counter, len(cfg.Outputs)),
+	}
+	for _, output := range cfg.Outputs {
+		// Removing output args for metric label value in outputs like quantile(arg1, arg2)
+		if ri := strings.IndexRune(output, '('); ri >= 0 {
+			output = output[:ri]
+		}
+		a.staleInputSamples[output] = ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_stale_samples_total{key="input", output=%q, %s}`, output, metricLabels))
+		a.staleOutputSamples[output] = ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_stale_samples_total{key="output", output=%q, %s}`, output, metricLabels))
+		a.flushedSamples[output] = ms.GetOrCreateCounter(fmt.Sprintf(`vm_streamaggr_flushed_samples_total{output=%q, %s}`, output, metricLabels))
 	}
 	if dedupInterval > 0 {
 		a.da = newDedupAggr()
@@ -821,8 +857,7 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 	dropLabels := a.dropInputLabels
 	ignoreOldSamples := a.ignoreOldSamples
 	minTimestamp := a.minTimestamp.Load()
-	var totalLag int64
-	var totalSamples int
+	var maxLag int64
 	for idx, ts := range tss {
 		if !a.match.Match(ts.Labels) {
 			continue
@@ -865,8 +900,9 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 				// Skip old samples outside the current aggregation interval
 				continue
 			}
-			totalLag += now - sample.Timestamp
-			totalSamples++
+			if maxLag < now-sample.Timestamp {
+				maxLag = now - sample.Timestamp
+			}
 			samples = append(samples, pushSample{
 				key:       key,
 				value:     sample.Value,
@@ -874,8 +910,9 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 			})
 		}
 	}
-	if totalSamples > 0 {
-		a.samplesLag.Update(float64(totalLag/int64(totalSamples)) / 1000)
+	if len(samples) > 0 {
+		a.matchedSamples.Add(len(samples))
+		a.samplesLag.Update(float64(maxLag) / 1_000)
 	}
 	ctx.samples = samples
 	ctx.buf = buf
@@ -1088,6 +1125,7 @@ func (ctx *flushCtx) appendSeries(key, suffix string, timestamp int64, value flo
 		Labels:  ctx.labels[labelsLen:],
 		Samples: ctx.samples[samplesLen:],
 	})
+	ctx.a.flushedSamples[suffix].Add(len(ctx.tss))
 
 	// Limit the maximum length of ctx.tss in order to limit memory usage.
 	if len(ctx.tss) >= 10_000 {
@@ -1115,6 +1153,7 @@ func (ctx *flushCtx) appendSeriesWithExtraLabel(key, suffix string, timestamp in
 		Labels:  ctx.labels[labelsLen:],
 		Samples: ctx.samples[samplesLen:],
 	})
+	ctx.a.flushedSamples[suffix].Add(len(ctx.tss))
 }
 
 func addMetricSuffix(labels []prompbmarshal.Label, offset int, firstSuffix, lastSuffix string) []prompbmarshal.Label {
diff --git a/lib/streamaggr/total.go b/lib/streamaggr/total.go
index 365fdb35d..6598d7c15 100644
--- a/lib/streamaggr/total.go
+++ b/lib/streamaggr/total.go
@@ -124,8 +124,9 @@ func (as *totalAggrState) pushSamples(samples []pushSample) {
 	}
 }
 
-func (as *totalAggrState) removeOldEntries(currentTime uint64) {
+func (as *totalAggrState) removeOldEntries(ctx *flushCtx, currentTime uint64) {
 	m := &as.m
+	var staleInputSamples, staleOutputSamples int
 	m.Range(func(k, v interface{}) bool {
 		sv := v.(*totalStateValue)
 
@@ -134,12 +135,14 @@ func (as *totalAggrState) removeOldEntries(currentTime uint64) {
 		if deleted {
 			// Mark the current entry as deleted
 			sv.deleted = deleted
+			staleOutputSamples++
 		} else {
 			// Delete outdated entries in sv.lastValues
 			m := sv.lastValues
 			for k1, v1 := range m {
 				if currentTime > v1.deleteDeadline {
 					delete(m, k1)
+					staleInputSamples++
 				}
 			}
 		}
@@ -150,13 +153,15 @@ func (as *totalAggrState) removeOldEntries(currentTime uint64) {
 		}
 		return true
 	})
+	ctx.a.staleInputSamples[as.suffix].Add(staleInputSamples)
+	ctx.a.staleOutputSamples[as.suffix].Add(staleOutputSamples)
 }
 
 func (as *totalAggrState) flushState(ctx *flushCtx, resetState bool) {
 	currentTime := fasttime.UnixTimestamp()
 	currentTimeMsec := int64(currentTime) * 1000
 
-	as.removeOldEntries(currentTime)
+	as.removeOldEntries(ctx, currentTime)
 
 	m := &as.m
 	m.Range(func(k, v interface{}) bool {
diff --git a/snap/local/Dockerfile b/snap/local/Dockerfile
deleted file mode 100644
index 8faf32ea9..000000000
--- a/snap/local/Dockerfile
+++ /dev/null
@@ -1,9 +0,0 @@
-FROM snapcore/snapcraft:stable
-ARG GO_VERSION
-RUN apt-get update && apt-get install  -y git make wget  binutils build-essential bzip2 cpp cpp-5 dpkg-dev fakeroot g++ g++-5 gcc gcc-5
-
-RUN cd /usr/local &&\
-    wget https://dl.google.com/go/go$GO_VERSION.linux-amd64.tar.gz &&\
-    tar -zxvf go$GO_VERSION.linux-amd64.tar.gz && rm go$GO_VERSION.linux-amd64.tar.gz
-ENV PATH="/usr/local/go/bin:${PATH}" 
-
diff --git a/snap/local/Makefile b/snap/local/Makefile
deleted file mode 100644
index fbc2eff2e..000000000
--- a/snap/local/Makefile
+++ /dev/null
@@ -1,31 +0,0 @@
-GO_VERSION ?= 1.22.1
-SNAP_BUILDER_IMAGE := local/snap-builder:2.0.0-$(shell echo $(GO_VERSION) | tr :/ __)
-
-
-# truncate pkg_tag, snap has limitations for name length
-SNAP_TAG := $(shell echo "$(PKG_TAG)" | cut -b1-31 )
-
-snap-builder-image:
-	(docker image ls --format '{{.Repository}}:{{.Tag}}' | grep -q '$(SNAP_BUILDER_IMAGE)$$') \
-	|| docker build \
-		--build-arg GO_VERSION=$(GO_VERSION) \
-		--tag $(SNAP_BUILDER_IMAGE) \
-		snap/local
-
-build-snap: snap-builder-image
-	docker run --rm \
-		--mount type=bind,src="$(shell pwd)",dst=/builder \
-		-w /builder \
-		--env PKG_TAG=$(SNAP_TAG) \
-		$(SNAP_BUILDER_IMAGE) \
-		bash -c 'snapcraft clean && snapcraft snap -o "bin/victoriametrics_$(SNAP_TAG)_multi.snap"'
-
-release-snap: snap-builder-image build-snap
-	docker run --rm \
-		--mount type=bind,src=${HOME}/.snap,dst=/root/.snap \
-		--mount type=bind,src="$(shell pwd)",dst=/builder \
-		-w /builder \
-		--env PKG_TAG=$(SNAP_TAG) \
-		$(SNAP_BUILDER_IMAGE) \
-		bash -c 'snapcraft login --with /root/.snap/login.json && snapcraft upload "bin/victoriametrics_$(SNAP_TAG)_multi.snap" --release beta,edge,candidate,stable'
-
diff --git a/snap/local/README.md b/snap/local/README.md
deleted file mode 100644
index 4862bf381..000000000
--- a/snap/local/README.md
+++ /dev/null
@@ -1,50 +0,0 @@
-### Snap integration
-
-<https://snapcraft.io/>
-
-snap link: <https://snapcraft.io/victoriametrics>
-
-#### develop
-
-Install snapcraft or docker
-
-build snap package with command
-
- ```console
-make build-snap
-```
-
-It produces snap package with current git version - `victoriametrics_v1.46.0+git1.1bebd021a-dirty_all.snap`.
-You can install it with command: `snap install victoriametrics_v1.46.0+git1.1bebd021a-dirty_all.snap --dangerous`
-
-#### usage
-
-installation and configuration:
-
-```console
-# install
-snap install victoriametrics
-# logs
-snap logs victoriametrics
-# restart
- snap restart victoriametrics
-```
-
-Configuration management:
-
- Prometheus scrape config can be edited with your favorite editor, its located at
-
-```console
-vi /var/snap/victoriametrics/current/etc/victoriametrics-scrape-config.yaml
-```
-
-after changes, you can trigger config reread with `curl localhost:8428/-/reload`.
-
-Configuration tuning is possible with editing extra_flags:
-
-```console
-echo 'FLAGS="-selfScrapeInterval=10s -search.logSlowQueryDuration=20s"' > /var/snap/victoriametrics/current/extra_flags
-snap restart victoriametrics
-```
-
-Data folder located at `/var/snap/victoriametrics/current/var/lib/victoriametrics/`
diff --git a/snap/local/victoriametrics-wrapper b/snap/local/victoriametrics-wrapper
deleted file mode 100755
index afb3a4994..000000000
--- a/snap/local/victoriametrics-wrapper
+++ /dev/null
@@ -1,7 +0,0 @@
-#!/bin/bash
-
-test -e $SNAP_DATA/extra_flags || echo 'FLAGS=""' > $SNAP_DATA/extra_flags
-
-# get args
-. $SNAP_DATA/extra_flags
-$SNAP/bin/victoria-metrics -storageDataPath /var/lib/victoriametrics --promscrape.config /etc/victoriametrics-scrape-config.yaml $FLAGS
\ No newline at end of file
diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml
deleted file mode 100644
index 6eb60e4a0..000000000
--- a/snap/snapcraft.yaml
+++ /dev/null
@@ -1,86 +0,0 @@
-name: victoriametrics
-base: core18
-adopt-info: build
-icon: logo.png
-summary: VictoriaMetrics is fast, cost-effective and scalable time-series database.
-description: |
-  * VictoriaMetrics can be used as long-term storage for Prometheus or for vmagent.
-  See https://docs.victoriametrics.com/single-server-victoriametrics/#prometheus-setup for details.
-  * Supports Prometheus querying API, so it can be used as Prometheus drop-in replacement in Grafana.
-  VictoriaMetrics implements MetricsQL, https://docs.victoriametrics.com/metricsql/ query language, which is inspired by PromQL.
-  * Supports global query view. Multiple Prometheus instances may write data into VictoriaMetrics. Later this data may be used in a single query.
-  * High performance and good scalability for both inserts  and selects.
-  * High data compression, so up to 70x more data points  may be crammed into limited storage comparing to TimescaleDB.
-  * Optimized for storage with high-latency IO and low IOPS (HDD and network storage in AWS, Google Cloud, Microsoft Azure, etc).
-  * A single-node VictoriaMetrics may substitute moderately sized clusters built with competing solutions such as Thanos, M3DB, Cortex, InfluxDB or TimescaleDB.
-  * Easy operation:
-      * VictoriaMetrics consists of a single without external dependencies.
-      * All the configuration is done via explicit command-line flags with reasonable defaults.
-      * All the data is stored in a single directory pointed by `-storageDataPath` flag.
-      * Easy and fast backups from
-      to S3 or GCS with https://docs.victoriametrics.com/vmbackup/ / https://docs.victoriametrics.com/vmrestore/.
-  * Storage is protected from corruption on unclean shutdown (i.e. OOM, hardware reset or `kill -9`) thanks to the storage architecture.
-  * Supports metrics' scraping, ingestion and backfilling via the following protocols:
-    * [Metrics from Prometheus exporters]
-      such as node_exporter.
-      *  Prometheus remote write API(
-      * InfluxDB line protocol over HTTP, TCP and UDP.
-      * Graphite plaintext protocol
-      if `-graphiteListenAddr` is set.
-      * OpenTSDB put message, if `-opentsdbListenAddr` is set.
-      * HTTP OpenTSDB /api/put requests, if `-opentsdbHTTPListenAddr` is set.
-      * JSON line format.
-      * Native binary format.
-      * Prometheus exposition format.
-      * Arbitrary CSV data.
-  * Supports metrics' relabeling. See https://docs.victoriametrics.com/#relabeling for details.
-  * Ideally works with big amounts of time series data from Kubernetes, IoT sensors, connected cars, industrial telemetry, financial data and various Enterprise workloads.
-  * Has open source cluster version (https://github.com/VictoriaMetrics/VictoriaMetrics/tree/cluster).
-
-  ** Configuration management**
-  * Prometheus scrape config can be edited with your favorite editor, its located at
-      `vi /var/snap/victoriametrics/current/etc/victoriametrics-scrape-config.yaml`
-      after changes, you can trigger config reread with `curl localhost:8428/-/reload`.
-  * Configuration tuning is possible with editing extra_flags:
-      `echo 'FLAGS="-selfScrapeInterval=10s -search.logSlowQueryDuration=20s"' > /var/snap/victoriametrics/current/extra_flags && snap restart victoriametrics`
-  * Data folder located at `/var/snap/victoriametrics/current/var/lib/victoriametrics/`.
-
-grade: stable # must be 'stable' to release into candidate/stable channels
-confinement: strict # use 'strict' once you have the right plugs and slots
-
-parts:
-  build:
-    plugin: go
-    go-channel: 1.22/stable
-    go-importpath: github.com/VictoriaMetrics/VictoriaMetrics
-    source: .
-    source-type: local
-    build-packages: [build-essential, make]
-    override-build: |
-      make app-local APP_NAME=victoria-metrics
-      mkdir -p $SNAPCRAFT_PART_INSTALL/bin
-      cp -p $SNAPCRAFT_PART_BUILD/bin/victoria-metrics $SNAPCRAFT_PART_INSTALL/bin/
-      cp -p $SNAPCRAFT_PART_SRC/snap/local/victoriametrics-wrapper $SNAPCRAFT_PART_INSTALL/bin/
-    override-pull: |
-      snapcraftctl set-version "$PKG_TAG"
-      snapcraftctl pull
-apps:
-  victoriametrics:
-    command: bin/victoriametrics-wrapper
-    daemon: simple
-    restart-condition: always
-    restart-delay: 10s
-    plugs: [network,network-bind]
-
-plugs:
-  network:
-  network-bind:
-
-layout:
-  /var/lib/victoriametrics:
-    bind: $SNAP_DATA/var/lib/victoriametrics
-  /etc/victoriametrics-scrape-config.yaml:
-    bind-file: $SNAP_DATA/etc/victoriametrics-scrape-config.yaml
-architectures:
-  - build-on: ['amd64']
-    run-on: ['amd64']


Test patch:
diff --git a/app/vlinsert/syslog/syslog_test.go b/app/vlinsert/syslog/syslog_test.go
index e984d197a..7eb329b85 100644
--- a/app/vlinsert/syslog/syslog_test.go
+++ b/app/vlinsert/syslog/syslog_test.go
@@ -30,10 +30,19 @@ func TestSyslogLineReader_Success(t *testing.T) {
 	}
 
 	f("", nil)
+	f("\n", nil)
+	f("\n\n\n", nil)
+
 	f("foobar", []string{"foobar"})
+	f("foobar\n", []string{"foobar\n"})
+	f("\n\nfoo\n\nbar\n\n", []string{"foo\n\nbar\n\n"})
+
+	f(`Jun  3 12:08:33 abcd systemd: Starting Update the local ESM caches...`, []string{"Jun  3 12:08:33 abcd systemd: Starting Update the local ESM caches..."})
 
 	f(`Jun  3 12:08:33 abcd systemd: Starting Update the local ESM caches...
+
 48 <165>Jun  4 12:08:33 abcd systemd[345]: abc defg<123>1 2023-06-03T17:42:12.345Z mymachine.example.com appname 12345 ID47 [exampleSDID@32473 iut="3" eventSource="Application 123 = ] 56" eventID="11211"] This is a test message with structured data.
+
 `, []string{
 		"Jun  3 12:08:33 abcd systemd: Starting Update the local ESM caches...",
 		"<165>Jun  4 12:08:33 abcd systemd[345]: abc defg",
diff --git a/app/vmagent/remotewrite/remotewrite_test.go b/app/vmagent/remotewrite/remotewrite_test.go
index 6c384ba2e..7fd2179a1 100644
--- a/app/vmagent/remotewrite/remotewrite_test.go
+++ b/app/vmagent/remotewrite/remotewrite_test.go
@@ -79,12 +79,12 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 			rowsDroppedByRelabel:   metrics.GetOrCreateCounter(`bar`),
 		}
 		if dedupInterval > 0 {
-			rwctx.deduplicator = streamaggr.NewDeduplicator(nil, dedupInterval, nil)
+			rwctx.deduplicator = streamaggr.NewDeduplicator(nil, dedupInterval, nil, "global")
 		}
 
 		if len(streamAggrConfig) > 0 {
 			f := createFile(t, []byte(streamAggrConfig))
-			sas, err := streamaggr.LoadFromFile(f.Name(), nil, nil)
+			sas, err := streamaggr.LoadFromFile(f.Name(), nil, streamaggr.Options{})
 			if err != nil {
 				t.Fatalf("cannot load streamaggr configs: %s", err)
 			}
diff --git a/lib/logstorage/filter_test.go b/lib/logstorage/filter_test.go
index dc3358b64..67cc5764a 100644
--- a/lib/logstorage/filter_test.go
+++ b/lib/logstorage/filter_test.go
@@ -196,11 +196,7 @@ func testFilterMatchForColumns(t *testing.T, columns []column, f filter, neededC
 func testFilterMatchForStorage(t *testing.T, s *Storage, tenantID TenantID, f filter, neededColumnName string, expectedValues []string, expectedTimestamps []int64) {
 	t.Helper()
 
-	so := &genericSearchOptions{
-		tenantIDs:         []TenantID{tenantID},
-		filter:            f,
-		neededColumnNames: []string{neededColumnName, "_time"},
-	}
+	so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{neededColumnName, "_time"})
 
 	type result struct {
 		value     string
diff --git a/lib/logstorage/parser_test.go b/lib/logstorage/parser_test.go
index 78558e329..1d3abf15b 100644
--- a/lib/logstorage/parser_test.go
+++ b/lib/logstorage/parser_test.go
@@ -2070,6 +2070,7 @@ func TestQueryCanLiveTail(t *testing.T) {
 	f("* | replace_regexp ('foo', 'bar')", true)
 	f("* | sort by (a)", false)
 	f("* | stats count() rows", false)
+	f("* | stream_context after 10", false)
 	f("* | top 10 by (x)", false)
 	f("* | uniq by (a)", false)
 	f("* | unpack_json", true)
diff --git a/lib/logstorage/pipe_sort_test.go b/lib/logstorage/pipe_sort_test.go
index f14a0d2aa..671f4b0e7 100644
--- a/lib/logstorage/pipe_sort_test.go
+++ b/lib/logstorage/pipe_sort_test.go
@@ -11,9 +11,11 @@ func TestParsePipeSortSuccess(t *testing.T) {
 	}
 
 	f(`sort`)
+	f(`sort rank as foo`)
 	f(`sort by (x)`)
 	f(`sort by (x) limit 10`)
 	f(`sort by (x) offset 20 limit 10`)
+	f(`sort by (x) offset 20 limit 10 rank as bar`)
 	f(`sort by (x desc, y) desc`)
 }
 
@@ -24,6 +26,7 @@ func TestParsePipeSortFailure(t *testing.T) {
 	}
 
 	f(`sort a`)
+	f(`sort rank`)
 	f(`sort by`)
 	f(`sort by(x) foo`)
 	f(`sort by(x) limit`)
@@ -59,6 +62,29 @@ func TestPipeSort(t *testing.T) {
 		},
 	})
 
+	// Sort by all fields with rank
+	f("sort rank x", [][]Field{
+		{
+			{"_msg", `def`},
+			{"a", `1`},
+		},
+		{
+			{"_msg", `abc`},
+			{"a", `2`},
+		},
+	}, [][]Field{
+		{
+			{"_msg", `abc`},
+			{"a", `2`},
+			{"x", "1"},
+		},
+		{
+			{"_msg", `def`},
+			{"a", `1`},
+			{"x", "2"},
+		},
+	})
+
 	// Sort by a single field
 	f("sort by (a asc) asc", [][]Field{
 		{
@@ -205,6 +231,35 @@ func TestPipeSort(t *testing.T) {
 		},
 	})
 
+	// Sort by multiple fields with offset and rank
+	f("sort by (a, b) offset 1 rank x", [][]Field{
+		{
+			{"_msg", `abc`},
+			{"a", `2`},
+			{"b", `3`},
+		},
+		{
+			{"_msg", `def`},
+			{"a", `1`},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+		},
+	}, [][]Field{
+		{
+			{"_msg", `abc`},
+			{"a", `2`},
+			{"b", `3`},
+			{"x", "2"},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+			{"x", "3"},
+		},
+	})
+
 	// Sort by multiple fields with offset and limit
 	f("sort by (a, b) offset 1 limit 1", [][]Field{
 		{
@@ -228,6 +283,30 @@ func TestPipeSort(t *testing.T) {
 		},
 	})
 
+	// Sort by multiple fields with offset, limit and rank
+	f("sort by (a, b) offset 1 limit 1 rank x", [][]Field{
+		{
+			{"_msg", `abc`},
+			{"a", `2`},
+			{"b", `3`},
+		},
+		{
+			{"_msg", `def`},
+			{"a", `1`},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+		},
+	}, [][]Field{
+		{
+			{"_msg", `abc`},
+			{"a", `2`},
+			{"b", `3`},
+			{"x", "2"},
+		},
+	})
+
 	// Sort by multiple fields with offset and limit
 	f("sort by (a, b) desc offset 2 limit 100", [][]Field{
 		{
@@ -259,17 +338,27 @@ func TestPipeSortUpdateNeededFields(t *testing.T) {
 	}
 
 	// all the needed fields
+	f("sort", "*", "", "*", "")
+	f("sort rank x", "*", "", "*", "")
 	f("sort by(s1,s2)", "*", "", "*", "")
+	f("sort by(s1,s2) rank as x", "*", "", "*", "x")
+	f("sort by(x,s2) rank as x", "*", "", "*", "")
 
 	// all the needed fields, unneeded fields do not intersect with src
 	f("sort by(s1,s2)", "*", "f1,f2", "*", "f1,f2")
+	f("sort by(s1,s2) rank as x", "*", "f1,f2", "*", "f1,f2,x")
+	f("sort by(x,s2) rank as x", "*", "f1,f2", "*", "f1,f2")
 
 	// all the needed fields, unneeded fields intersect with src
 	f("sort by(s1,s2)", "*", "s1,f1,f2", "*", "f1,f2")
+	f("sort by(s1,s2) rank as x", "*", "s1,f1,f2", "*", "f1,f2,x")
+	f("sort by(x,s2) rank as x", "*", "s1,f1,f2", "*", "f1,f2,s1")
 
 	// needed fields do not intersect with src
 	f("sort by(s1,s2)", "f1,f2", "", "s1,s2,f1,f2", "")
+	f("sort by(s1,s2) rank as x", "f1,f2", "", "s1,s2,f1,f2", "")
 
 	// needed fields intersect with src
 	f("sort by(s1,s2)", "s1,f1,f2", "", "s1,s2,f1,f2", "")
+	f("sort by(s1,s2) rank as x", "s1,f1,f2,x", "", "s1,s2,f1,f2", "")
 }
diff --git a/lib/logstorage/pipe_stream_context_test.go b/lib/logstorage/pipe_stream_context_test.go
new file mode 100644
index 000000000..12943f85f
--- /dev/null
+++ b/lib/logstorage/pipe_stream_context_test.go
@@ -0,0 +1,116 @@
+package logstorage
+
+import (
+	"testing"
+)
+
+func TestParsePipeStreamContextSuccess(t *testing.T) {
+	f := func(pipeStr string) {
+		t.Helper()
+		expectParsePipeSuccess(t, pipeStr)
+	}
+
+	f(`stream_context before 5`)
+	f(`stream_context after 10`)
+	f(`stream_context before 10 after 20`)
+}
+
+func TestParsePipeStreamContextFailure(t *testing.T) {
+	f := func(pipeStr string) {
+		t.Helper()
+		expectParsePipeFailure(t, pipeStr)
+	}
+
+	f(`stream_context`)
+	f(`stream_context before`)
+	f(`stream_context after`)
+	f(`stream_context before after`)
+	f(`stream_context after before`)
+	f(`stream_context before -4`)
+	f(`stream_context after -4`)
+}
+
+func TestPipeStreamContext(t *testing.T) {
+	f := func(pipeStr string, rows, rowsExpected [][]Field) {
+		t.Helper()
+		expectPipeResults(t, pipeStr, rows, rowsExpected)
+	}
+
+	f("stream_context before 0", [][]Field{
+		{
+			{"a", `2`},
+			{"b", `3`},
+		},
+		{
+			{"a", "2"},
+			{"b", "3"},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+			{"c", "d"},
+		},
+	}, [][]Field{
+		{
+			{"a", `2`},
+			{"b", `3`},
+		},
+		{
+			{"a", "2"},
+			{"b", "3"},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+			{"c", "d"},
+		},
+	})
+
+	f("stream_context after 0", [][]Field{
+		{
+			{"a", `2`},
+			{"b", `3`},
+		},
+		{
+			{"a", "2"},
+			{"b", "3"},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+			{"c", "d"},
+		},
+	}, [][]Field{
+		{
+			{"a", `2`},
+			{"b", `3`},
+		},
+		{
+			{"a", "2"},
+			{"b", "3"},
+		},
+		{
+			{"a", `2`},
+			{"b", `54`},
+			{"c", "d"},
+		},
+	})
+}
+
+func TestPipeStreamContextUpdateNeededFields(t *testing.T) {
+	f := func(s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected string) {
+		t.Helper()
+		expectPipeNeededFields(t, s, neededFields, unneededFields, neededFieldsExpected, unneededFieldsExpected)
+	}
+
+	// all the needed fields
+	f("stream_context before 10", "*", "", "*", "")
+
+	// plus unneeded fields
+	f("stream_context after 4", "*", "f1,f2", "*", "f1,f2")
+	f("stream_context after 4", "*", "_time,f1,_stream_id", "*", "f1")
+
+	// needed fields
+	f("stream_context before 3", "f1,f2", "", "_stream_id,_time,f1,f2", "")
+	f("stream_context before 3", "_time,f1,_stream_id", "", "_stream_id,_time,f1", "")
+}
diff --git a/lib/logstorage/storage_search_test.go b/lib/logstorage/storage_search_test.go
index 49fc59956..be39d2d05 100644
--- a/lib/logstorage/storage_search_test.go
+++ b/lib/logstorage/storage_search_test.go
@@ -3,6 +3,7 @@ package logstorage
 import (
 	"context"
 	"fmt"
+	"math"
 	"reflect"
 	"sort"
 	"strings"
@@ -87,7 +88,7 @@ func TestStorageRunQuery(t *testing.T) {
 		t.Helper()
 		err := s.RunQuery(context.Background(), tenantIDs, q, writeBlock)
 		if err != nil {
-			t.Fatalf("unexpected error returned from the query %s: %s", q, err)
+			t.Fatalf("unexpected error returned from the query [%s]: %s", q, err)
 		}
 	}
 
@@ -656,6 +657,78 @@ func TestStorageRunQuery(t *testing.T) {
 			},
 		})
 	})
+	t.Run("stream_context-noop-1", func(t *testing.T) {
+		f(t, `"message 3 at block 1"
+			| stream_context before 0
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "33"},
+			},
+		})
+	})
+	t.Run("stream_context-noop-2", func(t *testing.T) {
+		f(t, `"message 3 at block 1"
+			| stream_context before 0 after 0
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "33"},
+			},
+		})
+	})
+	t.Run("stream_context-before-1", func(t *testing.T) {
+		f(t, `"message 3 at block 1"
+			| stream_context before 1
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "66"},
+			},
+		})
+	})
+	t.Run("stream_context-after-1", func(t *testing.T) {
+		f(t, `"message 3 at block 1"
+			| stream_context after 1
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "66"},
+			},
+		})
+	})
+	t.Run("stream_context-before-after-1", func(t *testing.T) {
+		f(t, `"message 3 at block 1"
+			| stream_context before 1 after 1
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "99"},
+			},
+		})
+	})
+	t.Run("stream_context-before-1000", func(t *testing.T) {
+		f(t, `"message 4"
+			| stream_context before 1000
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "825"},
+			},
+		})
+	})
+	t.Run("stream_context-after-1000", func(t *testing.T) {
+		f(t, `"message 4"
+			| stream_context after 1000
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "495"},
+			},
+		})
+	})
+	t.Run("stream_context-before-after-1000", func(t *testing.T) {
+		f(t, `"message 4"
+			| stream_context before 1000 after 1000
+			| stats count() rows`, [][]Field{
+			{
+				{"rows", "1155"},
+			},
+		})
+	})
 
 	// Close the storage and delete its data
 	s.MustClose()
@@ -757,11 +830,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp
 		maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 		f := getBaseFilter(minTimestamp, maxTimestamp, nil)
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		processBlock := func(_ uint, _ *blockResult) {
 			panic(fmt.Errorf("unexpected match"))
 		}
@@ -775,11 +844,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp
 		maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 		f := getBaseFilter(minTimestamp, maxTimestamp, nil)
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		processBlock := func(_ uint, _ *blockResult) {
 			panic(fmt.Errorf("unexpected match"))
 		}
@@ -793,11 +858,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp
 		maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 		f := getBaseFilter(minTimestamp, maxTimestamp, nil)
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		processBlock := func(_ uint, _ *blockResult) {
 			panic(fmt.Errorf("unexpected match"))
 		}
@@ -812,11 +873,7 @@ func TestStorageSearch(t *testing.T) {
 			minTimestamp := baseTimestamp
 			maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 			f := getBaseFilter(minTimestamp, maxTimestamp, nil)
-			so := &genericSearchOptions{
-				tenantIDs:         []TenantID{tenantID},
-				filter:            f,
-				neededColumnNames: []string{"_msg"},
-			}
+			so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 			var rowsCountTotal atomic.Uint32
 			processBlock := func(_ uint, br *blockResult) {
 				rowsCountTotal.Add(uint32(len(br.timestamps)))
@@ -833,11 +890,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp
 		maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 		f := getBaseFilter(minTimestamp, maxTimestamp, nil)
-		so := &genericSearchOptions{
-			tenantIDs:         allTenantIDs,
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions(allTenantIDs, f, []string{"_msg"})
 		var rowsCountTotal atomic.Uint32
 		processBlock := func(_ uint, br *blockResult) {
 			rowsCountTotal.Add(uint32(len(br.timestamps)))
@@ -854,11 +907,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp
 		maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 		f := getBaseFilter(minTimestamp, maxTimestamp, sf)
-		so := &genericSearchOptions{
-			tenantIDs:         allTenantIDs,
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions(allTenantIDs, f, []string{"_msg"})
 		processBlock := func(_ uint, _ *blockResult) {
 			panic(fmt.Errorf("unexpected match"))
 		}
@@ -874,11 +923,7 @@ func TestStorageSearch(t *testing.T) {
 			minTimestamp := baseTimestamp
 			maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 			f := getBaseFilter(minTimestamp, maxTimestamp, sf)
-			so := &genericSearchOptions{
-				tenantIDs:         []TenantID{tenantID},
-				filter:            f,
-				neededColumnNames: []string{"_msg"},
-			}
+			so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 			var rowsCountTotal atomic.Uint32
 			processBlock := func(_ uint, br *blockResult) {
 				rowsCountTotal.Add(uint32(len(br.timestamps)))
@@ -900,11 +945,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp
 		maxTimestamp := baseTimestamp + rowsPerBlock*1e9 + blocksPerStream
 		f := getBaseFilter(minTimestamp, maxTimestamp, sf)
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		var rowsCountTotal atomic.Uint32
 		processBlock := func(_ uint, br *blockResult) {
 			rowsCountTotal.Add(uint32(len(br.timestamps)))
@@ -934,11 +975,7 @@ func TestStorageSearch(t *testing.T) {
 				},
 			},
 		}
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		var rowsCountTotal atomic.Uint32
 		processBlock := func(_ uint, br *blockResult) {
 			rowsCountTotal.Add(uint32(len(br.timestamps)))
@@ -959,11 +996,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp + (rowsPerBlock-2)*1e9
 		maxTimestamp := baseTimestamp + (rowsPerBlock-1)*1e9 - 1
 		f := getBaseFilter(minTimestamp, maxTimestamp, sf)
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		var rowsCountTotal atomic.Uint32
 		processBlock := func(_ uint, br *blockResult) {
 			rowsCountTotal.Add(uint32(len(br.timestamps)))
@@ -984,11 +1017,7 @@ func TestStorageSearch(t *testing.T) {
 		minTimestamp := baseTimestamp + (rowsPerBlock+1)*1e9
 		maxTimestamp := baseTimestamp + (rowsPerBlock+2)*1e9
 		f := getBaseFilter(minTimestamp, maxTimestamp, sf)
-		so := &genericSearchOptions{
-			tenantIDs:         []TenantID{tenantID},
-			filter:            f,
-			neededColumnNames: []string{"_msg"},
-		}
+		so := newTestGenericSearchOptions([]TenantID{tenantID}, f, []string{"_msg"})
 		processBlock := func(_ uint, _ *blockResult) {
 			panic(fmt.Errorf("unexpected match"))
 		}
@@ -1020,3 +1049,13 @@ func TestParseStreamFieldsSuccess(t *testing.T) {
 	f(`{a="b",c="d"}`, `{"a":"b","c":"d"}`)
 	f(`{a="a=,b\"c}",b="d"}`, `{"a":"a=,b\"c}","b":"d"}`)
 }
+
+func newTestGenericSearchOptions(tenantIDs []TenantID, f filter, neededColumns []string) *genericSearchOptions {
+	return &genericSearchOptions{
+		tenantIDs:         tenantIDs,
+		minTimestamp:      math.MinInt64,
+		maxTimestamp:      math.MaxInt64,
+		filter:            f,
+		neededColumnNames: neededColumns,
+	}
+}
diff --git a/lib/streamaggr/deduplicator_test.go b/lib/streamaggr/deduplicator_test.go
index 96be55ef8..fe32cbb7a 100644
--- a/lib/streamaggr/deduplicator_test.go
+++ b/lib/streamaggr/deduplicator_test.go
@@ -29,7 +29,7 @@ foo{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="as
 baz_aaa_aaa_fdd{instance="x",job="aaa",pod="sdfd-dfdfdfs",node="aosijjewrerfd",namespace="asdff",container="ohohffd"} -2.3
 `)
 
-	d := NewDeduplicator(pushFunc, time.Hour, []string{"node", "instance"})
+	d := NewDeduplicator(pushFunc, time.Hour, []string{"node", "instance"}, "global")
 	for i := 0; i < 10; i++ {
 		d.Push(tss)
 	}
diff --git a/lib/streamaggr/deduplicator_timing_test.go b/lib/streamaggr/deduplicator_timing_test.go
index e2b764041..ed98b638c 100644
--- a/lib/streamaggr/deduplicator_timing_test.go
+++ b/lib/streamaggr/deduplicator_timing_test.go
@@ -9,7 +9,7 @@ import (
 
 func BenchmarkDeduplicatorPush(b *testing.B) {
 	pushFunc := func(_ []prompbmarshal.TimeSeries) {}
-	d := NewDeduplicator(pushFunc, time.Hour, nil)
+	d := NewDeduplicator(pushFunc, time.Hour, nil, "global")
 
 	b.ReportAllocs()
 	b.SetBytes(int64(len(benchSeries)))
diff --git a/lib/streamaggr/streamaggr_test.go b/lib/streamaggr/streamaggr_test.go
index 5090ee31c..c6d17b832 100644
--- a/lib/streamaggr/streamaggr_test.go
+++ b/lib/streamaggr/streamaggr_test.go
@@ -20,7 +20,7 @@ func TestAggregatorsFailure(t *testing.T) {
 		pushFunc := func(_ []prompbmarshal.TimeSeries) {
 			panic(fmt.Errorf("pushFunc shouldn't be called"))
 		}
-		a, err := newAggregatorsFromData([]byte(config), pushFunc, nil)
+		a, err := newAggregatorsFromData([]byte(config), pushFunc, Options{})
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -158,11 +158,11 @@ func TestAggregatorsEqual(t *testing.T) {
 		t.Helper()
 
 		pushFunc := func(_ []prompbmarshal.TimeSeries) {}
-		aa, err := newAggregatorsFromData([]byte(a), pushFunc, nil)
+		aa, err := newAggregatorsFromData([]byte(a), pushFunc, Options{})
 		if err != nil {
 			t.Fatalf("cannot initialize aggregators: %s", err)
 		}
-		ab, err := newAggregatorsFromData([]byte(b), pushFunc, nil)
+		ab, err := newAggregatorsFromData([]byte(b), pushFunc, Options{})
 		if err != nil {
 			t.Fatalf("cannot initialize aggregators: %s", err)
 		}
@@ -221,7 +221,7 @@ func TestAggregatorsSuccess(t *testing.T) {
 			tssOutput = appendClonedTimeseries(tssOutput, tss)
 			tssOutputLock.Unlock()
 		}
-		opts := &Options{
+		opts := Options{
 			FlushOnShutdown:        true,
 			NoAlignFlushToInterval: true,
 		}
@@ -917,7 +917,7 @@ func TestAggregatorsWithDedupInterval(t *testing.T) {
 			}
 			tssOutputLock.Unlock()
 		}
-		opts := &Options{
+		opts := Options{
 			DedupInterval:   30 * time.Second,
 			FlushOnShutdown: true,
 		}
diff --git a/lib/streamaggr/streamaggr_timing_test.go b/lib/streamaggr/streamaggr_timing_test.go
index f06da26cc..b328a61a0 100644
--- a/lib/streamaggr/streamaggr_timing_test.go
+++ b/lib/streamaggr/streamaggr_timing_test.go
@@ -92,7 +92,7 @@ func newBenchAggregators(outputs []string, pushFunc PushFunc) *Aggregators {
   outputs: [%s]
 `, strings.Join(outputsQuoted, ","))
 
-	a, err := newAggregatorsFromData([]byte(config), pushFunc, nil)
+	a, err := newAggregatorsFromData([]byte(config), pushFunc, Options{})
 	if err != nil {
 		panic(fmt.Errorf("unexpected error when initializing aggregators: %s", err))
 	}