Repository: VictoriaMetrics/VictoriaMetrics
Pull request #858 (2020-10-24 18:30:22)

dashboard: add `Storage full ETA` panel
The new panel suppose to help to estimate the time needed to run out of free
disk space.
Thx to @belm0 @hekmon

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=h1) Report
> Merging [#858](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=desc) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/bca468bb55f0015c6a5d32cafcb1074f833448fe?el=desc) will **increase** coverage by `0.05%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #858      +/-   ##
==========================================
+ Coverage   62.88%   62.93%   +0.05%     
==========================================
  Files         189      189              
  Lines       21532    21532              
==========================================
+ Hits        13540    13551      +11     
+ Misses       6710     6699      -11     
  Partials     1282     1282              
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `68.38% <0.00%> (-0.52%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `65.99% <0.00%> (+0.31%)` | :arrow_up: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `56.60% <0.00%> (+1.88%)` | :arrow_up: |
| [lib/storage/merge.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvbWVyZ2UuZ28=) | `79.48% <0.00%> (+2.56%)` | :arrow_up: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `86.43% <0.00%> (+3.10%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=footer). Last update [bca468b...d5b1e5e](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/858?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

hagen1778: > would you attach a screenshot?

Sure thing!
![image](https://user-images.githubusercontent.com/2902918/97109877-bb374380-16cd-11eb-8ecc-f1901f3742a4.png)


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index e3671e5dc..4306f7488 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,11 +9,24 @@
   * `num_or_scalar op foo{filters} op bar` -> `num_or_scalar op foo{filters} op bar{filters}`
 * FEATURE: improve time series search for queries with multiple label filters. I.e. `foo{label1="value", label2=~"regexp"}`.
   See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/781
+* FEATURE: vmagent: add `stream parse` mode. This mode allows reducing memory usage when individual scrape targets expose tens of millions of metrics.
+  For example, during scraping Prometheus in [federation](https://prometheus.io/docs/prometheus/latest/federation/) mode.
+  See `-promscrape.streamParse` command-line option and `stream_parse: true` config option for `scrape_config` section in `-promscrape.config`.
+  See also [troubleshooting docs for vmagent](https://victoriametrics.github.io/vmagent.html#troubleshooting).
 * FEATURE: vmalert: add `-dryRun` command-line option for validating the provided config files without the need to start `vmalert` service.
+* FEATURE: accept optional third argument of string type at `topk_*` and `bottomk_*` functions. This is label name for additional time series to return with the sum of time series outside top/bottom K. See [MetricsQL docs](https://victoriametrics.github.io/MetricsQL.html) for more details.
+* FEATURE: vmagent: expose `/api/v1/targets` page according to [the corresponding Prometheus API](https://prometheus.io/docs/prometheus/latest/querying/api/#targets).
+  See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/643
 
 * BUGFIX: vmagent: properly handle OpenStack endpoint ending with `v3.0` such as `https://ostack.example.com:5000/v3.0`
   in the same way as Prometheus does. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/728#issuecomment-709914803
-* BUGFIX: Drop trailing data points for time series with a single raw sample. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/748
+* BUGFIX: drop trailing data points for time series with a single raw sample. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/748
+* BUGFIX: vmbackup: fix panic when `-origin` isn't specified. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/856
+* BUGFIX: vmalert: skip automatically added labels on alerts restore. Label `alertgroup` was introduced in [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/611)
+  and automatically added to generated time series. By mistake, this new label wasn't correctly purged on restore event and affected alert's ID uniqueness.
+  See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/870
+* BUGFIX: vmagent: fix panic at scrape error body formating. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/864
+* BUGFIX: vmagent: add leading missing slash to metrics path like Prometheus does. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/835
 
 
 # [v1.44.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.44.0)
diff --git a/README.md b/README.md
index 16b85aa26..40b212b6d 100644
--- a/README.md
+++ b/README.md
@@ -495,6 +495,7 @@ VictoriaMetrics supports the following handlers from [Prometheus querying API](h
 * [/api/v1/labels](https://prometheus.io/docs/prometheus/latest/querying/api/#getting-label-names)
 * [/api/v1/label/.../values](https://prometheus.io/docs/prometheus/latest/querying/api/#querying-label-values)
 * [/api/v1/status/tsdb](https://prometheus.io/docs/prometheus/latest/querying/api/#tsdb-stats)
+* [/api/v1/targets](https://prometheus.io/docs/prometheus/latest/querying/api/#targets) - see [these docs](#how-to-scrape-prometheus-exporters-such-as-node-exporter) for more details.
 
 These handlers can be queried from Prometheus-compatible clients such as Grafana or curl.
 
diff --git a/app/vmagent/README.md b/app/vmagent/README.md
index 428ea6e97..22d2c6403 100644
--- a/app/vmagent/README.md
+++ b/app/vmagent/README.md
@@ -211,9 +211,13 @@ either via `vmagent` itself or via Prometheus, so the exported metrics could be
 Use official [Grafana dashboard](https://grafana.com/grafana/dashboards/12683) for `vmagent` state overview.
 If you have suggestions, improvements or found a bug - feel free to open an issue on github or add review to the dashboard.
 
-`vmagent` also exports target statuses at `http://vmagent-host:8429/targets` page in plaintext format.
-`/targets` handler accepts optional `show_original_labels=1` query arg, which shows the original labels per each target
-before applying relabeling. This information may be useful for debugging target relabeling.
+`vmagent` also exports target statuses at the following handlers:
+
+* `http://vmagent-host:8429/targets`. This handler returns human-readable plaintext status for every active target.
+This page is convenient to query from command line with `wget`, `curl` or similar tools.
+It accepts optional `show_original_labels=1` query arg, which shows the original labels per each target before applying relabeling.
+This information may be useful for debugging target relabeling.
+* `http://vmagent-host:8429/api/v1/targets`. This handler returns data compatible with [the corresponding page from Prometheus API](https://prometheus.io/docs/prometheus/latest/querying/api/#targets).
 
 
 ### Troubleshooting
@@ -224,7 +228,26 @@ before applying relabeling. This information may be useful for debugging target
   since `vmagent` establishes at least a single TCP connection per each target.
 
 * When `vmagent` scrapes many unreliable targets, it can flood error log with scrape errors. These errors can be suppressed
-  by passing `-promscrape.suppressScrapeErrors` command-line flag to `vmagent`. The most recent scrape error per each target can be observed at `http://vmagent-host:8429/targets`.
+  by passing `-promscrape.suppressScrapeErrors` command-line flag to `vmagent`. The most recent scrape error per each target can be observed at `http://vmagent-host:8429/targets`
+  and `http://vmagent-host:8429/api/v1/targets`.
+
+* If `vmagent` scrapes targets with millions of metrics per each target (for instance, when scraping [federation endpoints](https://prometheus.io/docs/prometheus/latest/federation/)),
+  then it is recommended enabling `stream parsing mode` in order to reduce memory usage during scraping. This mode may be enabled either globally for all the scrape targets
+  by passing `-promscrape.streamParse` command-line flag or on a per-scrape target basis with `stream_parse: true` option. For example:
+
+  ```yml
+  scrape_configs:
+  - job_name: 'big-federate'
+    stream_parse: true
+    static_configs:
+    - targets:
+      - big-prometeus1
+      - big-prometeus2
+    honor_labels: true
+    metrics_path: /federate
+    params:
+      'match[]': ['{__name__!=""}']
+  ```
 
 * It is recommended to increase `-remoteWrite.queues` if `vmagent_remotewrite_pending_data_bytes` metric exported at `http://vmagent-host:8429/metrics` page constantly grows.
 
diff --git a/app/vmagent/main.go b/app/vmagent/main.go
index 2a7432ca4..d3fa1d6b9 100644
--- a/app/vmagent/main.go
+++ b/app/vmagent/main.go
@@ -211,6 +211,12 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		showOriginalLabels, _ := strconv.ParseBool(r.FormValue("show_original_labels"))
 		promscrape.WriteHumanReadableTargetsStatus(w, showOriginalLabels)
 		return true
+	case "/api/v1/targets":
+		promscrapeAPIV1TargetsRequests.Inc()
+		w.Header().Set("Content-Type", "application/json")
+		state := r.FormValue("state")
+		promscrape.WriteAPIV1Targets(w, state)
+		return true
 	case "/-/reload":
 		promscrapeConfigReloadRequests.Inc()
 		procutil.SelfSIGHUP()
@@ -241,7 +247,8 @@ var (
 
 	influxQueryRequests = metrics.NewCounter(`vmagent_http_requests_total{path="/query", protocol="influx"}`)
 
-	promscrapeTargetsRequests = metrics.NewCounter(`vmagent_http_requests_total{path="/targets"}`)
+	promscrapeTargetsRequests      = metrics.NewCounter(`vmagent_http_requests_total{path="/targets"}`)
+	promscrapeAPIV1TargetsRequests = metrics.NewCounter(`vmagent_http_requests_total{path="/api/v1/targets"}`)
 
 	promscrapeConfigReloadRequests = metrics.NewCounter(`vmagent_http_requests_total{path="/-/reload"}`)
 )
diff --git a/app/vmalert/alerting.go b/app/vmalert/alerting.go
index 7e11ad850..51eee637f 100644
--- a/app/vmalert/alerting.go
+++ b/app/vmalert/alerting.go
@@ -403,7 +403,7 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
 		labelsFilter += fmt.Sprintf(",%s=%q", k, v)
 	}
 
-	// Get the last datapoint in range via MetricsQL `last_over_time`.
+	// Get the last data point in range via MetricsQL `last_over_time`.
 	// We don't use plain PromQL since Prometheus doesn't support
 	// remote write protocol which is used for state persistence in vmalert.
 	expr := fmt.Sprintf("last_over_time(%s{alertname=%q%s}[%ds])",
@@ -417,11 +417,14 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
 		labels := m.Labels
 		m.Labels = make([]datasource.Label, 0)
 		// drop all extra labels, so hash key will
-		// be identical to timeseries received in Exec
+		// be identical to time series received in Exec
 		for _, l := range labels {
 			if l.Name == alertNameLabel {
 				continue
 			}
+			if l.Name == alertGroupNameLabel {
+				continue
+			}
 			// drop all overridden labels
 			if _, ok := ar.Labels[l.Name]; ok {
 				continue
@@ -436,7 +439,7 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
 		a.ID = hash(m)
 		a.State = notifier.StatePending
 		ar.alerts[a.ID] = a
-		logger.Infof("alert %q(%d) restored to state at %v", a.Name, a.ID, a.Start)
+		logger.Infof("alert %q (%d) restored to state at %v", a.Name, a.ID, a.Start)
 	}
 	return nil
 }
diff --git a/app/vmbackup/main.go b/app/vmbackup/main.go
index 49bb72874..bbaa2a77a 100644
--- a/app/vmbackup/main.go
+++ b/app/vmbackup/main.go
@@ -10,6 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/actions"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/fslocal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/fsnil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
@@ -146,9 +147,9 @@ func newDstFS() (common.RemoteFS, error) {
 	return fs, nil
 }
 
-func newOriginFS() (common.RemoteFS, error) {
+func newOriginFS() (common.OriginFS, error) {
 	if len(*origin) == 0 {
-		return nil, nil
+		return &fsnil.FS{}, nil
 	}
 	fs, err := actions.NewRemoteFS(*origin)
 	if err != nil {
diff --git a/app/vminsert/main.go b/app/vminsert/main.go
index 616ced4b7..504922c94 100644
--- a/app/vminsert/main.go
+++ b/app/vminsert/main.go
@@ -159,6 +159,12 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		showOriginalLabels, _ := strconv.ParseBool(r.FormValue("show_original_labels"))
 		promscrape.WriteHumanReadableTargetsStatus(w, showOriginalLabels)
 		return true
+	case "/api/v1/targets":
+		promscrapeAPIV1TargetsRequests.Inc()
+		w.Header().Set("Content-Type", "application/json")
+		state := r.FormValue("state")
+		promscrape.WriteAPIV1Targets(w, state)
+		return true
 	case "/-/reload":
 		promscrapeConfigReloadRequests.Inc()
 		procutil.SelfSIGHUP()
@@ -191,7 +197,8 @@ var (
 
 	influxQueryRequests = metrics.NewCounter(`vm_http_requests_total{path="/query", protocol="influx"}`)
 
-	promscrapeTargetsRequests = metrics.NewCounter(`vm_http_requests_total{path="/targets"}`)
+	promscrapeTargetsRequests      = metrics.NewCounter(`vm_http_requests_total{path="/targets"}`)
+	promscrapeAPIV1TargetsRequests = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/targets"}`)
 
 	promscrapeConfigReloadRequests = metrics.NewCounter(`vm_http_requests_total{path="/-/reload"}`)
 
diff --git a/app/vmselect/promql/aggr.go b/app/vmselect/promql/aggr.go
index b87b34a64..050a8cbd4 100644
--- a/app/vmselect/promql/aggr.go
+++ b/app/vmselect/promql/aggr.go
@@ -69,7 +69,9 @@ func newAggrFunc(afe func(tss []*timeseries) []*timeseries) aggrFunc {
 		if err != nil {
 			return nil, err
 		}
-		return aggrFuncExt(afe, tss, &afa.ae.Modifier, afa.ae.Limit, false)
+		return aggrFuncExt(func(tss []*timeseries, modififer *metricsql.ModifierExpr) []*timeseries {
+			return afe(tss)
+		}, tss, &afa.ae.Modifier, afa.ae.Limit, false)
 	}
 }
 
@@ -98,7 +100,8 @@ func removeGroupTags(metricName *storage.MetricName, modifier *metricsql.Modifie
 	}
 }
 
-func aggrFuncExt(afe func(tss []*timeseries) []*timeseries, argOrig []*timeseries, modifier *metricsql.ModifierExpr, maxSeries int, keepOriginal bool) ([]*timeseries, error) {
+func aggrFuncExt(afe func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries, argOrig []*timeseries,
+	modifier *metricsql.ModifierExpr, maxSeries int, keepOriginal bool) ([]*timeseries, error) {
 	arg := copyTimeseriesMetricNames(argOrig, keepOriginal)
 
 	// Perform grouping.
@@ -124,7 +127,7 @@ func aggrFuncExt(afe func(tss []*timeseries) []*timeseries, argOrig []*timeserie
 	dstTssCount := 0
 	rvs := make([]*timeseries, 0, len(m))
 	for _, tss := range m {
-		rv := afe(tss)
+		rv := afe(tss, modifier)
 		rvs = append(rvs, rv...)
 		srcTssCount += len(tss)
 		dstTssCount += len(rv)
@@ -141,7 +144,7 @@ func aggrFuncAny(afa *aggrFuncArg) ([]*timeseries, error) {
 	if err != nil {
 		return nil, err
 	}
-	afe := func(tss []*timeseries) []*timeseries {
+	afe := func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
 		return tss[:1]
 	}
 	limit := afa.ae.Limit
@@ -178,10 +181,11 @@ func aggrFuncSum(tss []*timeseries) []*timeseries {
 		sum := float64(0)
 		count := 0
 		for _, ts := range tss {
-			if math.IsNaN(ts.Values[i]) {
+			v := ts.Values[i]
+			if math.IsNaN(v) {
 				continue
 			}
-			sum += ts.Values[i]
+			sum += v
 			count++
 		}
 		if count == 0 {
@@ -449,7 +453,7 @@ func aggrFuncZScore(afa *aggrFuncArg) ([]*timeseries, error) {
 	if err != nil {
 		return nil, err
 	}
-	afe := func(tss []*timeseries) []*timeseries {
+	afe := func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
 		for i := range tss[0].Values {
 			// Calculate avg and stddev for tss points at position i.
 			// See `Rapid calculation methods` at https://en.wikipedia.org/wiki/Standard_deviation
@@ -550,7 +554,7 @@ func aggrFuncCountValues(afa *aggrFuncArg) ([]*timeseries, error) {
 		// Do nothing
 	}
 
-	afe := func(tss []*timeseries) []*timeseries {
+	afe := func(tss []*timeseries, modififer *metricsql.ModifierExpr) []*timeseries {
 		m := make(map[float64]bool)
 		for _, ts := range tss {
 			for _, v := range ts.Values {
@@ -602,7 +606,7 @@ func newAggrFuncTopK(isReverse bool) aggrFunc {
 		if err != nil {
 			return nil, err
 		}
-		afe := func(tss []*timeseries) []*timeseries {
+		afe := func(tss []*timeseries, modififer *metricsql.ModifierExpr) []*timeseries {
 			for n := range tss[0].Values {
 				sort.Slice(tss, func(i, j int) bool {
 					a := tss[i].Values[n]
@@ -623,21 +627,32 @@ func newAggrFuncTopK(isReverse bool) aggrFunc {
 func newAggrFuncRangeTopK(f func(values []float64) float64, isReverse bool) aggrFunc {
 	return func(afa *aggrFuncArg) ([]*timeseries, error) {
 		args := afa.args
-		if err := expectTransformArgsNum(args, 2); err != nil {
-			return nil, err
+		if len(args) < 2 {
+			return nil, fmt.Errorf(`unexpected number of args; got %d; want at least %d`, len(args), 2)
+		}
+		if len(args) > 3 {
+			return nil, fmt.Errorf(`unexpected number of args; got %d; want no more than %d`, len(args), 3)
 		}
 		ks, err := getScalar(args[0], 0)
 		if err != nil {
 			return nil, err
 		}
-		afe := func(tss []*timeseries) []*timeseries {
-			return getRangeTopKTimeseries(tss, ks, f, isReverse)
+		remainingSumTagName := ""
+		if len(args) == 3 {
+			remainingSumTagName, err = getString(args[2], 2)
+			if err != nil {
+				return nil, err
+			}
+		}
+		afe := func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
+			return getRangeTopKTimeseries(tss, modifier, ks, remainingSumTagName, f, isReverse)
 		}
 		return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
 	}
 }
 
-func getRangeTopKTimeseries(tss []*timeseries, ks []float64, f func(values []float64) float64, isReverse bool) []*timeseries {
+func getRangeTopKTimeseries(tss []*timeseries, modifier *metricsql.ModifierExpr, ks []float64, remainingSumTagName string,
+	f func(values []float64) float64, isReverse bool) []*timeseries {
 	type tsWithValue struct {
 		ts    *timeseries
 		value float64
@@ -661,26 +676,64 @@ func getRangeTopKTimeseries(tss []*timeseries, ks []float64, f func(values []flo
 	for i := range maxs {
 		tss[i] = maxs[i].ts
 	}
+	remainingSumTS := getRemainingSumTimeseries(tss, modifier, ks, remainingSumTagName)
 	for i, k := range ks {
 		fillNaNsAtIdx(i, k, tss)
 	}
+	if remainingSumTS != nil {
+		tss = append(tss, remainingSumTS)
+	}
 	return removeNaNs(tss)
 }
 
+func getRemainingSumTimeseries(tss []*timeseries, modifier *metricsql.ModifierExpr, ks []float64, remainingSumTagName string) *timeseries {
+	if len(remainingSumTagName) == 0 || len(tss) == 0 {
+		return nil
+	}
+	var dst timeseries
+	dst.CopyFromShallowTimestamps(tss[0])
+	removeGroupTags(&dst.MetricName, modifier)
+	dst.MetricName.RemoveTag(remainingSumTagName)
+	dst.MetricName.AddTag(remainingSumTagName, remainingSumTagName)
+	for i, k := range ks {
+		kn := getIntK(k, len(tss))
+		var sum float64
+		count := 0
+		for _, ts := range tss[:len(tss)-kn] {
+			v := ts.Values[i]
+			if math.IsNaN(v) {
+				continue
+			}
+			sum += v
+			count++
+		}
+		if count == 0 {
+			sum = nan
+		}
+		dst.Values[i] = sum
+	}
+	return &dst
+}
+
 func fillNaNsAtIdx(idx int, k float64, tss []*timeseries) {
+	kn := getIntK(k, len(tss))
+	for _, ts := range tss[:len(tss)-kn] {
+		ts.Values[idx] = nan
+	}
+}
+
+func getIntK(k float64, kMax int) int {
 	if math.IsNaN(k) {
-		k = 0
+		return 0
 	}
 	kn := int(k)
 	if kn < 0 {
-		kn = 0
-	}
-	if kn > len(tss) {
-		kn = len(tss)
+		return 0
 	}
-	for _, ts := range tss[:len(tss)-kn] {
-		ts.Values[idx] = nan
+	if kn > kMax {
+		return kMax
 	}
+	return kn
 }
 
 func minValue(values []float64) float64 {
@@ -746,7 +799,7 @@ func aggrFuncOutliersK(afa *aggrFuncArg) ([]*timeseries, error) {
 	if err != nil {
 		return nil, err
 	}
-	afe := func(tss []*timeseries) []*timeseries {
+	afe := func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
 		// Calculate medians for each point across tss.
 		medians := make([]float64, len(ks))
 		h := histogram.GetFast()
@@ -771,7 +824,7 @@ func aggrFuncOutliersK(afa *aggrFuncArg) ([]*timeseries, error) {
 			}
 			return sum2
 		}
-		return getRangeTopKTimeseries(tss, ks, f, false)
+		return getRangeTopKTimeseries(tss, &afa.ae.Modifier, ks, "", f, false)
 	}
 	return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
 }
@@ -792,7 +845,7 @@ func aggrFuncLimitK(afa *aggrFuncArg) ([]*timeseries, error) {
 			maxK = k
 		}
 	}
-	afe := func(tss []*timeseries) []*timeseries {
+	afe := func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
 		if len(tss) > maxK {
 			tss = tss[:maxK]
 		}
@@ -833,8 +886,8 @@ func aggrFuncMedian(afa *aggrFuncArg) ([]*timeseries, error) {
 	return aggrFuncExt(afe, tss, &afa.ae.Modifier, afa.ae.Limit, false)
 }
 
-func newAggrQuantileFunc(phis []float64) func(tss []*timeseries) []*timeseries {
-	return func(tss []*timeseries) []*timeseries {
+func newAggrQuantileFunc(phis []float64) func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
+	return func(tss []*timeseries, modifier *metricsql.ModifierExpr) []*timeseries {
 		dst := tss[0]
 		h := histogram.GetFast()
 		defer histogram.PutFast(h)
diff --git a/dashboards/victoriametrics.json b/dashboards/victoriametrics.json
index 7cd4f18d5..948c79f43 100644
--- a/dashboards/victoriametrics.json
+++ b/dashboards/victoriametrics.json
@@ -56,7 +56,7 @@
   "gnetId": 10229,
   "graphTooltip": 0,
   "id": null,
-  "iteration": 1599034965731,
+  "iteration": 1603307754894,
   "links": [
     {
       "icon": "doc",
@@ -925,7 +925,7 @@
       "dashLength": 10,
       "dashes": false,
       "datasource": "$ds",
-      "description": "Shows how many ongoing insertions are taking place.\n* `max` - equal to number of CPU * 2\n* `current` - current number of goroutines busy with inserting rows into storage\n\nWhen `current` hits `max` constantly, it means storage is overloaded and require more CPU.",
+      "description": "Shows how many ongoing insertions (not API /write calls) on disk are taking place, where:\n* `max` - equal to number of CPUs;\n* `current` - current number of goroutines busy with inserting rows into underlying storage.\n\nEvery successful API /write call results into flush on disk. However, these two actions are separated and controlled via different concurrency limiters. The `max` on this panel can't be changed and always equal to number of CPUs. \n\nWhen `current` hits `max` constantly, it means storage is overloaded and requires more CPU.\n\n",
       "fieldConfig": {
         "defaults": {
           "custom": {},
@@ -979,6 +979,7 @@
         {
           "expr": "sum(vm_concurrent_addrows_capacity{job=\"$job\", instance=\"$instance\"})",
           "format": "time_series",
+          "interval": "",
           "intervalFactor": 1,
           "legendFormat": "max",
           "refId": "A"
@@ -995,7 +996,7 @@
       "timeFrom": null,
       "timeRegions": [],
       "timeShift": null,
-      "title": "Concurrent inserts ($instance)",
+      "title": "Concurrent flushes on disk ($instance)",
       "tooltip": {
         "shared": true,
         "sort": 2,
@@ -1164,7 +1165,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 36
+            "y": 3
           },
           "hiddenSeries": false,
           "id": 10,
@@ -1250,7 +1251,7 @@
           "dashLength": 10,
           "dashes": false,
           "datasource": "$ds",
-          "description": "How many datapoints are in RAM queue waiting to be written into storage. The number of pending data points should be in the range from 0 to `2*<ingestion_rate>`, since VictoriaMetrics pushes pending data to persistent storage every second.",
+          "description": "Shows the time needed to reach the 100% of disk capacity based on the following params:\n* free disk space;\n* rows ingestion rate;\n* compression.\n\nUse this panel for capacity planning in order to estimate the time remaining for running out of the disk space.\n\n",
           "fieldConfig": {
             "defaults": {
               "custom": {},
@@ -1264,63 +1265,53 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 36
+            "y": 3
           },
           "hiddenSeries": false,
-          "id": 34,
+          "id": 73,
           "legend": {
-            "avg": false,
-            "current": false,
+            "alignAsTable": true,
+            "avg": true,
+            "current": true,
+            "hideZero": true,
             "max": false,
             "min": false,
             "show": false,
             "total": false,
-            "values": false
+            "values": true
           },
           "lines": true,
           "linewidth": 1,
           "links": [],
-          "nullPointMode": "null",
+          "nullPointMode": "null as zero",
           "percentage": false,
           "pluginVersion": "7.1.1",
           "pointradius": 2,
           "points": false,
           "renderer": "flot",
-          "seriesOverrides": [
-            {
-              "alias": "pending index entries",
-              "yaxis": 2
-            }
-          ],
+          "seriesOverrides": [],
           "spaceLength": 10,
           "stack": false,
           "steppedLine": false,
           "targets": [
             {
-              "expr": "vm_pending_rows{job=\"$job\", instance=~\"$instance\", type=\"storage\"}",
+              "expr": "vm_free_disk_space_bytes{job=\"$job\", instance=\"$instance\"} / (sum(rate(vm_rows_added_to_storage_total{job=\"$job\", instance=\"$instance\"}[1d])) * (sum(vm_data_size_bytes{job=\"$job\", instance=\"$instance\", type!=\"indexdb\"}) / sum(vm_rows{job=\"$job\", instance=\"$instance\", type!=\"indexdb\"})))",
               "format": "time_series",
               "hide": false,
+              "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "pending datapoints",
+              "legendFormat": "",
               "refId": "A"
-            },
-            {
-              "expr": "vm_pending_rows{job=\"$job\", instance=~\"$instance\", type=\"indexdb\"}",
-              "format": "time_series",
-              "hide": false,
-              "intervalFactor": 1,
-              "legendFormat": "pending index entries",
-              "refId": "B"
             }
           ],
           "thresholds": [],
           "timeFrom": null,
           "timeRegions": [],
           "timeShift": null,
-          "title": "Pending datapoints ($instance)",
+          "title": "Storage full ETA ($instance)",
           "tooltip": {
             "shared": true,
-            "sort": 0,
+            "sort": 2,
             "value_type": "individual"
           },
           "type": "graph",
@@ -1333,7 +1324,8 @@
           },
           "yaxes": [
             {
-              "format": "short",
+              "decimals": null,
+              "format": "s",
               "label": null,
               "logBase": 1,
               "max": null,
@@ -1341,8 +1333,7 @@
               "show": true
             },
             {
-              "decimals": 3,
-              "format": "none",
+              "format": "short",
               "label": null,
               "logBase": 1,
               "max": null,
@@ -1375,7 +1366,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 44
+            "y": 11
           },
           "hiddenSeries": false,
           "id": 30,
@@ -1472,7 +1463,7 @@
           "dashLength": 10,
           "dashes": false,
           "datasource": "$ds",
-          "description": "Data parts of LSM tree.\nHigh number of parts could be an evidence of slow merge performance - check the resource utilization.\n* `indexdb` - inverted index\n* `storage/small` - recently added parts of data ingested into storage(hot data)\n* `storage/big` -  small parts gradually merged into big parts (cold data)",
+          "description": "How many datapoints are in RAM queue waiting to be written into storage. The number of pending data points should be in the range from 0 to `2*<ingestion_rate>`, since VictoriaMetrics pushes pending data to persistent storage every second.",
           "fieldConfig": {
             "defaults": {
               "custom": {},
@@ -1486,16 +1477,16 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 44
+            "y": 11
           },
           "hiddenSeries": false,
-          "id": 36,
+          "id": 34,
           "legend": {
             "avg": false,
             "current": false,
             "max": false,
             "min": false,
-            "show": true,
+            "show": false,
             "total": false,
             "values": false
           },
@@ -1508,27 +1499,41 @@
           "pointradius": 2,
           "points": false,
           "renderer": "flot",
-          "seriesOverrides": [],
+          "seriesOverrides": [
+            {
+              "alias": "pending index entries",
+              "yaxis": 2
+            }
+          ],
           "spaceLength": 10,
           "stack": false,
           "steppedLine": false,
           "targets": [
             {
-              "expr": "sum(vm_parts{job=\"$job\", instance=\"$instance\"}) by (type)",
+              "expr": "vm_pending_rows{job=\"$job\", instance=~\"$instance\", type=\"storage\"}",
               "format": "time_series",
+              "hide": false,
               "intervalFactor": 1,
-              "legendFormat": "{{type}}",
+              "legendFormat": "pending datapoints",
               "refId": "A"
+            },
+            {
+              "expr": "vm_pending_rows{job=\"$job\", instance=~\"$instance\", type=\"indexdb\"}",
+              "format": "time_series",
+              "hide": false,
+              "intervalFactor": 1,
+              "legendFormat": "pending index entries",
+              "refId": "B"
             }
           ],
           "thresholds": [],
           "timeFrom": null,
           "timeRegions": [],
           "timeShift": null,
-          "title": "LSM parts ($instance)",
+          "title": "Pending datapoints ($instance)",
           "tooltip": {
             "shared": true,
-            "sort": 2,
+            "sort": 0,
             "value_type": "individual"
           },
           "type": "graph",
@@ -1549,7 +1554,8 @@
               "show": true
             },
             {
-              "format": "short",
+              "decimals": 3,
+              "format": "none",
               "label": null,
               "logBase": 1,
               "max": null,
@@ -1582,7 +1588,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 52
+            "y": 19
           },
           "hiddenSeries": false,
           "id": 53,
@@ -1675,7 +1681,7 @@
           "dashLength": 10,
           "dashes": false,
           "datasource": "$ds",
-          "description": "Shows amount of on-disk space occupied by inverted index.",
+          "description": "Data parts of LSM tree.\nHigh number of parts could be an evidence of slow merge performance - check the resource utilization.\n* `indexdb` - inverted index\n* `storage/small` - recently added parts of data ingested into storage(hot data)\n* `storage/big` -  small parts gradually merged into big parts (cold data)",
           "fieldConfig": {
             "defaults": {
               "custom": {},
@@ -1689,16 +1695,16 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 52
+            "y": 19
           },
           "hiddenSeries": false,
-          "id": 55,
+          "id": 36,
           "legend": {
             "avg": false,
             "current": false,
             "max": false,
             "min": false,
-            "show": false,
+            "show": true,
             "total": false,
             "values": false
           },
@@ -1717,9 +1723,10 @@
           "steppedLine": false,
           "targets": [
             {
-              "expr": "vm_data_size_bytes{job=\"$job\", instance=~\"$instance\", type=\"indexdb\"}",
+              "expr": "sum(vm_parts{job=\"$job\", instance=\"$instance\"}) by (type)",
               "format": "time_series",
               "intervalFactor": 1,
+              "legendFormat": "{{type}}",
               "refId": "A"
             }
           ],
@@ -1727,10 +1734,10 @@
           "timeFrom": null,
           "timeRegions": [],
           "timeShift": null,
-          "title": "Disk space usage - index ($instance)",
+          "title": "LSM parts ($instance)",
           "tooltip": {
             "shared": true,
-            "sort": 0,
+            "sort": 2,
             "value_type": "individual"
           },
           "type": "graph",
@@ -1743,7 +1750,7 @@
           },
           "yaxes": [
             {
-              "format": "bytes",
+              "format": "short",
               "label": null,
               "logBase": 1,
               "max": null,
@@ -1784,7 +1791,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 60
+            "y": 27
           },
           "hiddenSeries": false,
           "id": 62,
@@ -1864,7 +1871,7 @@
           "dashLength": 10,
           "dashes": false,
           "datasource": "$ds",
-          "description": "The number of rows merged per second by storage nodes.",
+          "description": "Shows amount of on-disk space occupied by inverted index.",
           "fieldConfig": {
             "defaults": {
               "custom": {},
@@ -1878,21 +1885,22 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 60
+            "y": 27
           },
           "hiddenSeries": false,
-          "id": 64,
+          "id": 55,
           "legend": {
             "avg": false,
             "current": false,
             "max": false,
             "min": false,
-            "show": true,
+            "show": false,
             "total": false,
             "values": false
           },
           "lines": true,
           "linewidth": 1,
+          "links": [],
           "nullPointMode": "null",
           "percentage": false,
           "pluginVersion": "7.1.1",
@@ -1905,8 +1913,9 @@
           "steppedLine": false,
           "targets": [
             {
-              "expr": "sum(rate(vm_rows_merged_total{job=\"$job\", instance=\"$instance\"}[5m])) by(type)",
-              "legendFormat": "{{type}}",
+              "expr": "vm_data_size_bytes{job=\"$job\", instance=~\"$instance\", type=\"indexdb\"}",
+              "format": "time_series",
+              "intervalFactor": 1,
               "refId": "A"
             }
           ],
@@ -1914,7 +1923,7 @@
           "timeFrom": null,
           "timeRegions": [],
           "timeShift": null,
-          "title": "Merge speed ($instance)",
+          "title": "Disk space usage - index ($instance)",
           "tooltip": {
             "shared": true,
             "sort": 0,
@@ -1930,8 +1939,7 @@
           },
           "yaxes": [
             {
-              "decimals": 0,
-              "format": "short",
+              "format": "bytes",
               "label": null,
               "logBase": 1,
               "max": null,
@@ -1972,7 +1980,7 @@
             "h": 8,
             "w": 12,
             "x": 0,
-            "y": 68
+            "y": 35
           },
           "hiddenSeries": false,
           "id": 58,
@@ -2050,6 +2058,100 @@
             "alignLevel": null
           }
         },
+        {
+          "aliasColors": {},
+          "bars": false,
+          "dashLength": 10,
+          "dashes": false,
+          "datasource": "$ds",
+          "description": "The number of rows merged per second by storage nodes.",
+          "fieldConfig": {
+            "defaults": {
+              "custom": {},
+              "links": []
+            },
+            "overrides": []
+          },
+          "fill": 1,
+          "fillGradient": 0,
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 35
+          },
+          "hiddenSeries": false,
+          "id": 64,
+          "legend": {
+            "avg": false,
+            "current": false,
+            "max": false,
+            "min": false,
+            "show": true,
+            "total": false,
+            "values": false
+          },
+          "lines": true,
+          "linewidth": 1,
+          "nullPointMode": "null",
+          "percentage": false,
+          "pluginVersion": "7.1.1",
+          "pointradius": 2,
+          "points": false,
+          "renderer": "flot",
+          "seriesOverrides": [],
+          "spaceLength": 10,
+          "stack": false,
+          "steppedLine": false,
+          "targets": [
+            {
+              "expr": "sum(rate(vm_rows_merged_total{job=\"$job\", instance=\"$instance\"}[5m])) by(type)",
+              "legendFormat": "{{type}}",
+              "refId": "A"
+            }
+          ],
+          "thresholds": [],
+          "timeFrom": null,
+          "timeRegions": [],
+          "timeShift": null,
+          "title": "Merge speed ($instance)",
+          "tooltip": {
+            "shared": true,
+            "sort": 0,
+            "value_type": "individual"
+          },
+          "type": "graph",
+          "xaxis": {
+            "buckets": null,
+            "mode": "time",
+            "name": null,
+            "show": true,
+            "values": []
+          },
+          "yaxes": [
+            {
+              "decimals": 0,
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": "0",
+              "show": true
+            },
+            {
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": "0",
+              "show": true
+            }
+          ],
+          "yaxis": {
+            "align": false,
+            "alignLevel": null
+          }
+        },
         {
           "aliasColors": {},
           "bars": false,
@@ -2070,7 +2172,7 @@
             "h": 8,
             "w": 12,
             "x": 12,
-            "y": 68
+            "y": 43
           },
           "hiddenSeries": false,
           "id": 67,
@@ -3293,4 +3395,4 @@
   "title": "VictoriaMetrics",
   "uid": "wNf0q_kZk",
   "version": 1
-}
+}
\ No newline at end of file
diff --git a/deployment/docker/docker-compose.yml b/deployment/docker/docker-compose.yml
index 0f856b515..8ccab8569 100644
--- a/deployment/docker/docker-compose.yml
+++ b/deployment/docker/docker-compose.yml
@@ -21,7 +21,10 @@ services:
     image: victoriametrics/victoria-metrics
     ports:
       - 8428:8428
+      - 8089:8089
+      - 8089:8089/udp
       - 2003:2003
+      - 2003:2003/udp
       - 4242:4242
     volumes:
       - vmdata:/storage
@@ -30,6 +33,7 @@ services:
       - '--graphiteListenAddr=:2003'
       - '--opentsdbListenAddr=:4242'
       - '--httpListenAddr=:8428'
+      - '--influxListenAddr=:8089'
     networks:
       - vm_net
     restart: always
diff --git a/docs/MetricsQL.md b/docs/MetricsQL.md
index c9c085019..c1f3a9872 100644
--- a/docs/MetricsQL.md
+++ b/docs/MetricsQL.md
@@ -113,6 +113,7 @@ This functionality can be tried at [an editable Grafana dashboard](http://play-g
    - `bottomk_max(k, q)` - returns bottom K time series with the min maximums on the given time range
    - `bottomk_avg(k, q)` - returns bottom K time series with the min averages on the given time range
    - `bottomk_median(k, q)` - returns bottom K time series with the min medians on the given time range
+  All the `topk_*` and `bottomk_*` functions accept optional third argument - label name for the sum of the remaining time series outside top K or bottom K time series. For example, `topk_max(3, process_resident_memory_bytes, "remaining_sum")` would return up to 3 time series with the maximum value for `process_resident_memory_bytes` plus fourth time series with the sum of the remaining time series if any. The fourth time series will contain `remaining_sum="remaining_sum"` additional label.
 - `share_le_over_time(m[d], le)` - returns share (in the range 0..1) of values in `m` over `d`, which are smaller or equal to `le`. Useful for calculating SLI and SLO.
   Example: `share_le_over_time(memory_usage_bytes[24h], 100*1024*1024)` returns the share of time series values for the last 24 hours when memory usage was below or equal to 100MB.
 - `share_gt_over_time(m[d], gt)` - returns share (in the range 0..1) of values in `m` over `d`, which are bigger than `gt`. Useful for calculating SLI and SLO.
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index 16b85aa26..40b212b6d 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -495,6 +495,7 @@ VictoriaMetrics supports the following handlers from [Prometheus querying API](h
 * [/api/v1/labels](https://prometheus.io/docs/prometheus/latest/querying/api/#getting-label-names)
 * [/api/v1/label/.../values](https://prometheus.io/docs/prometheus/latest/querying/api/#querying-label-values)
 * [/api/v1/status/tsdb](https://prometheus.io/docs/prometheus/latest/querying/api/#tsdb-stats)
+* [/api/v1/targets](https://prometheus.io/docs/prometheus/latest/querying/api/#targets) - see [these docs](#how-to-scrape-prometheus-exporters-such-as-node-exporter) for more details.
 
 These handlers can be queried from Prometheus-compatible clients such as Grafana or curl.
 
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 428ea6e97..22d2c6403 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -211,9 +211,13 @@ either via `vmagent` itself or via Prometheus, so the exported metrics could be
 Use official [Grafana dashboard](https://grafana.com/grafana/dashboards/12683) for `vmagent` state overview.
 If you have suggestions, improvements or found a bug - feel free to open an issue on github or add review to the dashboard.
 
-`vmagent` also exports target statuses at `http://vmagent-host:8429/targets` page in plaintext format.
-`/targets` handler accepts optional `show_original_labels=1` query arg, which shows the original labels per each target
-before applying relabeling. This information may be useful for debugging target relabeling.
+`vmagent` also exports target statuses at the following handlers:
+
+* `http://vmagent-host:8429/targets`. This handler returns human-readable plaintext status for every active target.
+This page is convenient to query from command line with `wget`, `curl` or similar tools.
+It accepts optional `show_original_labels=1` query arg, which shows the original labels per each target before applying relabeling.
+This information may be useful for debugging target relabeling.
+* `http://vmagent-host:8429/api/v1/targets`. This handler returns data compatible with [the corresponding page from Prometheus API](https://prometheus.io/docs/prometheus/latest/querying/api/#targets).
 
 
 ### Troubleshooting
@@ -224,7 +228,26 @@ before applying relabeling. This information may be useful for debugging target
   since `vmagent` establishes at least a single TCP connection per each target.
 
 * When `vmagent` scrapes many unreliable targets, it can flood error log with scrape errors. These errors can be suppressed
-  by passing `-promscrape.suppressScrapeErrors` command-line flag to `vmagent`. The most recent scrape error per each target can be observed at `http://vmagent-host:8429/targets`.
+  by passing `-promscrape.suppressScrapeErrors` command-line flag to `vmagent`. The most recent scrape error per each target can be observed at `http://vmagent-host:8429/targets`
+  and `http://vmagent-host:8429/api/v1/targets`.
+
+* If `vmagent` scrapes targets with millions of metrics per each target (for instance, when scraping [federation endpoints](https://prometheus.io/docs/prometheus/latest/federation/)),
+  then it is recommended enabling `stream parsing mode` in order to reduce memory usage during scraping. This mode may be enabled either globally for all the scrape targets
+  by passing `-promscrape.streamParse` command-line flag or on a per-scrape target basis with `stream_parse: true` option. For example:
+
+  ```yml
+  scrape_configs:
+  - job_name: 'big-federate'
+    stream_parse: true
+    static_configs:
+    - targets:
+      - big-prometeus1
+      - big-prometeus2
+    honor_labels: true
+    metrics_path: /federate
+    params:
+      'match[]': ['{__name__!=""}']
+  ```
 
 * It is recommended to increase `-remoteWrite.queues` if `vmagent_remotewrite_pending_data_bytes` metric exported at `http://vmagent-host:8429/metrics` page constantly grows.
 
diff --git a/lib/promscrape/client.go b/lib/promscrape/client.go
index e7f4d0f5e..9504f9f03 100644
--- a/lib/promscrape/client.go
+++ b/lib/promscrape/client.go
@@ -1,13 +1,18 @@
 package promscrape
 
 import (
+	"context"
 	"crypto/tls"
 	"flag"
 	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
 	"strings"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/fasthttp"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -22,11 +27,19 @@ var (
 		"This may be useful when targets has no support for HTTP keep-alive connection. "+
 		"It is possible to set `disable_keepalive: true` individually per each 'scrape_config` section in '-promscrape.config' for fine grained control. "+
 		"Note that disabling HTTP keep-alive may increase load on both vmagent and scrape targets")
+	streamParse = flag.Bool("promscrape.streamParse", false, "Whether to enable stream parsing for metrics obtained from scrape targets. This may be useful "+
+		"for reducing memory usage when millions of metrics are exposed per each scrape target. "+
+		"It is posible to set `stream_parse: true` individually per each `scrape_config` section in `-promscrape.config` for fine grained control")
 )
 
 type client struct {
+	// hc is the default client optimized for common case of scraping targets with moderate number of metrics.
 	hc *fasthttp.HostClient
 
+	// sc (aka `stream client`) is used instead of hc if ScrapeWork.ParseStream is set.
+	// It may be useful for scraping targets with millions of metrics per target.
+	sc *http.Client
+
 	scrapeURL          string
 	host               string
 	requestURI         string
@@ -64,8 +77,23 @@ func newClient(sw *ScrapeWork) *client {
 		MaxResponseBodySize:          maxScrapeSize.N,
 		MaxIdempotentRequestAttempts: 1,
 	}
+	var sc *http.Client
+	if *streamParse || sw.StreamParse {
+		sc = &http.Client{
+			Transport: &http.Transport{
+				TLSClientConfig:     tlsCfg,
+				TLSHandshakeTimeout: 10 * time.Second,
+				IdleConnTimeout:     2 * sw.ScrapeInterval,
+				DisableCompression:  *disableCompression || sw.DisableCompression,
+				DisableKeepAlives:   *disableKeepAlive || sw.DisableKeepAlive,
+				DialContext:         statStdDial,
+			},
+			Timeout: sw.ScrapeTimeout,
+		}
+	}
 	return &client{
 		hc: hc,
+		sc: sc,
 
 		scrapeURL:          sw.ScrapeURL,
 		host:               host,
@@ -76,6 +104,43 @@ func newClient(sw *ScrapeWork) *client {
 	}
 }
 
+func (c *client) GetStreamReader() (*streamReader, error) {
+	deadline := time.Now().Add(c.hc.ReadTimeout)
+	ctx, cancel := context.WithDeadline(context.Background(), deadline)
+	req, err := http.NewRequestWithContext(ctx, "GET", c.scrapeURL, nil)
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("cannot create request for %q: %w", c.scrapeURL, err)
+	}
+	// The following `Accept` header has been copied from Prometheus sources.
+	// See https://github.com/prometheus/prometheus/blob/f9d21f10ecd2a343a381044f131ea4e46381ce09/scrape/scrape.go#L532 .
+	// This is needed as a workaround for scraping stupid Java-based servers such as Spring Boot.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/608 for details.
+	// Do not bloat the `Accept` header with OpenMetrics shit, since it looks like dead standard now.
+	req.Header.Set("Accept", "text/plain;version=0.0.4;q=1,*/*;q=0.1")
+	if c.authHeader != "" {
+		req.Header.Set("Authorization", c.authHeader)
+	}
+	resp, err := c.sc.Do(req)
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("cannot scrape %q: %w", c.scrapeURL, err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_scrapes_total{status_code="%d"}`, resp.StatusCode)).Inc()
+		respBody, _ := ioutil.ReadAll(resp.Body)
+		_ = resp.Body.Close()
+		cancel()
+		return nil, fmt.Errorf("unexpected status code returned when scraping %q: %d; expecting %d; response body: %q",
+			c.scrapeURL, resp.StatusCode, http.StatusOK, respBody)
+	}
+	scrapesOK.Inc()
+	return &streamReader{
+		r:      resp.Body,
+		cancel: cancel,
+	}, nil
+}
+
 func (c *client) ReadData(dst []byte) ([]byte, error) {
 	deadline := time.Now().Add(c.hc.ReadTimeout)
 	req := fasthttp.AcquireRequest()
@@ -87,7 +152,7 @@ func (c *client) ReadData(dst []byte) ([]byte, error) {
 	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/608 for details.
 	// Do not bloat the `Accept` header with OpenMetrics shit, since it looks like dead standard now.
 	req.Header.Set("Accept", "text/plain;version=0.0.4;q=1,*/*;q=0.1")
-	if !*disableCompression || c.disableCompression {
+	if !*disableCompression && !c.disableCompression {
 		req.Header.Set("Accept-Encoding", "gzip")
 	}
 	if *disableKeepAlive || c.disableKeepAlive {
@@ -131,7 +196,6 @@ func (c *client) ReadData(dst []byte) ([]byte, error) {
 		}
 		return dst, fmt.Errorf("error when scraping %q: %w", c.scrapeURL, err)
 	}
-	dstLen := len(dst)
 	if ce := resp.Header.Peek("Content-Encoding"); string(ce) == "gzip" {
 		var err error
 		var src []byte
@@ -154,7 +218,7 @@ func (c *client) ReadData(dst []byte) ([]byte, error) {
 	if statusCode != fasthttp.StatusOK {
 		metrics.GetOrCreateCounter(fmt.Sprintf(`vm_promscrape_scrapes_total{status_code="%d"}`, statusCode)).Inc()
 		return dst, fmt.Errorf("unexpected status code returned when scraping %q: %d; expecting %d; response body: %q",
-			c.scrapeURL, statusCode, fasthttp.StatusOK, dst[dstLen:])
+			c.scrapeURL, statusCode, fasthttp.StatusOK, dst)
 	}
 	scrapesOK.Inc()
 	fasthttp.ReleaseResponse(resp)
@@ -185,3 +249,22 @@ func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request,
 		}
 	}
 }
+
+type streamReader struct {
+	r         io.ReadCloser
+	cancel    context.CancelFunc
+	bytesRead int64
+}
+
+func (sr *streamReader) Read(p []byte) (int, error) {
+	n, err := sr.r.Read(p)
+	sr.bytesRead += int64(n)
+	return n, err
+}
+
+func (sr *streamReader) MustClose() {
+	sr.cancel()
+	if err := sr.r.Close(); err != nil {
+		logger.Errorf("cannot close reader: %s", err)
+	}
+}
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index 9595e7833..d8ca91469 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -84,6 +84,7 @@ type ScrapeConfig struct {
 	// These options are supported only by lib/promscrape.
 	DisableCompression bool `yaml:"disable_compression"`
 	DisableKeepAlive   bool `yaml:"disable_keepalive"`
+	StreamParse        bool `yaml:"stream_parse"`
 
 	// This is set in loadConfig
 	swc *scrapeWorkConfig
@@ -473,6 +474,7 @@ func getScrapeWorkConfig(sc *ScrapeConfig, baseDir string, globalCfg *GlobalConf
 		sampleLimit:          sc.SampleLimit,
 		disableCompression:   sc.DisableCompression,
 		disableKeepAlive:     sc.DisableKeepAlive,
+		streamParse:          sc.StreamParse,
 	}
 	return swc, nil
 }
@@ -493,6 +495,7 @@ type scrapeWorkConfig struct {
 	sampleLimit          int
 	disableCompression   bool
 	disableKeepAlive     bool
+	streamParse          bool
 }
 
 func appendKubernetesScrapeWork(dst []ScrapeWork, sdc *kubernetes.SDConfig, baseDir string, swc *scrapeWorkConfig) ([]ScrapeWork, bool) {
@@ -642,6 +645,7 @@ func appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfig, target string, ex
 	labels = promrelabel.RemoveMetaLabels(labels[:0], labels)
 	if len(labels) == 0 {
 		// Drop target without labels.
+		droppedTargetsMap.Register(originalLabels)
 		return dst, nil
 	}
 	// See https://www.robustperception.io/life-of-a-label
@@ -652,10 +656,12 @@ func appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfig, target string, ex
 	addressRelabeled := promrelabel.GetLabelValueByName(labels, "__address__")
 	if len(addressRelabeled) == 0 {
 		// Drop target without scrape address.
+		droppedTargetsMap.Register(originalLabels)
 		return dst, nil
 	}
 	if strings.Contains(addressRelabeled, "/") {
 		// Drop target with '/'
+		droppedTargetsMap.Register(originalLabels)
 		return dst, nil
 	}
 	addressRelabeled = addMissingPort(schemeRelabeled, addressRelabeled)
@@ -663,6 +669,9 @@ func appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfig, target string, ex
 	if metricsPathRelabeled == "" {
 		metricsPathRelabeled = "/metrics"
 	}
+	if !strings.HasPrefix(metricsPathRelabeled, "/") {
+		metricsPathRelabeled = "/" + metricsPathRelabeled
+	}
 	paramsRelabeled := getParamsFromLabels(labels, swc.params)
 	optionalQuestion := "?"
 	if len(paramsRelabeled) == 0 || strings.Contains(metricsPathRelabeled, "?") {
@@ -696,6 +705,7 @@ func appendScrapeWork(dst []ScrapeWork, swc *scrapeWorkConfig, target string, ex
 		SampleLimit:          swc.sampleLimit,
 		DisableCompression:   swc.disableCompression,
 		DisableKeepAlive:     swc.disableKeepAlive,
+		StreamParse:          swc.streamParse,
 
 		jobNameOriginal: swc.jobName,
 	})
diff --git a/lib/promscrape/scraper.go b/lib/promscrape/scraper.go
index bd611c13e..341e66462 100644
--- a/lib/promscrape/scraper.go
+++ b/lib/promscrape/scraper.go
@@ -284,6 +284,7 @@ func (sg *scraperGroup) update(sws []ScrapeWork) {
 					"original labels for target1: %s; original labels for target2: %s",
 					sw.ScrapeURL, sw.LabelsString(), promLabelsString(originalLabels), promLabelsString(sw.OriginalLabels))
 			}
+			droppedTargetsMap.Register(sw.OriginalLabels)
 			continue
 		}
 		swsMap[key] = sw.OriginalLabels
@@ -333,6 +334,7 @@ func newScraper(sw *ScrapeWork, group string, pushData func(wr *prompbmarshal.Wr
 	sc.sw.Config = *sw
 	sc.sw.ScrapeGroup = group
 	sc.sw.ReadData = c.ReadData
+	sc.sw.GetStreamReader = c.GetStreamReader
 	sc.sw.PushData = pushData
 	return sc
 }
diff --git a/lib/promscrape/scrapework.go b/lib/promscrape/scrapework.go
index 939fff88d..81515c157 100644
--- a/lib/promscrape/scrapework.go
+++ b/lib/promscrape/scrapework.go
@@ -82,19 +82,22 @@ type ScrapeWork struct {
 	// Whether to disable HTTP keep-alive when querying ScrapeURL.
 	DisableKeepAlive bool
 
+	// Whether to parse target responses in a streaming manner.
+	StreamParse bool
+
 	// The original 'job_name'
 	jobNameOriginal string
 }
 
 // key returns unique identifier for the given sw.
 //
-// it can be used for comparing for equality two ScrapeWork objects.
+// it can be used for comparing for equality for two ScrapeWork objects.
 func (sw *ScrapeWork) key() string {
 	// Do not take into account OriginalLabels.
 	key := fmt.Sprintf("ScrapeURL=%s, ScrapeInterval=%s, ScrapeTimeout=%s, HonorLabels=%v, HonorTimestamps=%v, Labels=%s, "+
-		"AuthConfig=%s, MetricRelabelConfigs=%s, SampleLimit=%d, DisableCompression=%v, DisableKeepAlive=%v",
+		"AuthConfig=%s, MetricRelabelConfigs=%s, SampleLimit=%d, DisableCompression=%v, DisableKeepAlive=%v, StreamParse=%v",
 		sw.ScrapeURL, sw.ScrapeInterval, sw.ScrapeTimeout, sw.HonorLabels, sw.HonorTimestamps, sw.LabelsString(),
-		sw.AuthConfig.String(), sw.metricRelabelConfigsString(), sw.SampleLimit, sw.DisableCompression, sw.DisableKeepAlive)
+		sw.AuthConfig.String(), sw.metricRelabelConfigsString(), sw.SampleLimit, sw.DisableCompression, sw.DisableKeepAlive, sw.StreamParse)
 	return key
 }
 
@@ -132,6 +135,9 @@ type scrapeWork struct {
 	// ReadData is called for reading the data.
 	ReadData func(dst []byte) ([]byte, error)
 
+	// GetStreamReader is called if Config.StreamParse is set.
+	GetStreamReader func() (*streamReader, error)
+
 	// PushData is called for pushing collected data.
 	PushData func(wr *prompbmarshal.WriteRequest)
 
@@ -221,6 +227,15 @@ var (
 )
 
 func (sw *scrapeWork) scrapeInternal(scrapeTimestamp, realTimestamp int64) error {
+	if *streamParse || sw.Config.StreamParse {
+		// Read data from scrape targets in streaming manner.
+		// This case is optimized for targets exposing millions and more of metrics per target.
+		return sw.scrapeStream(scrapeTimestamp, realTimestamp)
+	}
+
+	// Common case: read all the data from scrape target to memory (body) and then process it.
+	// This case should work more optimally for than stream parse code above for common case when scrape target exposes
+	// up to a few thouthand metrics.
 	body := leveledbytebufferpool.Get(sw.prevBodyLen)
 	var err error
 	body.B, err = sw.ReadData(body.B[:0])
@@ -281,6 +296,66 @@ func (sw *scrapeWork) scrapeInternal(scrapeTimestamp, realTimestamp int64) error
 	return err
 }
 
+func (sw *scrapeWork) scrapeStream(scrapeTimestamp, realTimestamp int64) error {
+	sr, err := sw.GetStreamReader()
+	if err != nil {
+		return fmt.Errorf("cannot read data: %s", err)
+	}
+	samplesScraped := 0
+	samplesPostRelabeling := 0
+	wc := writeRequestCtxPool.Get(sw.prevRowsLen)
+	var mu sync.Mutex
+	err = parser.ParseStream(sr, scrapeTimestamp, false, func(rows []parser.Row) error {
+		mu.Lock()
+		defer mu.Unlock()
+		samplesScraped += len(rows)
+		for i := range rows {
+			sw.addRowToTimeseries(wc, &rows[i], scrapeTimestamp, true)
+			if len(wc.labels) > 40000 {
+				// Limit the maximum size of wc.writeRequest.
+				// This should reduce memory usage when scraping targets with millions of metrics and/or labels.
+				// For example, when scraping /federate handler from Prometheus - see https://prometheus.io/docs/prometheus/latest/federation/
+				samplesPostRelabeling += len(wc.writeRequest.Timeseries)
+				sw.updateSeriesAdded(wc)
+				startTime := time.Now()
+				sw.PushData(&wc.writeRequest)
+				pushDataDuration.UpdateDuration(startTime)
+				wc.resetNoRows()
+			}
+		}
+		return nil
+	})
+	scrapedSamples.Update(float64(samplesScraped))
+	endTimestamp := time.Now().UnixNano() / 1e6
+	duration := float64(endTimestamp-realTimestamp) / 1e3
+	scrapeDuration.Update(duration)
+	scrapeResponseSize.Update(float64(sr.bytesRead))
+	sr.MustClose()
+	up := 1
+	if err != nil {
+		if samplesScraped == 0 {
+			up = 0
+		}
+		scrapesFailed.Inc()
+	}
+	samplesPostRelabeling += len(wc.writeRequest.Timeseries)
+	sw.updateSeriesAdded(wc)
+	seriesAdded := sw.finalizeSeriesAdded(samplesPostRelabeling)
+	sw.addAutoTimeseries(wc, "up", float64(up), scrapeTimestamp)
+	sw.addAutoTimeseries(wc, "scrape_duration_seconds", duration, scrapeTimestamp)
+	sw.addAutoTimeseries(wc, "scrape_samples_scraped", float64(samplesScraped), scrapeTimestamp)
+	sw.addAutoTimeseries(wc, "scrape_samples_post_metric_relabeling", float64(samplesPostRelabeling), scrapeTimestamp)
+	sw.addAutoTimeseries(wc, "scrape_series_added", float64(seriesAdded), scrapeTimestamp)
+	startTime := time.Now()
+	sw.PushData(&wc.writeRequest)
+	pushDataDuration.UpdateDuration(startTime)
+	sw.prevRowsLen = len(wc.rows.Rows)
+	wc.reset()
+	writeRequestCtxPool.Put(wc)
+	tsmGlobal.Update(&sw.Config, sw.ScrapeGroup, up == 1, realTimestamp, int64(duration*1000), err)
+	return nil
+}
+
 // leveledWriteRequestCtxPool allows reducing memory usage when writeRequesCtx
 // structs contain mixed number of labels.
 //
diff --git a/lib/promscrape/statconn.go b/lib/promscrape/statconn.go
index 10d45f4aa..420b8b624 100644
--- a/lib/promscrape/statconn.go
+++ b/lib/promscrape/statconn.go
@@ -1,14 +1,48 @@
 package promscrape
 
 import (
+	"context"
 	"net"
+	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/fasthttp"
 	"github.com/VictoriaMetrics/metrics"
 )
 
+func statStdDial(ctx context.Context, network, addr string) (net.Conn, error) {
+	d := getStdDialer()
+	conn, err := d.DialContext(ctx, network, addr)
+	dialsTotal.Inc()
+	if err != nil {
+		dialErrors.Inc()
+		return nil, err
+	}
+	conns.Inc()
+	sc := &statConn{
+		Conn: conn,
+	}
+	return sc, nil
+}
+
+func getStdDialer() *net.Dialer {
+	stdDialerOnce.Do(func() {
+		stdDialer = &net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+			DualStack: netutil.TCP6Enabled(),
+		}
+	})
+	return stdDialer
+}
+
+var (
+	stdDialer     *net.Dialer
+	stdDialerOnce sync.Once
+)
+
 func statDial(addr string) (conn net.Conn, err error) {
 	if netutil.TCP6Enabled() {
 		conn, err = fasthttp.DialDualStack(addr)
diff --git a/lib/promscrape/targetstatus.go b/lib/promscrape/targetstatus.go
index 03777a244..48240368c 100644
--- a/lib/promscrape/targetstatus.go
+++ b/lib/promscrape/targetstatus.go
@@ -6,6 +6,10 @@ import (
 	"sort"
 	"sync"
 	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
 var tsmGlobal = newTargetStatusMap()
@@ -15,6 +19,26 @@ func WriteHumanReadableTargetsStatus(w io.Writer, showOriginalLabels bool) {
 	tsmGlobal.WriteHumanReadable(w, showOriginalLabels)
 }
 
+// WriteAPIV1Targets writes /api/v1/targets to w according to https://prometheus.io/docs/prometheus/latest/querying/api/#targets
+func WriteAPIV1Targets(w io.Writer, state string) {
+	if state == "" {
+		state = "any"
+	}
+	fmt.Fprintf(w, `{"status":"success","data":{"activeTargets":`)
+	if state == "active" || state == "any" {
+		tsmGlobal.WriteActiveTargetsJSON(w)
+	} else {
+		fmt.Fprintf(w, `[]`)
+	}
+	fmt.Fprintf(w, `,"droppedTargets":`)
+	if state == "dropped" || state == "any" {
+		droppedTargetsMap.WriteDroppedTargetsJSON(w)
+	} else {
+		fmt.Fprintf(w, `[]`)
+	}
+	fmt.Fprintf(w, `}}`)
+}
+
 type targetStatusMap struct {
 	mu sync.Mutex
 	m  map[uint64]targetStatus
@@ -73,6 +97,66 @@ func (tsm *targetStatusMap) StatusByGroup(group string, up bool) int {
 	return count
 }
 
+// WriteActiveTargetsJSON writes `activeTargets` contents to w according to https://prometheus.io/docs/prometheus/latest/querying/api/#targets
+func (tsm *targetStatusMap) WriteActiveTargetsJSON(w io.Writer) {
+	tsm.mu.Lock()
+	type keyStatus struct {
+		key string
+		st  targetStatus
+	}
+	kss := make([]keyStatus, 0, len(tsm.m))
+	for _, st := range tsm.m {
+		key := promLabelsString(st.sw.OriginalLabels)
+		kss = append(kss, keyStatus{
+			key: key,
+			st:  st,
+		})
+	}
+	tsm.mu.Unlock()
+
+	sort.Slice(kss, func(i, j int) bool {
+		return kss[i].key < kss[j].key
+	})
+	fmt.Fprintf(w, `[`)
+	for i, ks := range kss {
+		st := ks.st
+		fmt.Fprintf(w, `{"discoveredLabels":`)
+		writeLabelsJSON(w, st.sw.OriginalLabels)
+		fmt.Fprintf(w, `,"labels":`)
+		labelsFinalized := promrelabel.FinalizeLabels(nil, st.sw.Labels)
+		writeLabelsJSON(w, labelsFinalized)
+		fmt.Fprintf(w, `,"scrapePool":%q`, st.sw.Job())
+		fmt.Fprintf(w, `,"scrapeUrl":%q`, st.sw.ScrapeURL)
+		errMsg := ""
+		if st.err != nil {
+			errMsg = st.err.Error()
+		}
+		fmt.Fprintf(w, `,"lastError":%q`, errMsg)
+		fmt.Fprintf(w, `,"lastScrape":%q`, time.Unix(st.scrapeTime/1000, (st.scrapeTime%1000)*1e6).Format(time.RFC3339Nano))
+		fmt.Fprintf(w, `,"lastScrapeDuration":%g`, (time.Millisecond * time.Duration(st.scrapeDuration)).Seconds())
+		state := "up"
+		if !st.up {
+			state = "down"
+		}
+		fmt.Fprintf(w, `,"health":%q}`, state)
+		if i+1 < len(kss) {
+			fmt.Fprintf(w, `,`)
+		}
+	}
+	fmt.Fprintf(w, `]`)
+}
+
+func writeLabelsJSON(w io.Writer, labels []prompbmarshal.Label) {
+	fmt.Fprintf(w, `{`)
+	for i, label := range labels {
+		fmt.Fprintf(w, "%q:%q", label.Name, label.Value)
+		if i+1 < len(labels) {
+			fmt.Fprintf(w, `,`)
+		}
+	}
+	fmt.Fprintf(w, `}`)
+}
+
 func (tsm *targetStatusMap) WriteHumanReadable(w io.Writer, showOriginalLabels bool) {
 	byJob := make(map[string][]targetStatus)
 	tsm.mu.Lock()
@@ -143,3 +227,69 @@ type targetStatus struct {
 func (st *targetStatus) getDurationFromLastScrape() time.Duration {
 	return time.Since(time.Unix(st.scrapeTime/1000, (st.scrapeTime%1000)*1e6))
 }
+
+type droppedTargets struct {
+	mu              sync.Mutex
+	m               map[string]droppedTarget
+	lastCleanupTime uint64
+}
+
+type droppedTarget struct {
+	originalLabels []prompbmarshal.Label
+	deadline       uint64
+}
+
+func (dt *droppedTargets) Register(originalLabels []prompbmarshal.Label) {
+	key := promLabelsString(originalLabels)
+	currentTime := fasttime.UnixTimestamp()
+	dt.mu.Lock()
+	dt.m[key] = droppedTarget{
+		originalLabels: originalLabels,
+		deadline:       currentTime + 10*60,
+	}
+	if currentTime-dt.lastCleanupTime > 60 {
+		for k, v := range dt.m {
+			if currentTime > v.deadline {
+				delete(dt.m, k)
+			}
+		}
+		dt.lastCleanupTime = currentTime
+	}
+	dt.mu.Unlock()
+}
+
+// WriteDroppedTargetsJSON writes `droppedTargets` contents to w according to https://prometheus.io/docs/prometheus/latest/querying/api/#targets
+func (dt *droppedTargets) WriteDroppedTargetsJSON(w io.Writer) {
+	dt.mu.Lock()
+	type keyStatus struct {
+		key            string
+		originalLabels []prompbmarshal.Label
+	}
+	kss := make([]keyStatus, 0, len(dt.m))
+	for _, v := range dt.m {
+		key := promLabelsString(v.originalLabels)
+		kss = append(kss, keyStatus{
+			key:            key,
+			originalLabels: v.originalLabels,
+		})
+	}
+	dt.mu.Unlock()
+
+	sort.Slice(kss, func(i, j int) bool {
+		return kss[i].key < kss[j].key
+	})
+	fmt.Fprintf(w, `[`)
+	for i, ks := range kss {
+		fmt.Fprintf(w, `{"discoveredLabels":`)
+		writeLabelsJSON(w, ks.originalLabels)
+		fmt.Fprintf(w, `}`)
+		if i+1 < len(kss) {
+			fmt.Fprintf(w, `,`)
+		}
+	}
+	fmt.Fprintf(w, `]`)
+}
+
+var droppedTargetsMap = &droppedTargets{
+	m: make(map[string]droppedTarget),
+}
diff --git a/lib/protoparser/csvimport/streamparser.go b/lib/protoparser/csvimport/streamparser.go
index 247526b8b..027ecc458 100644
--- a/lib/protoparser/csvimport/streamparser.go
+++ b/lib/protoparser/csvimport/streamparser.go
@@ -23,7 +23,8 @@ var (
 
 // ParseStream parses csv from req and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from req.
+// The callback can be called concurrently multiple times for streamed data from req.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(req *http.Request, callback func(rows []Row) error) error {
diff --git a/lib/protoparser/graphite/streamparser.go b/lib/protoparser/graphite/streamparser.go
index 2eea19bd5..7fa734f9f 100644
--- a/lib/protoparser/graphite/streamparser.go
+++ b/lib/protoparser/graphite/streamparser.go
@@ -23,7 +23,8 @@ var (
 
 // ParseStream parses Graphite lines from r and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from r.
+// The callback can be called concurrently multiple times for streamed data from r.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, callback func(rows []Row) error) error {
diff --git a/lib/protoparser/influx/streamparser.go b/lib/protoparser/influx/streamparser.go
index 408d0a344..d302e3ef0 100644
--- a/lib/protoparser/influx/streamparser.go
+++ b/lib/protoparser/influx/streamparser.go
@@ -24,7 +24,8 @@ var (
 
 // ParseStream parses r with the given args and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from r.
+// The callback can be called concurrently multiple times for streamed data from r.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, isGzipped bool, precision, db string, callback func(db string, rows []Row) error) error {
diff --git a/lib/protoparser/native/streamparser.go b/lib/protoparser/native/streamparser.go
index 16af474e6..0e87d8787 100644
--- a/lib/protoparser/native/streamparser.go
+++ b/lib/protoparser/native/streamparser.go
@@ -17,7 +17,8 @@ import (
 
 // ParseStream parses /api/v1/import/native lines from req and calls callback for parsed blocks.
 //
-// The callback can be called multiple times for streamed data from req.
+// The callback can be called concurrently multiple times for streamed data from req.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold block after returning.
 // callback can be called in parallel from multiple concurrent goroutines.
diff --git a/lib/protoparser/opentsdb/streamparser.go b/lib/protoparser/opentsdb/streamparser.go
index cf2b4f89e..752f1cd5e 100644
--- a/lib/protoparser/opentsdb/streamparser.go
+++ b/lib/protoparser/opentsdb/streamparser.go
@@ -23,7 +23,8 @@ var (
 
 // ParseStream parses OpenTSDB lines from r and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from r.
+// The callback can be called concurrently multiple times for streamed data from r.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, callback func(rows []Row) error) error {
diff --git a/lib/protoparser/opentsdbhttp/streamparser.go b/lib/protoparser/opentsdbhttp/streamparser.go
index 9a7ffaa5f..5a1f9012e 100644
--- a/lib/protoparser/opentsdbhttp/streamparser.go
+++ b/lib/protoparser/opentsdbhttp/streamparser.go
@@ -26,7 +26,8 @@ var (
 
 // ParseStream parses OpenTSDB http lines from req and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from req.
+// The callback can be called concurrently multiple times for streamed data from req.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(req *http.Request, callback func(rows []Row) error) error {
diff --git a/lib/protoparser/prometheus/streamparser.go b/lib/protoparser/prometheus/streamparser.go
index cfee948cb..8c915c9fd 100644
--- a/lib/protoparser/prometheus/streamparser.go
+++ b/lib/protoparser/prometheus/streamparser.go
@@ -16,7 +16,8 @@ import (
 
 // ParseStream parses lines with Prometheus exposition format from r and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from r.
+// The callback can be called concurrently multiple times for streamed data from r.
+// It is guaranteed that the callback isn't called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
 func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback func(rows []Row) error) error {
@@ -32,11 +33,17 @@ func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback f
 	defer putStreamContext(ctx)
 	for ctx.Read() {
 		uw := getUnmarshalWork()
-		uw.callback = callback
+		uw.callback = func(rows []Row) error {
+			err := callback(rows)
+			ctx.wg.Done()
+			return err
+		}
 		uw.defaultTimestamp = defaultTimestamp
 		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
+		ctx.wg.Add(1)
 		common.ScheduleUnmarshalWork(uw)
 	}
+	ctx.wg.Wait() // wait for all the outstanding callback calls before returning
 	return ctx.Error()
 }
 
@@ -61,6 +68,8 @@ type streamContext struct {
 	reqBuf  []byte
 	tailBuf []byte
 	err     error
+
+	wg sync.WaitGroup
 }
 
 func (ctx *streamContext) Error() error {
diff --git a/lib/protoparser/promremotewrite/streamparser.go b/lib/protoparser/promremotewrite/streamparser.go
index 4312310d6..0ce5632ce 100644
--- a/lib/protoparser/promremotewrite/streamparser.go
+++ b/lib/protoparser/promremotewrite/streamparser.go
@@ -21,6 +21,9 @@ var maxInsertRequestSize = flagutil.NewBytes("maxInsertRequestSize", 32*1024*102
 
 // ParseStream parses Prometheus remote_write message req and calls callback for the parsed timeseries.
 //
+// The callback can be called concurrently multiple times for streamed data from req.
+// The callback can be called after ParseStream returns.
+//
 // callback shouldn't hold tss after returning.
 func ParseStream(req *http.Request, callback func(tss []prompb.TimeSeries) error) error {
 	ctx := getPushCtx(req.Body)
diff --git a/lib/protoparser/vmimport/streamparser.go b/lib/protoparser/vmimport/streamparser.go
index 915ac18e8..66ad6c322 100644
--- a/lib/protoparser/vmimport/streamparser.go
+++ b/lib/protoparser/vmimport/streamparser.go
@@ -20,10 +20,10 @@ var maxLineLen = flagutil.NewBytes("import.maxLineLen", 100*1024*1024, "The maxi
 
 // ParseStream parses /api/v1/import lines from req and calls callback for the parsed rows.
 //
-// The callback can be called multiple times for streamed data from req.
+// The callback can be called concurrently multiple times for streamed data from req.
+// The callback can be called after ParseStream returns.
 //
 // callback shouldn't hold rows after returning.
-// callback is called from multiple concurrent goroutines.
 func ParseStream(req *http.Request, callback func(rows []Row) error) error {
 	r := req.Body
 	if req.Header.Get("Content-Encoding") == "gzip" {
diff --git a/lib/storage/block.go b/lib/storage/block.go
index ec0b3fdeb..c446c370e 100644
--- a/lib/storage/block.go
+++ b/lib/storage/block.go
@@ -23,7 +23,7 @@ const (
 type Block struct {
 	bh blockHeader
 
-	// nextIdx is the next row index for timestamps and values.
+	// nextIdx is the next index for reading timestamps and values.
 	nextIdx int
 
 	timestamps []int64
diff --git a/lib/storage/merge.go b/lib/storage/merge.go
index 6a1845ce9..0fc16d434 100644
--- a/lib/storage/merge.go
+++ b/lib/storage/merge.go
@@ -40,8 +40,11 @@ var errForciblyStopped = fmt.Errorf("forcibly stopped")
 
 func mergeBlockStreamsInternal(ph *partHeader, bsw *blockStreamWriter, bsm *blockStreamMerger, stopCh <-chan struct{},
 	dmis *uint64set.Set, retentionDeadline int64, rowsMerged, rowsDeleted *uint64) error {
-	// Search for the first block to merge
-	var pendingBlock *Block
+	pendingBlockIsEmpty := true
+	pendingBlock := getBlock()
+	defer putBlock(pendingBlock)
+	tmpBlock := getBlock()
+	defer putBlock(tmpBlock)
 	for bsm.NextBlock() {
 		select {
 		case <-stopCh:
@@ -58,31 +61,10 @@ func mergeBlockStreamsInternal(ph *partHeader, bsw *blockStreamWriter, bsm *bloc
 			*rowsDeleted += uint64(bsm.Block.bh.RowsCount)
 			continue
 		}
-		pendingBlock = getBlock()
-		pendingBlock.CopyFrom(bsm.Block)
-		break
-	}
-	if pendingBlock != nil {
-		defer putBlock(pendingBlock)
-	}
-
-	// Merge blocks.
-	tmpBlock := getBlock()
-	defer putBlock(tmpBlock)
-	for bsm.NextBlock() {
-		select {
-		case <-stopCh:
-			return errForciblyStopped
-		default:
-		}
-		if dmis.Has(bsm.Block.bh.TSID.MetricID) {
-			// Skip blocks for deleted metrics.
-			*rowsDeleted += uint64(bsm.Block.bh.RowsCount)
-			continue
-		}
-		if bsm.Block.bh.MaxTimestamp < retentionDeadline {
-			// skip blocks out of the given retention.
-			*rowsDeleted += uint64(bsm.Block.bh.RowsCount)
+		if pendingBlockIsEmpty {
+			// Load the next block if pendingBlock is empty.
+			pendingBlock.CopyFrom(bsm.Block)
+			pendingBlockIsEmpty = false
 			continue
 		}
 
@@ -114,16 +96,20 @@ func mergeBlockStreamsInternal(ph *partHeader, bsw *blockStreamWriter, bsm *bloc
 		tmpBlock.bh.TSID = bsm.Block.bh.TSID
 		tmpBlock.bh.Scale = bsm.Block.bh.Scale
 		tmpBlock.bh.PrecisionBits = minUint8(pendingBlock.bh.PrecisionBits, bsm.Block.bh.PrecisionBits)
-		mergeBlocks(tmpBlock, pendingBlock, bsm.Block)
+		mergeBlocks(tmpBlock, pendingBlock, bsm.Block, retentionDeadline, rowsDeleted)
 		if len(tmpBlock.timestamps) <= maxRowsPerBlock {
 			// More entries may be added to tmpBlock. Swap it with pendingBlock,
 			// so more entries may be added to pendingBlock on the next iteration.
-			tmpBlock.fixupTimestamps()
+			if len(tmpBlock.timestamps) > 0 {
+				tmpBlock.fixupTimestamps()
+			} else {
+				pendingBlockIsEmpty = true
+			}
 			pendingBlock, tmpBlock = tmpBlock, pendingBlock
 			continue
 		}
 
-		// Write the first len(maxRowsPerBlock) of tmpBlock.timestamps to bsw,
+		// Write the first maxRowsPerBlock of tmpBlock.timestamps to bsw,
 		// leave the rest in pendingBlock.
 		tmpBlock.nextIdx = maxRowsPerBlock
 		pendingBlock.CopyFrom(tmpBlock)
@@ -137,18 +123,21 @@ func mergeBlockStreamsInternal(ph *partHeader, bsw *blockStreamWriter, bsm *bloc
 	if err := bsm.Error(); err != nil {
 		return fmt.Errorf("cannot read block to be merged: %w", err)
 	}
-	if pendingBlock != nil {
+	if !pendingBlockIsEmpty {
 		bsw.WriteExternalBlock(pendingBlock, ph, rowsMerged)
 	}
 	return nil
 }
 
 // mergeBlocks merges ib1 and ib2 to ob.
-func mergeBlocks(ob, ib1, ib2 *Block) {
+func mergeBlocks(ob, ib1, ib2 *Block, retentionDeadline int64, rowsDeleted *uint64) {
 	ib1.assertMergeable(ib2)
 	ib1.assertUnmarshaled()
 	ib2.assertUnmarshaled()
 
+	skipSamplesOutsideRetention(ib1, retentionDeadline, rowsDeleted)
+	skipSamplesOutsideRetention(ib2, retentionDeadline, rowsDeleted)
+
 	if ib1.bh.MaxTimestamp < ib2.bh.MinTimestamp {
 		// Fast path - ib1 values have smaller timestamps than ib2 values.
 		appendRows(ob, ib1)
@@ -186,6 +175,16 @@ func mergeBlocks(ob, ib1, ib2 *Block) {
 	}
 }
 
+func skipSamplesOutsideRetention(b *Block, retentionDeadline int64, rowsDeleted *uint64) {
+	timestamps := b.timestamps
+	nextIdx := b.nextIdx
+	for nextIdx < len(timestamps) && timestamps[nextIdx] < retentionDeadline {
+		nextIdx++
+	}
+	*rowsDeleted += uint64(nextIdx - b.nextIdx)
+	b.nextIdx = nextIdx
+}
+
 func appendRows(ob, ib *Block) {
 	ob.timestamps = append(ob.timestamps, ib.timestamps[ib.nextIdx:]...)
 	ob.values = append(ob.values, ib.values[ib.nextIdx:]...)
@@ -199,7 +198,7 @@ func unmarshalAndCalibrateScale(b1, b2 *Block) error {
 		return err
 	}
 
-	scale := decimal.CalibrateScale(b1.values, b1.bh.Scale, b2.values, b2.bh.Scale)
+	scale := decimal.CalibrateScale(b1.values[b1.nextIdx:], b1.bh.Scale, b2.values[b2.nextIdx:], b2.bh.Scale)
 	b1.bh.Scale = scale
 	b2.bh.Scale = scale
 	return nil
diff --git a/lib/storage/metric_name.go b/lib/storage/metric_name.go
index 385f51a9b..f70b83c2a 100644
--- a/lib/storage/metric_name.go
+++ b/lib/storage/metric_name.go
@@ -196,6 +196,10 @@ func (mn *MetricName) CopyFrom(src *MetricName) {
 
 // AddTag adds new tag to mn with the given key and value.
 func (mn *MetricName) AddTag(key, value string) {
+	if key == string(metricGroupTagKey) {
+		mn.MetricGroup = append(mn.MetricGroup, value...)
+		return
+	}
 	tag := mn.addNextTag()
 	tag.Key = append(tag.Key[:0], key...)
 	tag.Value = append(tag.Value[:0], value...)
@@ -203,6 +207,10 @@ func (mn *MetricName) AddTag(key, value string) {
 
 // AddTagBytes adds new tag to mn with the given key and value.
 func (mn *MetricName) AddTagBytes(key, value []byte) {
+	if string(key) == string(metricGroupTagKey) {
+		mn.MetricGroup = append(mn.MetricGroup, value...)
+		return
+	}
 	tag := mn.addNextTag()
 	tag.Key = append(tag.Key[:0], key...)
 	tag.Value = append(tag.Value[:0], value...)
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index ee8e53ca7..a4106901c 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -50,9 +50,9 @@ type Storage struct {
 	slowPerDayIndexInserts uint64
 	slowMetricNameLoads    uint64
 
-	path            string
-	cachePath       string
-	retentionMonths int
+	path           string
+	cachePath      string
+	retentionMsecs int64
 
 	// lock file for exclusive access to the storage on the given path.
 	flockF *os.File
@@ -118,11 +118,10 @@ func OpenStorage(path string, retentionMsecs int64) (*Storage, error) {
 	if retentionMsecs <= 0 {
 		retentionMsecs = maxRetentionMsecs
 	}
-	retentionMonths := (retentionMsecs + (msecsPerMonth - 1)) / msecsPerMonth
 	s := &Storage{
-		path:            path,
-		cachePath:       path + "/cache",
-		retentionMonths: int(retentionMonths),
+		path:           path,
+		cachePath:      path + "/cache",
+		retentionMsecs: retentionMsecs,
 
 		stop: make(chan struct{}),
 	}
@@ -473,8 +472,9 @@ func (s *Storage) startRetentionWatcher() {
 }
 
 func (s *Storage) retentionWatcher() {
+	retentionMonths := int((s.retentionMsecs + (msecsPerMonth - 1)) / msecsPerMonth)
 	for {
-		d := nextRetentionDuration(s.retentionMonths)
+		d := nextRetentionDuration(retentionMonths)
 		select {
 		case <-s.stop:
 			return


Test patch:
diff --git a/app/vmalert/alerting_test.go b/app/vmalert/alerting_test.go
index cace27e6d..55aa2b48d 100644
--- a/app/vmalert/alerting_test.go
+++ b/app/vmalert/alerting_test.go
@@ -355,6 +355,7 @@ func TestAlertingRule_Restore(t *testing.T) {
 				metricWithValueAndLabels(t, float64(time.Now().Truncate(time.Hour).Unix()),
 					"__name__", alertForStateMetricName,
 					alertNameLabel, "",
+					alertGroupNameLabel, "groupID",
 					"foo", "bar",
 					"namespace", "baz",
 				),
diff --git a/app/vmselect/promql/exec_test.go b/app/vmselect/promql/exec_test.go
index 3c5c5501d..977ae0575 100644
--- a/app/vmselect/promql/exec_test.go
+++ b/app/vmselect/promql/exec_test.go
@@ -4193,7 +4193,7 @@ func TestExecSuccess(t *testing.T) {
 	})
 	t.Run(`topk_max(1)`, func(t *testing.T) {
 		t.Parallel()
-		q := `sort(topk_max(1, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss")))`
+		q := `topk_max(1, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss"))`
 		r1 := netstorage.Result{
 			MetricName: metricNameExpected,
 			Values:     []float64{nan, nan, nan, 10.666666666666666, 12, 13.333333333333334},
@@ -4206,6 +4206,84 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1}
 		f(q, resultExpected)
 	})
+	t.Run(`topk_max(1, remaining_sum)`, func(t *testing.T) {
+		t.Parallel()
+		q := `sort_desc(topk_max(1, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss"), "remaining_sum"))`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{nan, nan, nan, 10.666666666666666, 12, 13.333333333333334},
+			Timestamps: timestampsExpected,
+		}
+		r1.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("baz"),
+			Value: []byte("sss"),
+		}}
+		r2 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{10, 10, 10, 10, 10, 10},
+			Timestamps: timestampsExpected,
+		}
+		r2.MetricName.Tags = []storage.Tag{
+			{
+				Key:   []byte("remaining_sum"),
+				Value: []byte("remaining_sum"),
+			},
+		}
+		resultExpected := []netstorage.Result{r1, r2}
+		f(q, resultExpected)
+	})
+	t.Run(`topk_max(2, remaining_sum)`, func(t *testing.T) {
+		t.Parallel()
+		q := `sort_desc(topk_max(2, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss"), "remaining_sum"))`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{nan, nan, nan, 10.666666666666666, 12, 13.333333333333334},
+			Timestamps: timestampsExpected,
+		}
+		r1.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("baz"),
+			Value: []byte("sss"),
+		}}
+		r2 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{10, 10, 10, 10, 10, 10},
+			Timestamps: timestampsExpected,
+		}
+		r2.MetricName.Tags = []storage.Tag{
+			{
+				Key:   []byte("foo"),
+				Value: []byte("bar"),
+			},
+		}
+		resultExpected := []netstorage.Result{r1, r2}
+		f(q, resultExpected)
+	})
+	t.Run(`topk_max(3, remaining_sum)`, func(t *testing.T) {
+		t.Parallel()
+		q := `sort_desc(topk_max(3, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss"), "remaining_sum"))`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{nan, nan, nan, 10.666666666666666, 12, 13.333333333333334},
+			Timestamps: timestampsExpected,
+		}
+		r1.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("baz"),
+			Value: []byte("sss"),
+		}}
+		r2 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{10, 10, 10, 10, 10, 10},
+			Timestamps: timestampsExpected,
+		}
+		r2.MetricName.Tags = []storage.Tag{
+			{
+				Key:   []byte("foo"),
+				Value: []byte("bar"),
+			},
+		}
+		resultExpected := []netstorage.Result{r1, r2}
+		f(q, resultExpected)
+	})
 	t.Run(`bottomk_max(1)`, func(t *testing.T) {
 		t.Parallel()
 		q := `sort(bottomk_max(1, label_set(10, "foo", "bar") or label_set(time()/150, "baz", "sss")))`
diff --git a/lib/promscrape/config_test.go b/lib/promscrape/config_test.go
index e84567b40..380c481e7 100644
--- a/lib/promscrape/config_test.go
+++ b/lib/promscrape/config_test.go
@@ -1276,6 +1276,7 @@ scrape_configs:
     sample_limit: 100
     disable_keepalive: true
     disable_compression: true
+    stream_parse: true
     static_configs:
       - targets:
         - 192.168.1.2  # SNMP device.
@@ -1328,9 +1329,49 @@ scrape_configs:
 			SampleLimit:        100,
 			DisableKeepAlive:   true,
 			DisableCompression: true,
+			StreamParse:        true,
 			jobNameOriginal:    "snmp",
 		},
 	})
+	f(`
+scrape_configs:
+- job_name: path wo slash
+  static_configs: 
+  - targets: ["foo.bar:1234"]
+  relabel_configs:
+  - replacement: metricspath
+    target_label: __metrics_path__
+`, []ScrapeWork{
+		{
+			ScrapeURL:      "http://foo.bar:1234/metricspath",
+			ScrapeInterval: defaultScrapeInterval,
+			ScrapeTimeout:  defaultScrapeTimeout,
+			Labels: []prompbmarshal.Label{
+				{
+					Name:  "__address__",
+					Value: "foo.bar:1234",
+				},
+				{
+					Name:  "__metrics_path__",
+					Value: "metricspath",
+				},
+				{
+					Name:  "__scheme__",
+					Value: "http",
+				},
+				{
+					Name:  "instance",
+					Value: "foo.bar:1234",
+				},
+				{
+					Name:  "job",
+					Value: "path wo slash",
+				},
+			},
+			jobNameOriginal: "path wo slash",
+			AuthConfig:      &promauth.Config{},
+		},
+	})
 }
 
 var defaultRegexForRelabelConfig = regexp.MustCompile("^(.*)$")