Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6173 (2024-04-23 11:58:51)

streamaggr: made labels compressor shared
### Describe Your Changes

Though labels compressor is quite resource intensive, each aggregator and deduplicator instance has it's own compressor. Made it shared across all aggregators to consume less resources while using multiple aggregators

### Checklist

The following checks are **mandatory**:

- [ ] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
Attention: Patch coverage is `52.38095%` with `10 lines` in your changes are missing coverage. Please review.
> Project coverage is 56.85%. Comparing base [(`8aaa828`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/8aaa828ba3f49a43cb8de89a0bcae160f7988c14?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) to head [(`d86d4ce`)](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Report is 606 commits behind head on master.

| [Files](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Patch % | Lines |
|---|---|---|
| [app/vmagent/remotewrite/remotewrite.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?src=pr&el=tree&filepath=app%2Fvmagent%2Fremotewrite%2Fremotewrite.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWdlbnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |
| [app/vminsert/common/streamaggr.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?src=pr&el=tree&filepath=app%2Fvminsert%2Fcommon%2Fstreamaggr.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtaW5zZXJ0L2NvbW1vbi9zdHJlYW1hZ2dyLmdv) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |
| [lib/streamaggr/streamaggr.go](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?src=pr&el=tree&filepath=lib%2Fstreamaggr%2Fstreamaggr.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0cmVhbWFnZ3Ivc3RyZWFtYWdnci5nbw==) | 77.77% | [2 Missing :warning: ](https://app.codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/6173?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) |

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


```diff
@@            Coverage Diff             @@
##           master    #6173      +/-   ##
==========================================
- Coverage   60.37%   56.85%   -3.53%     
==========================================
  Files         411      527     +116     
  Lines       76609    72414    -4195     
==========================================
- Hits        46253    41169    -5084     
- Misses      27794    28349     +555     
- Partials     2562     2896     +334     
```



</details>

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


Patch:
diff --git a/README.md b/README.md
index 87d5ceb80..4e2457d54 100644
--- a/README.md
+++ b/README.md
@@ -86,6 +86,7 @@ VictoriaMetrics has the following prominent features:
   * [Prometheus exposition format](#how-to-import-data-in-prometheus-exposition-format).
   * [InfluxDB line protocol](#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf) over HTTP, TCP and UDP.
   * [Graphite plaintext protocol](#how-to-send-data-from-graphite-compatible-agents-such-as-statsd) with [tags](https://graphite.readthedocs.io/en/latest/tags.html#carbon).
+  * [Statsd plaintext protocol](#how-to-send-data-from-statsd-compatible-clients)
   * [OpenTSDB put message](#sending-data-via-telnet-put-protocol).
   * [HTTP OpenTSDB /api/put requests](#sending-opentsdb-data-via-http-apiput-requests).
   * [JSON line format](#how-to-import-data-in-json-line-format).
@@ -701,6 +702,45 @@ The `/api/v1/export` endpoint should return the following response:
 {"metric":{"__name__":"measurement_field2","tag1":"value1","tag2":"value2"},"values":[1.23],"timestamps":[1695902762311]}
 ```
 
+## How to send data from Statsd-compatible clients
+
+VictoriaMetrics supports extended statsd protocol with tags. Also it does not support sampling and metric types(it will be ignored).  
+Enable Statsd receiver in VictoriaMetrics by setting `-statsdListenAddr` command line flag. For instance,
+the following command will enable Statsd receiver in VictoriaMetrics on TCP and UDP port `8125`:
+
+```console
+/path/to/victoria-metrics-prod -statsdListenAddr=:8125
+```
+
+Example for writing data with Statsd plaintext protocol to local VictoriaMetrics using `nc`:
+
+```console
+echo "foo.bar:123|g|#foo:bar" | nc -N localhost 8125
+```
+
+Explicit setting of timestamps is not supported for statsd protocol. Timestamp is set to the current time when VictoriaMetrics or vmagent receives it.
+
+An arbitrary number of lines delimited by `\n` (aka newline char) can be sent in one go.
+After that the data may be read via [/api/v1/export](#how-to-export-data-in-json-line-format) endpoint:
+
+<div class="with-copy" markdown="1">
+
+```console
+curl -G 'http://localhost:8428/api/v1/export' -d 'match=foo.bar.baz'
+```
+
+</div>
+
+The `/api/v1/export` endpoint should return the following response:
+
+```json
+{"metric":{"__name__":"foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123],"timestamps":[1560277406000]}
+```
+
+Some examples of compatible statsd clients:
+- [statsd-instrument](https://github.com/Shopify/statsd-instrument)
+- [dogstatsd-ruby](https://github.com/DataDog/dogstatsd-ruby)
+- [go-statsd-client](https://github.com/cactus/go-statsd-client)
 ## How to send data from Graphite-compatible agents such as [StatsD](https://github.com/etsy/statsd)
 
 Enable Graphite receiver in VictoriaMetrics by setting `-graphiteListenAddr` command line flag. For instance,
@@ -1356,6 +1396,7 @@ Additionally, VictoriaMetrics can accept metrics via the following popular data
 * DataDog `submit metrics` API. See [these docs](#how-to-send-data-from-datadog-agent) for details.
 * InfluxDB line protocol. See [these docs](#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf) for details.
 * Graphite plaintext protocol. See [these docs](#how-to-send-data-from-graphite-compatible-agents-such-as-statsd) for details.
+* Statsd plaintext protocol. See [these docs](#how-to-send-data-from-statsd-compatible-clients) for details.
 * OpenTelemetry http API. See [these docs](#sending-data-via-opentelemetry) for details.
 * OpenTSDB telnet put protocol. See [these docs](#sending-data-via-telnet-put-protocol) for details.
 * OpenTSDB http `/api/put` protocol. See [these docs](#sending-opentsdb-data-via-http-apiput-requests) for details.
diff --git a/app/vmagent/main.go b/app/vmagent/main.go
index 0d8c910ed..e22fb4703 100644
--- a/app/vmagent/main.go
+++ b/app/vmagent/main.go
@@ -24,6 +24,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/prometheusimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/promremotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/statsd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
@@ -36,6 +37,7 @@ import (
 	influxserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/influx"
 	opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
 	opentsdbhttpserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdbhttp"
+	statsdserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/statsd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
@@ -61,6 +63,10 @@ var (
 		"See also -graphiteListenAddr.useProxyProtocol")
 	graphiteUseProxyProtocol = flag.Bool("graphiteListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -graphiteListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
+	statsdListenAddr = flag.String("statsdListenAddr", "", "TCP and UDP address to listen for Statsd plaintext data. Usually :8125 must be set. Doesn't work if empty. "+
+		"See also -statsdListenAddr.useProxyProtocol")
+	statsdUseProxyProtocol = flag.Bool("statsdListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -statsdListenAddr . "+
+		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
 	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpenTSDB metrics. "+
 		"Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. "+
 		"Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol")
@@ -80,6 +86,7 @@ var (
 var (
 	influxServer       *influxserver.Server
 	graphiteServer     *graphiteserver.Server
+	statsdServer       *statsdserver.Server
 	opentsdbServer     *opentsdbserver.Server
 	opentsdbhttpServer *opentsdbhttpserver.Server
 )
@@ -137,6 +144,9 @@ func main() {
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, *graphiteUseProxyProtocol, graphite.InsertHandler)
 	}
+	if len(*statsdListenAddr) > 0 {
+		statsdServer = statsdserver.MustStart(*statsdListenAddr, *statsdUseProxyProtocol, statsd.InsertHandler)
+	}
 	if len(*opentsdbListenAddr) > 0 {
 		httpInsertHandler := getOpenTSDBHTTPInsertHandler()
 		opentsdbServer = opentsdbserver.MustStart(*opentsdbListenAddr, *opentsdbUseProxyProtocol, opentsdb.InsertHandler, httpInsertHandler)
@@ -172,6 +182,9 @@ func main() {
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer.MustStop()
 	}
+	if len(*statsdListenAddr) > 0 {
+		statsdServer.MustStop()
+	}
 	if len(*opentsdbListenAddr) > 0 {
 		opentsdbServer.MustStop()
 	}
diff --git a/app/vmagent/statsd/request_handler.go b/app/vmagent/statsd/request_handler.go
new file mode 100644
index 000000000..9cf8fd667
--- /dev/null
+++ b/app/vmagent/statsd/request_handler.go
@@ -0,0 +1,68 @@
+package statsd
+
+import (
+	"io"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/statsd"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/statsd/stream"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	rowsInserted  = metrics.NewCounter(`vmagent_rows_inserted_total{type="statsd"}`)
+	rowsPerInsert = metrics.NewHistogram(`vmagent_rows_per_insert{type="statsd"}`)
+)
+
+// InsertHandler processes remote write for statsd plaintext protocol.
+//
+// See https://github.com/statsd/statsd/blob/master/docs/metric_types.md
+func InsertHandler(r io.Reader) error {
+	return stream.Parse(r, false, func(rows []parser.Row) error {
+		return insertRows(nil, rows)
+	})
+}
+
+func insertRows(at *auth.Token, rows []parser.Row) error {
+	ctx := common.GetPushCtx()
+	defer common.PutPushCtx(ctx)
+
+	tssDst := ctx.WriteRequest.Timeseries[:0]
+	labels := ctx.Labels[:0]
+	samples := ctx.Samples[:0]
+	for i := range rows {
+		r := &rows[i]
+		labelsLen := len(labels)
+		labels = append(labels, prompbmarshal.Label{
+			Name:  "__name__",
+			Value: r.Metric,
+		})
+		for j := range r.Tags {
+			tag := &r.Tags[j]
+			labels = append(labels, prompbmarshal.Label{
+				Name:  tag.Key,
+				Value: tag.Value,
+			})
+		}
+		samples = append(samples, prompbmarshal.Sample{
+			Value:     r.Value,
+			Timestamp: r.Timestamp,
+		})
+		tssDst = append(tssDst, prompbmarshal.TimeSeries{
+			Labels:  labels[labelsLen:],
+			Samples: samples[len(samples)-1:],
+		})
+	}
+	ctx.WriteRequest.Timeseries = tssDst
+	ctx.Labels = labels
+	ctx.Samples = samples
+	if !remotewrite.TryPush(at, &ctx.WriteRequest) {
+		return remotewrite.ErrQueueFullHTTPRetry
+	}
+	rowsInserted.Add(len(rows))
+	rowsPerInsert.Update(float64(len(rows)))
+	return nil
+}
diff --git a/app/vminsert/main.go b/app/vminsert/main.go
index f7c383e11..3333e5c13 100644
--- a/app/vminsert/main.go
+++ b/app/vminsert/main.go
@@ -26,6 +26,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/prompush"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/promremotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/statsd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@@ -36,6 +37,7 @@ import (
 	influxserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/influx"
 	opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
 	opentsdbhttpserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdbhttp"
+	statsdserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/statsd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
@@ -49,6 +51,10 @@ var (
 		"See also -graphiteListenAddr.useProxyProtocol")
 	graphiteUseProxyProtocol = flag.Bool("graphiteListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -graphiteListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
+	statsdListenAddr = flag.String("statsdListenAddr", "", "TCP and UDP address to listen for Statsd plaintext data. Usually :8125 must be set. Doesn't work if empty. "+
+		"See also -statsdListenAddr.useProxyProtocol")
+	statsdUseProxyProtocol = flag.Bool("statsdListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -statsdListenAddr . "+
+		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
 	influxListenAddr = flag.String("influxListenAddr", "", "TCP and UDP address to listen for InfluxDB line protocol data. Usually :8089 must be set. Doesn't work if empty. "+
 		"This flag isn't needed when ingesting data over HTTP - just send it to http://<victoriametrics>:8428/write . "+
 		"See also -influxListenAddr.useProxyProtocol")
@@ -72,6 +78,7 @@ var (
 
 var (
 	graphiteServer     *graphiteserver.Server
+	statsdServer       *statsdserver.Server
 	influxServer       *influxserver.Server
 	opentsdbServer     *opentsdbserver.Server
 	opentsdbhttpServer *opentsdbhttpserver.Server
@@ -92,6 +99,9 @@ func Init() {
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, *graphiteUseProxyProtocol, graphite.InsertHandler)
 	}
+	if len(*statsdListenAddr) > 0 {
+		statsdServer = statsdserver.MustStart(*statsdListenAddr, *statsdUseProxyProtocol, statsd.InsertHandler)
+	}
 	if len(*influxListenAddr) > 0 {
 		influxServer = influxserver.MustStart(*influxListenAddr, *influxUseProxyProtocol, influx.InsertHandlerForReader)
 	}
@@ -112,6 +122,9 @@ func Stop() {
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer.MustStop()
 	}
+	if len(*statsdListenAddr) > 0 {
+		statsdServer.MustStop()
+	}
 	if len(*influxListenAddr) > 0 {
 		influxServer.MustStop()
 	}
diff --git a/app/vminsert/statsd/request_handler.go b/app/vminsert/statsd/request_handler.go
new file mode 100644
index 000000000..b5901cfc3
--- /dev/null
+++ b/app/vminsert/statsd/request_handler.go
@@ -0,0 +1,54 @@
+package statsd
+
+import (
+	"io"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
+	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/statsd"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/statsd/stream"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	rowsInserted  = metrics.NewCounter(`vm_rows_inserted_total{type="statsd"}`)
+	rowsPerInsert = metrics.NewHistogram(`vm_rows_per_insert{type="statsd"}`)
+)
+
+// InsertHandler processes remote write for statsd protocol with tags.
+//
+// https://github.com/statsd/statsd/blob/master/docs/metric_types.md
+func InsertHandler(r io.Reader) error {
+	return stream.Parse(r, false, insertRows)
+}
+
+func insertRows(rows []parser.Row) error {
+	ctx := common.GetInsertCtx()
+	defer common.PutInsertCtx(ctx)
+
+	ctx.Reset(len(rows))
+	hasRelabeling := relabel.HasRelabeling()
+	for i := range rows {
+		r := &rows[i]
+		ctx.Labels = ctx.Labels[:0]
+		ctx.AddLabel("", r.Metric)
+		for j := range r.Tags {
+			tag := &r.Tags[j]
+			ctx.AddLabel(tag.Key, tag.Value)
+		}
+		if hasRelabeling {
+			ctx.ApplyRelabeling()
+		}
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
+			continue
+		}
+		ctx.SortLabelsIfNeeded()
+		if err := ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value); err != nil {
+			return err
+		}
+	}
+	rowsInserted.Add(len(rows))
+	rowsPerInsert.Update(float64(len(rows)))
+	return ctx.FlushBufs()
+}
diff --git a/app/vmui/Dockerfile-web b/app/vmui/Dockerfile-web
index f54841860..1ccad9a66 100644
--- a/app/vmui/Dockerfile-web
+++ b/app/vmui/Dockerfile-web
@@ -1,4 +1,4 @@
-FROM golang:1.22.2 as build-web-stage
+FROM golang:1.22.3 as build-web-stage
 COPY build /build
 
 WORKDIR /build
diff --git a/deployment/docker/Makefile b/deployment/docker/Makefile
index 9ae67e9ae..929166cd3 100644
--- a/deployment/docker/Makefile
+++ b/deployment/docker/Makefile
@@ -5,7 +5,7 @@ DOCKER_NAMESPACE ?= victoriametrics
 ROOT_IMAGE ?= alpine:3.19.1
 CERTS_IMAGE := alpine:3.19.1
 
-GO_BUILDER_IMAGE := golang:1.22.2-alpine
+GO_BUILDER_IMAGE := golang:1.22.3-alpine
 BUILDER_IMAGE := local/builder:2.0.0-$(shell echo $(GO_BUILDER_IMAGE) | tr :/ __)-1
 BASE_IMAGE := local/base:1.1.4-$(shell echo $(ROOT_IMAGE) | tr :/ __)-$(shell echo $(CERTS_IMAGE) | tr :/ __)
 DOCKER ?= docker
diff --git a/deployment/logs-benchmark/docker-compose-elk.yml b/deployment/logs-benchmark/docker-compose-elk.yml
index 26fb5d9bf..40b02e50e 100644
--- a/deployment/logs-benchmark/docker-compose-elk.yml
+++ b/deployment/logs-benchmark/docker-compose-elk.yml
@@ -18,7 +18,7 @@ services:
       - vlogs
 
   generator:
-    image: golang:1.22.2-alpine
+    image: golang:1.22.3-alpine
     restart: always
     working_dir: /go/src/app
     volumes:
diff --git a/deployment/logs-benchmark/docker-compose-loki.yml b/deployment/logs-benchmark/docker-compose-loki.yml
index b090641f3..c20a8125c 100644
--- a/deployment/logs-benchmark/docker-compose-loki.yml
+++ b/deployment/logs-benchmark/docker-compose-loki.yml
@@ -2,7 +2,7 @@ version: '3'
 
 services:
   generator:
-    image: golang:1.22.2-alpine
+    image: golang:1.22.3-alpine
     restart: always
     working_dir: /go/src/app
     volumes:
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 76fb4a3dd..d5a8da1d0 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -30,11 +30,14 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 ## tip
 
+* SECURITY: upgrade Go builder from Go1.22.2 to Go1.22.3. See [the list of issues addressed in Go1.22.3](https://github.com/golang/go/issues?q=milestone%3AGo1.22.3+label%3ACherryPickApproved).
+
 * FEATURE: [dashboards/single](https://grafana.com/grafana/dashboards/10229): support selecting of multiple instances on the dashboard. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5869) for details.
 * FEATURE: [dashboards/single](https://grafana.com/grafana/dashboards/10229): properly display version in the Stats row for the custom builds of VictoriaMetrics.
 * FEATURE: [dashboards/single](https://grafana.com/grafana/dashboards/10229): add `Network Usage` panel to `Resource Usage` row.
 * FEATURE: [dashboards/operator](https://grafana.com/grafana/dashboards/17869), [dashboards/backupmanager](https://grafana.com/grafana/dashboards/17798) and [dashboard/tenant-statistic](https://grafana.com/grafana/dashboards/16399): update dashboard to be compatible with Grafana 10+ version.
 * FEATURE: [dashboards/cluster](https://grafana.com/grafana/dashboards/11176): add new panel `Concurrent selects` to `vmstorage` row. The panel will show how many ongoing select queries are processed by vmstorage and should help to identify resource bottlenecks. See panel description for more details.
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add service discovery support for [Vultr](https://www.vultr.com/). See [these docs](https://docs.victoriametrics.com/sd_configs/#vultr_sd_configs) and [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6041).
 
 * BUGFIX: [vmui](https://docs.victoriametrics.com/#vmui): fix bug that prevents the first query trace from expanding on click event. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6186). The issue was introduced in [v1.100.0](https://docs.victoriametrics.com/changelog/#v11000) release.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent/): prevent potential panic during [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html) if more than one `--remoteWrite.streamAggr.dedupInterval` is configured. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6205).
@@ -49,6 +52,7 @@ Released at 2024-04-26
 * FEATURE: [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/): support disable or log [implicit conversions](https://docs.victoriametrics.com/metricsql/#implicit-query-conversions) for subquery with cmd-line flags `-search.disableImplicitConversion` and `-search.logImplicitConversion`. See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4338).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): support data replication additionally to sharding among remote storage systems if `-remoteWrite.shardByURLReplicas=N` command-line flag is set additionally to `-remoteWrite.shardByURL` command-line flag, where `N` is desired replication factor. This allows setting up data replication among failure domains when the replication factor is smaller than the number of failure domains. See [these docs](https://docs.victoriametrics.com/vmagent/#sharding-among-remote-storages) and [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6054).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): reduce CPU usage when [sharding among remote storage systems](https://docs.victoriametrics.com/vmagent/#sharding-among-remote-storages) is enabled.
+* FEATURE: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): reduce memory usage during stream aggregation if multiple aggregation configs are used for the same set of data.
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): support [DNS SRV](https://en.wikipedia.org/wiki/SRV_record) addresses in `-remoteWrite.url` command-line option and in scrape target urls. For example, `-remoteWrite.url=http://srv+victoria-metrics/api/v1/write` automatically resolves the `victoria-metrics` DNS SRV to a list of hostnames with TCP ports and then sends the collected metrics to these TCP addresses. See [these docs](https://docs.victoriametrics.com/vmagent/#srv-urls) and [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6053).
 * FEATURE: [stream aggregation](https://docs.victoriametrics.com/stream-aggregation/): allow skipping first N aggregation intervals via cmd-line flag `-streamAggr.ignoreFirstIntervals` for [Single-node VictoriaMetrics](https://docs.victoriametrics.com/) or `-remoteWrite.streamAggr.ignoreFirstIntervals` for [vmagent](https://docs.victoriametrics.com/vmagent/). See more details [here](https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start).
 * FEATURE: [vmauth](https://docs.victoriametrics.com/vmauth/): support automatic discovering and load balancing for TCP addresses behind DNS SRV addresses. These addresses can be put inside `url_prefix` urls in the form `http://srv+addr/path`, where the `addr` is the [DNS SRV](https://en.wikipedia.org/wiki/SRV_record) address, which is automatically resolved to hostnames with TCP ports. See [these docs](https://docs.victoriametrics.com/vmauth/#srv-urls) for details.
diff --git a/docs/README.md b/docs/README.md
index 486003f45..cdceb283e 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -89,6 +89,7 @@ VictoriaMetrics has the following prominent features:
   * [Prometheus exposition format](#how-to-import-data-in-prometheus-exposition-format).
   * [InfluxDB line protocol](#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf) over HTTP, TCP and UDP.
   * [Graphite plaintext protocol](#how-to-send-data-from-graphite-compatible-agents-such-as-statsd) with [tags](https://graphite.readthedocs.io/en/latest/tags.html#carbon).
+  * [Statsd plaintext protocol](#how-to-send-data-from-statsd-compatible-clients)
   * [OpenTSDB put message](#sending-data-via-telnet-put-protocol).
   * [HTTP OpenTSDB /api/put requests](#sending-opentsdb-data-via-http-apiput-requests).
   * [JSON line format](#how-to-import-data-in-json-line-format).
@@ -704,6 +705,45 @@ The `/api/v1/export` endpoint should return the following response:
 {"metric":{"__name__":"measurement_field2","tag1":"value1","tag2":"value2"},"values":[1.23],"timestamps":[1695902762311]}
 ```
 
+## How to send data from Statsd-compatible clients
+
+VictoriaMetrics supports extended statsd protocol with tags. Also it does not support sampling and metric types(it will be ignored).  
+Enable Statsd receiver in VictoriaMetrics by setting `-statsdListenAddr` command line flag. For instance,
+the following command will enable Statsd receiver in VictoriaMetrics on TCP and UDP port `8125`:
+
+```console
+/path/to/victoria-metrics-prod -statsdListenAddr=:8125
+```
+
+Example for writing data with Statsd plaintext protocol to local VictoriaMetrics using `nc`:
+
+```console
+echo "foo.bar:123|g|#foo:bar" | nc -N localhost 8125
+```
+
+Explicit setting of timestamps is not supported for statsd protocol. Timestamp is set to the current time when VictoriaMetrics or vmagent receives it.
+
+An arbitrary number of lines delimited by `\n` (aka newline char) can be sent in one go.
+After that the data may be read via [/api/v1/export](#how-to-export-data-in-json-line-format) endpoint:
+
+<div class="with-copy" markdown="1">
+
+```console
+curl -G 'http://localhost:8428/api/v1/export' -d 'match=foo.bar.baz'
+```
+
+</div>
+
+The `/api/v1/export` endpoint should return the following response:
+
+```json
+{"metric":{"__name__":"foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123],"timestamps":[1560277406000]}
+```
+
+Some examples of compatible statsd clients:
+- [statsd-instrument](https://github.com/Shopify/statsd-instrument)
+- [dogstatsd-ruby](https://github.com/DataDog/dogstatsd-ruby)
+- [go-statsd-client](https://github.com/cactus/go-statsd-client)
 ## How to send data from Graphite-compatible agents such as [StatsD](https://github.com/etsy/statsd)
 
 Enable Graphite receiver in VictoriaMetrics by setting `-graphiteListenAddr` command line flag. For instance,
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index c98f0420b..2de774ed5 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -97,6 +97,7 @@ VictoriaMetrics has the following prominent features:
   * [Prometheus exposition format](#how-to-import-data-in-prometheus-exposition-format).
   * [InfluxDB line protocol](#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf) over HTTP, TCP and UDP.
   * [Graphite plaintext protocol](#how-to-send-data-from-graphite-compatible-agents-such-as-statsd) with [tags](https://graphite.readthedocs.io/en/latest/tags.html#carbon).
+  * [Statsd plaintext protocol](#how-to-send-data-from-statsd-compatible-clients)
   * [OpenTSDB put message](#sending-data-via-telnet-put-protocol).
   * [HTTP OpenTSDB /api/put requests](#sending-opentsdb-data-via-http-apiput-requests).
   * [JSON line format](#how-to-import-data-in-json-line-format).
@@ -712,6 +713,45 @@ The `/api/v1/export` endpoint should return the following response:
 {"metric":{"__name__":"measurement_field2","tag1":"value1","tag2":"value2"},"values":[1.23],"timestamps":[1695902762311]}
 ```
 
+## How to send data from Statsd-compatible clients
+
+VictoriaMetrics supports extended statsd protocol with tags. Also it does not support sampling and metric types(it will be ignored).  
+Enable Statsd receiver in VictoriaMetrics by setting `-statsdListenAddr` command line flag. For instance,
+the following command will enable Statsd receiver in VictoriaMetrics on TCP and UDP port `8125`:
+
+```console
+/path/to/victoria-metrics-prod -statsdListenAddr=:8125
+```
+
+Example for writing data with Statsd plaintext protocol to local VictoriaMetrics using `nc`:
+
+```console
+echo "foo.bar:123|g|#foo:bar" | nc -N localhost 8125
+```
+
+Explicit setting of timestamps is not supported for statsd protocol. Timestamp is set to the current time when VictoriaMetrics or vmagent receives it.
+
+An arbitrary number of lines delimited by `\n` (aka newline char) can be sent in one go.
+After that the data may be read via [/api/v1/export](#how-to-export-data-in-json-line-format) endpoint:
+
+<div class="with-copy" markdown="1">
+
+```console
+curl -G 'http://localhost:8428/api/v1/export' -d 'match=foo.bar.baz'
+```
+
+</div>
+
+The `/api/v1/export` endpoint should return the following response:
+
+```json
+{"metric":{"__name__":"foo.bar.baz","tag1":"value1","tag2":"value2"},"values":[123],"timestamps":[1560277406000]}
+```
+
+Some examples of compatible statsd clients:
+- [statsd-instrument](https://github.com/Shopify/statsd-instrument)
+- [dogstatsd-ruby](https://github.com/DataDog/dogstatsd-ruby)
+- [go-statsd-client](https://github.com/cactus/go-statsd-client)
 ## How to send data from Graphite-compatible agents such as [StatsD](https://github.com/etsy/statsd)
 
 Enable Graphite receiver in VictoriaMetrics by setting `-graphiteListenAddr` command line flag. For instance,
diff --git a/docs/operator/CHANGELOG.md b/docs/operator/CHANGELOG.md
index 3766393fa..0b98308df 100644
--- a/docs/operator/CHANGELOG.md
+++ b/docs/operator/CHANGELOG.md
@@ -16,8 +16,12 @@ aliases:
 
 ## Next release
 
+- [vmagent](./api.md#vmagent): adds new fields into `streamAggrConfig`: `dedup_interval`, `ignore_old_samples`, `keep_metric_names`, `no_align_flush_to_interval`. It's only possible to use it with v1.100+ version of `vmagent`. See this [issue](https://github.com/VictoriaMetrics/operator/issues/936) for details.
+- [operator](./README.md): use `Patch` for `finalizers` set/unset operations. It must fix possible issues with `CRD` objects mutations. See this [issue](https://github.com/VictoriaMetrics/operator/issues/946) for details.
+- [operator](./README.md): adds `spec.pause` field to `VMAgent`, `VMAlert`, `VMAuth`, `VMCluster`, `VMAlertmanager` and `VMSingle`. It allows to suspend object reconcile by operator. See this [issue](https://github.com/VictoriaMetrics/operator/issues/943) for details. Thanks @just1900
 - [vmagent](./api.md#vmagent): set `status.selector` field. It allows correctly use `VPA` with `vmagent`. See this [issue](https://github.com/VictoriaMetrics/operator/issues/693) for details.
 - [prometheus-converter](./README.md): fixes bug with prometheus-operator ScrapeConfig converter. Only copy `spec` field for it. See this [issue](https://github.com/VictoriaMetrics/operator/issues/942) for details.
+- [vmscrapeconfig](./resources/vmscrapeconfig.md): `authorization` section in sd configs works properly with empty `type` field (default value for this field is `Bearer`). 
 
 <a name="v0.43.5"></a>
 
diff --git a/docs/operator/api.md b/docs/operator/api.md
index c470b0c9d..2636ef0c8 100644
--- a/docs/operator/api.md
+++ b/docs/operator/api.md
@@ -925,6 +925,7 @@ VMAgentSpec defines the desired state of VMAgent
 | useStrictSecurity | UseStrictSecurity enables strict security mode for component it restricts disk writes access uses non-root user out of the box drops not needed security permissions | *bool | false |
 | ingestOnlyMode | IngestOnlyMode switches vmagent into unmanaged mode it disables any config generation for scraping Currently it prevents vmagent from managing tls and auth options for remote write | bool | false |
 | license | License allows to configure license key to be used for enterprise features. Using license key is supported starting from VictoriaMetrics v1.94.0. See: https://docs.victoriametrics.com/enterprise.html | *[License](#license) | false |
+| paused | Paused If set to true all actions on the underlaying managed objects are not going to be performed, except for delete actions. | bool | false |
 
 [Back to TOC](#table-of-contents)
 
@@ -1132,11 +1133,16 @@ StreamAggrRule defines the rule in stream aggregation config
 | ----- | ----------- | ------ | -------- |
 | match | Match is a label selector (or list of label selectors) for filtering time series for the given selector.\n\nIf the match isn&#39;t set, then all the input time series are processed. | StringOrArray | false |
 | interval | Interval is the interval between aggregations. | string | true |
-| staleness_interval | StalenessInterval defines an interval after which the series state will be reset if no samples have been sent during it. | string | false |
+| no_align_flush_to_interval | NoAlighFlushToInterval disables aligning of flushes to multiples of Interval. By default flushes are aligned to Interval. | *bool | false |
 | flush_on_shutdown | FlushOnShutdown defines whether to flush the aggregation state on process termination or config reload. Is `false` by default. It is not recommended changing this setting, unless unfinished aggregations states are preferred to missing data points. | bool | false |
+| dedup_interval | DedupInterval is an optional interval for deduplication. | string | false |
+| staleness_interval | Staleness interval is interval after which the series state will be reset if no samples have been sent during it. The parameter is only relevant for outputs: total, total_prometheus, increase, increase_prometheus and histogram_bucket. | string | false |
 | outputs | Outputs is a list of output aggregate functions to produce.\n\nThe following names are allowed:\n\n- total - aggregates input counters - increase - counts the increase over input counters - count_series - counts the input series - count_samples - counts the input samples - sum_samples - sums the input samples - last - the last biggest sample value - min - the minimum sample value - max - the maximum sample value - avg - the average value across all the samples - stddev - standard deviation across all the samples - stdvar - standard variance across all the samples - histogram_bucket - creates VictoriaMetrics histogram for input samples - quantiles(phi1, ..., phiN) - quantiles&#39; estimation for phi in the range [0..1]\n\nThe output time series will have the following names:\n\n  input_name:aggr_&lt;interval&gt;_&lt;output&gt; | []string | true |
+| keep_metric_names | KeepMetricNames instructs to leave metric names as is for the output time series without adding any suffix. | *bool | false |
+| ignore_old_samples | IgnoreOldSamples instructs to ignore samples with old timestamps outside the current aggregation interval. | *bool | false |
 | by | By is an optional list of labels for grouping input series.\n\nSee also Without.\n\nIf neither By nor Without are set, then the Outputs are calculated individually per each input time series. | []string | false |
 | without | Without is an optional list of labels, which must be excluded when grouping input series.\n\nSee also By.\n\nIf neither By nor Without are set, then the Outputs are calculated individually per each input time series. | []string | false |
+| drop_input_labels | DropInputLabels is an optional list with labels, which must be dropped before further processing of input samples.\n\nLabels are dropped before de-duplication and aggregation. | *[]string | false |
 | input_relabel_configs | InputRelabelConfigs is an optional relabeling rules, which are applied on the input before aggregation. | [][RelabelConfig](#relabelconfig) | false |
 | output_relabel_configs | OutputRelabelConfigs is an optional relabeling rules, which are applied on the aggregated output before being sent to remote storage. | [][RelabelConfig](#relabelconfig) | false |
 
@@ -1297,6 +1303,7 @@ VMAlertSpec defines the desired state of VMAlert
 | readinessGates | ReadinessGates defines pod readiness gates | []v1.PodReadinessGate | false |
 | useStrictSecurity | UseStrictSecurity enables strict security mode for component it restricts disk writes access uses non-root user out of the box drops not needed security permissions | *bool | false |
 | license | License allows to configure license key to be used for enterprise features. Using license key is supported starting from VictoriaMetrics v1.94.0. See: https://docs.victoriametrics.com/enterprise.html | *[License](#license) | false |
+| paused | Paused If set to true all actions on the underlaying managed objects are not going to be performed, except for delete actions. | bool | false |
 
 [Back to TOC](#table-of-contents)
 
@@ -1391,6 +1398,7 @@ VMSingleSpec defines the desired state of VMSingle
 | readinessGates | ReadinessGates defines pod readiness gates | []v1.PodReadinessGate | false |
 | streamAggrConfig | StreamAggrConfig defines stream aggregation configuration for VMSingle | *[StreamAggrConfig](#streamaggrconfig) | false |
 | useStrictSecurity | UseStrictSecurity enables strict security mode for component it restricts disk writes access uses non-root user out of the box drops not needed security permissions | *bool | false |
+| paused | Paused If set to true all actions on the underlaying managed objects are not going to be performed, except for delete actions. | bool | false |
 
 [Back to TOC](#table-of-contents)
 
@@ -1877,6 +1885,7 @@ VMClusterSpec defines the desired state of VMCluster
 | vmselect |  | *[VMSelect](#vmselect) | false |
 | vminsert |  | *[VMInsert](#vminsert) | false |
 | vmstorage |  | *[VMStorage](#vmstorage) | false |
+| paused | Paused If set to true all actions on the underlaying managed objects are not going to be performed, except for delete actions. | bool | false |
 | useStrictSecurity | UseStrictSecurity enables strict security mode for component it restricts disk writes access uses non-root user out of the box drops not needed security permissions | *bool | false |
 
 [Back to TOC](#table-of-contents)
@@ -2338,6 +2347,7 @@ VMAuthSpec defines the desired state of VMAuth
 | useStrictSecurity | UseStrictSecurity enables strict security mode for component it restricts disk writes access uses non-root user out of the box drops not needed security permissions | *bool | false |
 | license | License allows to configure license key to be used for enterprise features. Using license key is supported starting from VictoriaMetrics v1.94.0. See: https://docs.victoriametrics.com/enterprise.html | *[License](#license) | false |
 | configSecret | ConfigSecret is the name of a Kubernetes Secret in the same namespace as the VMAuth object, which contains auth configuration for vmauth, configuration must be inside secret key: config.yaml. It must be created and managed manually. If it&#39;s defined, configuration for vmauth becomes unmanaged and operator&#39;ll not create any related secrets/config-reloaders | string | false |
+| paused | Paused If set to true all actions on the underlaying managed objects are not going to be performed, except for delete actions. | bool | false |
 
 [Back to TOC](#table-of-contents)
 
diff --git a/docs/operator/vars.md b/docs/operator/vars.md
index 8598de344..c73b9a8f5 100644
--- a/docs/operator/vars.md
+++ b/docs/operator/vars.md
@@ -10,7 +10,7 @@ menu:
 
 <!-- this doc autogenerated - don't edit it manually -->
 # Auto Generated vars for package config 
- updated at Sat Apr 27 19:49:28 UTC 2024 
+ updated at Tue May  7 16:27:47 UTC 2024 
 
 
 | varible name | variable default value | variable required | variable description |
diff --git a/docs/scrape_config_examples.md b/docs/scrape_config_examples.md
index 26f476b8d..d292722fb 100644
--- a/docs/scrape_config_examples.md
+++ b/docs/scrape_config_examples.md
@@ -250,7 +250,7 @@ scrape_configs:
     # If node-exporter containers have another name in your Kubernetes cluster,
     # then adjust the regex value accordingly.
     #
-  - soruce_labels: [__meta_kubernetes_pod_container_name]
+  - source_labels: [__meta_kubernetes_pod_container_name]
     regex: node-exporter
     action: keep
 
diff --git a/docs/sd_configs.md b/docs/sd_configs.md
index f2357d020..eda9a4a8c 100644
--- a/docs/sd_configs.md
+++ b/docs/sd_configs.md
@@ -35,6 +35,7 @@ supports the following Prometheus-compatible service discovery options for Prome
 * `nomad_sd_configs` is for discovering and scraping targets registered in [HashiCorp Nomad](https://www.nomadproject.io/). See [these docs](#nomad_sd_configs).
 * `openstack_sd_configs` is for discovering and scraping OpenStack targets. See [these docs](#openstack_sd_configs).
 * `static_configs` is for scraping statically defined targets. See [these docs](#static_configs).
+* `vultr_sd_configs` is for discovering and scraping [Vultr](https://www.vultr.com/) targets. See [these docs](#vultr_sd_configs).
 * `yandexcloud_sd_configs` is for discovering and scraping [Yandex Cloud](https://cloud.yandex.com/en/) targets. See [these docs](#yandexcloud_sd_configs).
 
 Note that the `refresh_interval` option isn't supported for these scrape configs. Use the corresponding `-promscrape.*CheckInterval`
@@ -1498,6 +1499,79 @@ scrape_configs:
 
 See [these examples](https://docs.victoriametrics.com/scrape_config_examples/#static-configs) on how to configure scraping for static targets.
 
+## vultr_sd_configs
+Vultr SD configuration discovers scrape targets from [Vultr](https://www.vultr.com/) Instances.
+
+Configuration example:
+
+```yaml
+scrape_configs:
+- job_name: vultr
+  vultr_sd_configs:
+
+    # bearer_token is a Bearer token to send in every HTTP API request during service discovery (mandatory).
+    # See: https://my.vultr.com/settings/#settingsapi
+  - bearer_token: "..."
+
+    # Vultr provides query arguments to filter instances.
+    # See: https://www.vultr.com/api/#tag/instances
+
+    # label is an optional query arguments to filter instances by label.
+    #
+    # label: "..."
+
+    # main_ip is an optional query arguments to filter instances by main ip address.
+    #
+    # main_ip: "..."
+
+    # region is an optional query arguments to filter instances by region id.
+    #
+    # region: "..."
+
+    # firewall_group_id is an optional query arguments to filter instances by firewall group id.
+    #
+    # firewall_group_id: "..."
+
+    # hostname is an optional query arguments to filter instances by hostname.
+    #
+    # hostname: "..."
+
+    # port is an optional port to scrape metrics from.
+    # By default, port 80 is used.
+    #
+    # port: ...
+
+    # Additional HTTP API client options can be specified here.
+    # See https://docs.victoriametrics.com/sd_configs.html#http-api-client-options
+
+
+```
+
+Each discovered target has an [`__address__`](https://docs.victoriametrics.com/relabeling.html#how-to-modify-scrape-urls-in-targets) label set
+to `<FQDN>:<port>`, where FQDN is discovered instance address and `<port>` is the port from the `vultr_sd_configs` (default port is `80`).
+
+The following meta labels are available on discovered targets during [relabeling](https://docs.victoriametrics.com/vmagent.html#relabeling):
+
+* `__meta_vultr_instance_id`: A unique ID for the VPS Instance.
+* `__meta_vultr_instance_label`: The user-supplied label for this instance.
+* `__meta_vultr_instance_os`: The [Operating System name](https://www.vultr.com/api/#operation/list-os).
+* `__meta_vultr_instance_os_id`: The [Operating System id](https://www.vultr.com/api/#operation/list-os) used by this instance.
+* `__meta_vultr_instance_region`: The [Region id](https://www.vultr.com/api/#operation/list-regions) where the Instance is located.
+* `__meta_vultr_instance_plan`: A unique ID for the Plan.
+* `__meta_vultr_instance_main_ip`: The main IPv4 address.
+* `__meta_vultr_instance_internal_ip`: The internal IP used by this instance, if set. Only relevant when a VPC is attached.
+* `__meta_vultr_instance_main_ipv6`: The main IPv6 network address.
+* `__meta_vultr_instance_hostname`: The hostname for this instance.
+* `__meta_vultr_instance_server_status`: The server health status, which could be `none`, `locked`, `installingbooting`, `ok`.
+* `__meta_vultr_instance_vcpu_count`: Number of vCPUs.
+* `__meta_vultr_instance_ram_mb`: The amount of RAM in MB.
+* `__meta_vultr_instance_allowed_bandwidth_gb`: Monthly bandwidth quota in GB.
+* `__meta_vultr_instance_disk_gb`: The size of the disk in GB.
+* `__meta_vultr_instance_features`: "auto_backups", "ipv6", "ddos_protection".
+* `__meta_vultr_instance_tags`: Tags to apply to the instance.
+
+The list of discovered Vultr targets is refreshed at the interval, which can be configured via `-promscrape.vultrSDCheckInterval` command-line flag, default: 30s.
+
 ## yandexcloud_sd_configs
 
 [Yandex Cloud](https://cloud.yandex.com/en/) SD configurations allow retrieving scrape targets from accessible folders.
diff --git a/docs/stream-aggregation.md b/docs/stream-aggregation.md
index be2c2b5ba..820ca4eef 100644
--- a/docs/stream-aggregation.md
+++ b/docs/stream-aggregation.md
@@ -133,7 +133,7 @@ Stream aggregation can be used in the following cases:
 
 ### Statsd alternative
 
-Stream aggregation can be used as [statsd](https://github.com/statsd/statsd) alternative in the following cases:
+Stream aggregation can be used as [statsd](https://github.com/statsd/statsd) drop-in replacement in the following cases:
 
 * [Counting input samples](#counting-input-samples)
 * [Summing input metrics](#summing-input-metrics)
@@ -141,9 +141,6 @@ Stream aggregation can be used as [statsd](https://github.com/statsd/statsd) alt
 * [Histograms over input metrics](#histograms-over-input-metrics)
 * [Aggregating histograms](#aggregating-histograms)
 
-Currently, streaming aggregation is available only for [supported data ingestion protocols](https://docs.victoriametrics.com/#how-to-import-time-series-data)
-and not available for [Statsd metrics format](https://github.com/statsd/statsd/blob/master/docs/metric_types.md).
-
 ### Recording rules alternative
 
 Sometimes [alerting queries](https://docs.victoriametrics.com/vmalert/#alerting-rules) may require non-trivial amounts of CPU, RAM,
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 227c6e45d..7e368f437 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -107,6 +107,7 @@ additionally to pull-based Prometheus-compatible targets' scraping:
 * DataDog "submit metrics" API. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#how-to-send-data-from-datadog-agent).
 * InfluxDB line protocol via `http://<vmagent>:8429/write`. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#how-to-send-data-from-influxdb-compatible-agents-such-as-telegraf).
 * Graphite plaintext protocol if `-graphiteListenAddr` command-line flag is set. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#how-to-send-data-from-graphite-compatible-agents-such-as-statsd).
+* Statsd plaintext protocol if `-statsdListenAddr` command-line flag is set. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#how-to-send-data-from-statsd-compatible-clients).
 * OpenTelemetry http API. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#sending-data-via-opentelemetry).
 * NewRelic API. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#how-to-send-data-from-newrelic-agent).
 * OpenTSDB telnet and http protocols if `-opentsdbListenAddr` command-line flag is set. See [these docs](https://docs.victoriametrics.com/single-server-victoriametrics/#how-to-send-data-from-opentsdb-compatible-agents).
@@ -1707,6 +1708,10 @@ See the docs at https://docs.victoriametrics.com/vmagent/ .
      Whether to use proxy protocol for connections accepted at -graphiteListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -graphiteTrimTimestamp duration
      Trim timestamps for Graphite data to this duration. Minimum practical duration is 1s. Higher duration (i.e. 1m) may be used for reducing disk space usage for timestamp data (default 1s)
+  -statsdListenAddr string
+     TCP and UDP address to listen for Statsd plaintext data. Usually :8125 must be set. Doesn't work if empty. See also -statsdListenAddr.useProxyProtocol
+  -statsdListenAddr.useProxyProtocol
+     Whether to use proxy protocol for connections accepted at -statsdListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -http.connTimeout duration
      Incoming connections to -httpListenAddr are closed after the configured timeout. This may help evenly spreading load among a cluster of services behind TCP-level load balancer. Zero value disables closing of incoming connections (default 2m0s)
   -http.disableResponseCompression
@@ -1977,6 +1982,8 @@ See the docs at https://docs.victoriametrics.com/vmagent/ .
      Whether to suppress scrape errors logging. The last error for each target is always available at '/targets' page even if scrape errors logging is suppressed. See also -promscrape.suppressScrapeErrorsDelay
   -promscrape.suppressScrapeErrorsDelay duration
      The delay for suppressing repeated scrape errors logging per each scrape targets. This may be used for reducing the number of log lines related to scrape errors. See also -promscrape.suppressScrapeErrors
+  -promscrape.vultrSDCheckInterval duration
+     Interval for checking for changes in Vultr. This works only if vultr_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#vultr_sd_configs for details  (default 30s)
   -promscrape.yandexcloudSDCheckInterval duration
      Interval for checking for changes in Yandex Cloud API. This works only if yandexcloud_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs/#yandexcloud_sd_configs for details (default 30s)
   -pushmetrics.disableCompression
diff --git a/lib/ingestserver/statsd/server.go b/lib/ingestserver/statsd/server.go
new file mode 100644
index 000000000..533d2ed1b
--- /dev/null
+++ b/lib/ingestserver/statsd/server.go
@@ -0,0 +1,173 @@
+package statsd
+
+import (
+	"errors"
+	"io"
+	"net"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	writeRequestsTCP = metrics.NewCounter(`vm_ingestserver_requests_total{type="statsd", name="write", net="tcp"}`)
+	writeErrorsTCP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="statsd", name="write", net="tcp"}`)
+
+	writeRequestsUDP = metrics.NewCounter(`vm_ingestserver_requests_total{type="statsd", name="write", net="udp"}`)
+	writeErrorsUDP   = metrics.NewCounter(`vm_ingestserver_request_errors_total{type="statsd", name="write", net="udp"}`)
+)
+
+// Server accepts Statsd plaintext lines over TCP and UDP.
+type Server struct {
+	addr  string
+	lnTCP net.Listener
+	lnUDP net.PacketConn
+	wg    sync.WaitGroup
+	cm    ingestserver.ConnsMap
+}
+
+// MustStart starts statsd server on the given addr.
+//
+// The incoming connections are processed with insertHandler.
+//
+// If useProxyProtocol is set to true, then the incoming connections are accepted via proxy protocol.
+// See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
+//
+// MustStop must be called on the returned server when it is no longer needed.
+func MustStart(addr string, useProxyProtocol bool, insertHandler func(r io.Reader) error) *Server {
+	logger.Infof("starting TCP Statsd server at %q", addr)
+	lnTCP, err := netutil.NewTCPListener("statsd", addr, useProxyProtocol, nil)
+	if err != nil {
+		logger.Fatalf("cannot start TCP Statsd server at %q: %s", addr, err)
+	}
+
+	logger.Infof("starting UDP Statsd server at %q", addr)
+	lnUDP, err := net.ListenPacket(netutil.GetUDPNetwork(), addr)
+	if err != nil {
+		logger.Fatalf("cannot start UDP Statsd server at %q: %s", addr, err)
+	}
+
+	s := &Server{
+		addr:  addr,
+		lnTCP: lnTCP,
+		lnUDP: lnUDP,
+	}
+	s.cm.Init("statsd")
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		s.serveTCP(insertHandler)
+		logger.Infof("stopped TCP Statsd server at %q", addr)
+	}()
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		s.serveUDP(insertHandler)
+		logger.Infof("stopped UDP Statsd server at %q", addr)
+	}()
+	return s
+}
+
+// MustStop stops the server.
+func (s *Server) MustStop() {
+	logger.Infof("stopping TCP Statsd server at %q...", s.addr)
+	if err := s.lnTCP.Close(); err != nil {
+		logger.Errorf("cannot close TCP Statsd server: %s", err)
+	}
+	logger.Infof("stopping UDP Statsd server at %q...", s.addr)
+	if err := s.lnUDP.Close(); err != nil {
+		logger.Errorf("cannot close UDP Statsd server: %s", err)
+	}
+	s.cm.CloseAll(0)
+	s.wg.Wait()
+	logger.Infof("TCP and UDP Statsd servers at %q have been stopped", s.addr)
+}
+
+func (s *Server) serveTCP(insertHandler func(r io.Reader) error) {
+	var wg sync.WaitGroup
+	for {
+		c, err := s.lnTCP.Accept()
+		if err != nil {
+			var ne net.Error
+			if errors.As(err, &ne) {
+				if ne.Temporary() {
+					logger.Errorf("statsd: temporary error when listening for TCP addr %q: %s", s.lnTCP.Addr(), err)
+					time.Sleep(time.Second)
+					continue
+				}
+				if strings.Contains(err.Error(), "use of closed network connection") {
+					break
+				}
+				logger.Fatalf("unrecoverable error when accepting TCP Statsd connections: %s", err)
+			}
+			logger.Fatalf("unexpected error when accepting TCP Statsd connections: %s", err)
+		}
+		if !s.cm.Add(c) {
+			_ = c.Close()
+			break
+		}
+		wg.Add(1)
+		go func() {
+			defer func() {
+				s.cm.Delete(c)
+				_ = c.Close()
+				wg.Done()
+			}()
+			writeRequestsTCP.Inc()
+			if err := insertHandler(c); err != nil {
+				writeErrorsTCP.Inc()
+				logger.Errorf("error in TCP Statsd conn %q<->%q: %s", c.LocalAddr(), c.RemoteAddr(), err)
+			}
+		}()
+	}
+	wg.Wait()
+}
+
+func (s *Server) serveUDP(insertHandler func(r io.Reader) error) {
+	gomaxprocs := cgroup.AvailableCPUs()
+	var wg sync.WaitGroup
+	for i := 0; i < gomaxprocs; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var bb bytesutil.ByteBuffer
+			bb.B = bytesutil.ResizeNoCopyNoOverallocate(bb.B, 64*1024)
+			for {
+				bb.Reset()
+				bb.B = bb.B[:cap(bb.B)]
+				n, addr, err := s.lnUDP.ReadFrom(bb.B)
+				if err != nil {
+					writeErrorsUDP.Inc()
+					var ne net.Error
+					if errors.As(err, &ne) {
+						if ne.Temporary() {
+							logger.Errorf("statsd: temporary error when listening for UDP addr %q: %s", s.lnUDP.LocalAddr(), err)
+							time.Sleep(time.Second)
+							continue
+						}
+						if strings.Contains(err.Error(), "use of closed network connection") {
+							break
+						}
+					}
+					logger.Errorf("cannot read Statsd UDP data: %s", err)
+					continue
+				}
+				bb.B = bb.B[:n]
+				writeRequestsUDP.Inc()
+				if err := insertHandler(bb.NewReader()); err != nil {
+					writeErrorsUDP.Inc()
+					logger.Errorf("error in UDP Statsd conn %q<->%q: %s", s.lnUDP.LocalAddr(), addr, err)
+					continue
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index 5516b60f1..b0ea9603f 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -36,6 +36,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kuma"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/nomad"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/openstack"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/vultr"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/yandexcloud"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
@@ -308,6 +309,7 @@ type ScrapeConfig struct {
 	NomadSDConfigs        []nomad.SDConfig        `yaml:"nomad_sd_configs,omitempty"`
 	OpenStackSDConfigs    []openstack.SDConfig    `yaml:"openstack_sd_configs,omitempty"`
 	StaticConfigs         []StaticConfig          `yaml:"static_configs,omitempty"`
+	VultrConfigs          []vultr.SDConfig        `yaml:"vultr_configs,omitempty"`
 	YandexCloudSDConfigs  []yandexcloud.SDConfig  `yaml:"yandexcloud_sd_configs,omitempty"`
 
 	// These options are supported only by lib/promscrape.
@@ -388,6 +390,9 @@ func (sc *ScrapeConfig) mustStop() {
 	for i := range sc.OpenStackSDConfigs {
 		sc.OpenStackSDConfigs[i].MustStop()
 	}
+	for i := range sc.VultrConfigs {
+		sc.VultrConfigs[i].MustStop()
+	}
 }
 
 // FileSDConfig represents file-based service discovery config.
@@ -745,6 +750,16 @@ func (cfg *Config) getOpenStackSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
 	return cfg.getScrapeWorkGeneric(visitConfigs, "openstack_sd_config", prev)
 }
 
+// getVultrSDScrapeWork returns `vultr_sd_configs` ScrapeWork from cfg.
+func (cfg *Config) getVultrSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
+	visitConfigs := func(sc *ScrapeConfig, visitor func(sdc targetLabelsGetter)) {
+		for i := range sc.VultrConfigs {
+			visitor(&sc.VultrConfigs[i])
+		}
+	}
+	return cfg.getScrapeWorkGeneric(visitConfigs, "vultr_sd_config", prev)
+}
+
 // getYandexCloudSDScrapeWork returns `yandexcloud_sd_configs` ScrapeWork from cfg.
 func (cfg *Config) getYandexCloudSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
 	visitConfigs := func(sc *ScrapeConfig, visitor func(sdc targetLabelsGetter)) {
diff --git a/lib/promscrape/discovery/vultr/api.go b/lib/promscrape/discovery/vultr/api.go
new file mode 100644
index 000000000..86e82df3a
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/api.go
@@ -0,0 +1,76 @@
+package vultr
+
+import (
+	"fmt"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+)
+
+// apiConfig contains config for API server.
+type apiConfig struct {
+	c    *discoveryutils.Client
+	port int
+
+	listParams
+}
+
+// listParams is the query params of vultr ListInstance API.
+type listParams struct {
+	// paging params are not exposed to user, they will be filled
+	// dynamically during request. See `getInstances`.
+	// perPage int
+	// cursor string
+
+	// API query params for filtering.
+	label           string
+	mainIP          string
+	region          string
+	firewallGroupID string
+	hostname        string
+}
+
+// getAPIConfig get or create API config from configMap.
+func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	v, err := configMap.Get(sdc, func() (interface{}, error) { return newAPIConfig(sdc, baseDir) })
+	if err != nil {
+		return nil, err
+	}
+	return v.(*apiConfig), nil
+}
+
+// newAPIConfig create API Config.
+func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	port := sdc.Port
+	if port == 0 {
+		port = 80
+	}
+
+	// See: https://www.vultr.com/api/
+	apiServer := "https://api.vultr.com"
+
+	ac, err := sdc.HTTPClientConfig.NewConfig(baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse auth config: %w", err)
+	}
+	proxyAC, err := sdc.ProxyClientConfig.NewConfig(baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse proxy auth config: %w", err)
+	}
+
+	c, err := discoveryutils.NewClient(apiServer, ac, sdc.ProxyURL, proxyAC, &sdc.HTTPClientConfig)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create client for %q: %w", apiServer, err)
+	}
+	cfg := &apiConfig{
+		c:    c,
+		port: port,
+		listParams: listParams{
+			label:           sdc.Label,
+			mainIP:          sdc.MainIP,
+			region:          sdc.Region,
+			firewallGroupID: sdc.FirewallGroupID,
+			hostname:        sdc.Hostname,
+		},
+	}
+	return cfg, nil
+}
diff --git a/lib/promscrape/discovery/vultr/instance.go b/lib/promscrape/discovery/vultr/instance.go
new file mode 100644
index 000000000..e510b7353
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/instance.go
@@ -0,0 +1,109 @@
+package vultr
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/url"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+// ListInstanceResponse is the response structure of Vultr ListInstance API.
+type ListInstanceResponse struct {
+	Instances []Instance `json:"instances"`
+	Meta      *Meta      `json:"Meta"`
+}
+
+// Instance represents Vultr Instance (VPS).
+// See: https://github.com/vultr/govultr/blob/5125e02e715ae6eb3ce854f0e7116c7ce545a710/instance.go#L81
+type Instance struct {
+	ID               string   `json:"id"`
+	Os               string   `json:"os"`
+	RAM              int      `json:"ram"`
+	Disk             int      `json:"disk"`
+	MainIP           string   `json:"main_ip"`
+	VCPUCount        int      `json:"vcpu_count"`
+	Region           string   `json:"region"`
+	ServerStatus     string   `json:"server_status"`
+	AllowedBandwidth int      `json:"allowed_bandwidth"`
+	V6MainIP         string   `json:"v6_main_ip"`
+	Hostname         string   `json:"hostname"`
+	Label            string   `json:"label"`
+	InternalIP       string   `json:"internal_ip"`
+	OsID             int      `json:"os_id"`
+	Features         []string `json:"features"`
+	Plan             string   `json:"plan"`
+	Tags             []string `json:"tags"`
+
+	// The following fields are defined in the response but are not used during service discovery.
+	//DefaultPassword string `json:"default_password,omitempty"`
+	//DateCreated     string `json:"date_created"`
+	//Status          string `json:"status"`
+	//PowerStatus     string `json:"power_status"`
+	//NetmaskV4       string `json:"netmask_v4"`
+	//GatewayV4       string `json:"gateway_v4"`
+	//V6Network       string `json:"v6_network"`
+	//V6NetworkSize   int    `json:"v6_network_size"`
+	//// Deprecated: Tag should no longer be used. Instead, use Tags.
+	//Tag             string `json:"tag"`
+	//KVM             string `json:"kvm"`
+	//AppID           int    `json:"app_id"`
+	//ImageID         string `json:"image_id"`
+	//FirewallGroupID string `json:"firewall_group_id"`
+	//UserScheme      string `json:"user_scheme"`
+}
+
+// Meta represents the available pagination information
+type Meta struct {
+	Total int `json:"total"`
+	Links *Links
+}
+
+// Links represent the next/previous cursor in your pagination calls
+type Links struct {
+	Next string `json:"next"`
+	Prev string `json:"prev"`
+}
+
+// getInstances retrieve instance from Vultr HTTP API.
+func getInstances(cfg *apiConfig) ([]Instance, error) {
+	var instances []Instance
+
+	// prepare GET params
+	params := url.Values{}
+	params.Set("per_page", "100")
+	params.Set("label", cfg.label)
+	params.Set("main_ip", cfg.mainIP)
+	params.Set("region", cfg.region)
+	params.Set("firewall_group_id", cfg.firewallGroupID)
+	params.Set("hostname", cfg.hostname)
+
+	// send request to vultr API
+	for {
+		// See: https://www.vultr.com/api/#tag/instances/operation/list-instances
+		path := fmt.Sprintf("/v2/instances?%s", params.Encode())
+		resp, err := cfg.c.GetAPIResponse(path)
+		if err != nil {
+			logger.Errorf("get response from vultr failed, path:%s, err: %v", path, err)
+			return nil, err
+		}
+
+		var listInstanceResp ListInstanceResponse
+		if err = json.Unmarshal(resp, &listInstanceResp); err != nil {
+			logger.Errorf("unmarshal response from vultr failed, err: %v", err)
+			return nil, err
+		}
+
+		instances = append(instances, listInstanceResp.Instances...)
+
+		if listInstanceResp.Meta != nil && listInstanceResp.Meta.Links != nil && listInstanceResp.Meta.Links.Next != "" {
+			// if `next page` is available, set the cursor param and request again.
+			params.Set("cursor", listInstanceResp.Meta.Links.Next)
+		} else {
+			// otherwise exit the loop
+			break
+		}
+	}
+
+	return instances, nil
+}
diff --git a/lib/promscrape/discovery/vultr/vultr.go b/lib/promscrape/discovery/vultr/vultr.go
new file mode 100644
index 000000000..d8999b2bb
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/vultr.go
@@ -0,0 +1,105 @@
+package vultr
+
+import (
+	"flag"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
+)
+
+const (
+	separator = ","
+)
+
+// SDCheckInterval defines interval for docker targets refresh.
+var SDCheckInterval = flag.Duration("promscrape.vultrSDCheckInterval", 30*time.Second, "Interval for checking for changes in Vultr. "+
+	"This works only if vultr_sd_configs is configured in '-promscrape.config' file. "+
+	"See https://docs.victoriametrics.com/sd_configs.html#vultr_sd_configs for details")
+
+// SDConfig represents service discovery config for Vultr.
+// See: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#vultr_sd_config
+// Additional query params are supported, while Prometheus only supports `Port` and HTTP auth.
+type SDConfig struct {
+	// API query params for filtering. All of them are optional.
+	// See: https://www.vultr.com/api/#tag/instances/operation/list-instances
+	Label           string `yaml:"label,omitempty"`
+	MainIP          string `yaml:"main_ip,omitempty"`
+	Region          string `yaml:"region,omitempty"`
+	FirewallGroupID string `yaml:"firewall_group_id,omitempty"`
+	Hostname        string `yaml:"hostname,omitempty"`
+
+	// The port to scrape metrics from. Default 80.
+	Port int `yaml:"port"`
+
+	// General HTTP / Auth configs.
+	HTTPClientConfig  promauth.HTTPClientConfig  `yaml:",inline"`
+	ProxyURL          *proxy.URL                 `yaml:"proxy_url,omitempty"`
+	ProxyClientConfig promauth.ProxyClientConfig `yaml:",inline"`
+
+	// refresh_interval is obtained from `-promscrape.vultrSDCheckInterval` command-line option.
+}
+
+var configMap = discoveryutils.NewConfigMap()
+
+// GetLabels returns gce labels according to sdc.
+func (sdc *SDConfig) GetLabels(baseDir string) ([]*promutils.Labels, error) {
+	ac, err := getAPIConfig(sdc, baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot get API config: %w", err)
+	}
+	instances, err := getInstances(ac)
+	if err != nil {
+		return nil, err
+	}
+	return getInstanceLabels(instances, ac.port), nil
+}
+
+// MustStop stops further usage for sdc.
+func (sdc *SDConfig) MustStop() {
+	configMap.Delete(sdc)
+}
+
+// getInstanceLabels returns labels for vultr instances obtained from the given cfg
+func getInstanceLabels(instances []Instance, port int) []*promutils.Labels {
+	ms := make([]*promutils.Labels, 0, len(instances))
+
+	for _, instance := range instances {
+		m := promutils.NewLabels(18)
+		m.Add("__address__", discoveryutils.JoinHostPort(instance.MainIP, port))
+		m.Add("__meta_vultr_instance_id", instance.ID)
+		m.Add("__meta_vultr_instance_label", instance.Label)
+		m.Add("__meta_vultr_instance_os", instance.Os)
+		m.Add("__meta_vultr_instance_os_id", strconv.Itoa(instance.OsID))
+		m.Add("__meta_vultr_instance_region", instance.Region)
+		m.Add("__meta_vultr_instance_plan", instance.Plan)
+		m.Add("__meta_vultr_instance_main_ip", instance.MainIP)
+		m.Add("__meta_vultr_instance_internal_ip", instance.InternalIP)
+		m.Add("__meta_vultr_instance_main_ipv6", instance.V6MainIP)
+		m.Add("__meta_vultr_instance_hostname", instance.Hostname)
+		m.Add("__meta_vultr_instance_server_status", instance.ServerStatus)
+		m.Add("__meta_vultr_instance_vcpu_count", strconv.Itoa(instance.VCPUCount))
+		m.Add("__meta_vultr_instance_ram_mb", strconv.Itoa(instance.RAM))
+		m.Add("__meta_vultr_instance_allowed_bandwidth_gb", strconv.Itoa(instance.AllowedBandwidth))
+		m.Add("__meta_vultr_instance_disk_gb", strconv.Itoa(instance.Disk))
+
+		// We surround the separated list with the separator as well. This way regular expressions
+		// in relabeling rules don't have to consider feature positions.
+		if len(instance.Features) > 0 {
+			features := separator + strings.Join(instance.Features, separator) + separator
+			m.Add("__meta_vultr_instance_features", features)
+		}
+
+		if len(instance.Tags) > 0 {
+			tags := separator + strings.Join(instance.Tags, separator) + separator
+			m.Add("__meta_vultr_instance_tags", tags)
+		}
+		ms = append(ms, m)
+	}
+	return ms
+}
diff --git a/lib/promscrape/scraper.go b/lib/promscrape/scraper.go
index 797503298..d60511b8c 100644
--- a/lib/promscrape/scraper.go
+++ b/lib/promscrape/scraper.go
@@ -30,6 +30,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kuma"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/nomad"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/openstack"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/vultr"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/yandexcloud"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 	"github.com/VictoriaMetrics/metrics"
@@ -140,6 +141,7 @@ func runScraper(configFile string, pushData func(at *auth.Token, wr *prompbmarsh
 	scs.add("kuma_sd_configs", *kuma.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getKumaSDScrapeWork(swsPrev) })
 	scs.add("nomad_sd_configs", *nomad.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getNomadSDScrapeWork(swsPrev) })
 	scs.add("openstack_sd_configs", *openstack.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getOpenStackSDScrapeWork(swsPrev) })
+	scs.add("vultr_sd_configs", *vultr.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getVultrSDScrapeWork(swsPrev) })
 	scs.add("yandexcloud_sd_configs", *yandexcloud.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getYandexCloudSDScrapeWork(swsPrev) })
 	scs.add("static_configs", 0, func(cfg *Config, _ []*ScrapeWork) []*ScrapeWork { return cfg.getStaticScrapeWork() })
 
diff --git a/lib/protoparser/statsd/parser.go b/lib/protoparser/statsd/parser.go
new file mode 100644
index 000000000..3ae9800d6
--- /dev/null
+++ b/lib/protoparser/statsd/parser.go
@@ -0,0 +1,226 @@
+package statsd
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/metrics"
+	"github.com/valyala/fastjson/fastfloat"
+)
+
+// Statsd metric format with tags: MetricName:value|type|@sample_rate|#tag1:value,tag1...
+const statsdSeparator = '|'
+const statsdPairsSeparator = ':'
+const statsdTagsStartSeparator = '#'
+const statsdTagsSeparator = ','
+
+// Rows contains parsed statsd rows.
+type Rows struct {
+	Rows []Row
+
+	tagsPool []Tag
+}
+
+// Reset resets rs.
+func (rs *Rows) Reset() {
+	// Reset items, so they can be GC'ed
+
+	for i := range rs.Rows {
+		rs.Rows[i].reset()
+	}
+	rs.Rows = rs.Rows[:0]
+
+	for i := range rs.tagsPool {
+		rs.tagsPool[i].reset()
+	}
+	rs.tagsPool = rs.tagsPool[:0]
+}
+
+// Unmarshal unmarshals statsd plaintext protocol rows from s.
+//
+// s shouldn't be modified when rs is in use.
+func (rs *Rows) Unmarshal(s string) {
+	rs.Rows, rs.tagsPool = unmarshalRows(rs.Rows[:0], s, rs.tagsPool[:0])
+}
+
+// Row is a single statsd row.
+type Row struct {
+	Metric    string
+	Tags      []Tag
+	Value     float64
+	Timestamp int64
+}
+
+func (r *Row) reset() {
+	r.Metric = ""
+	r.Tags = nil
+	r.Value = 0
+	r.Timestamp = 0
+}
+
+func (r *Row) unmarshal(s string, tagsPool []Tag) ([]Tag, error) {
+	r.reset()
+	originalString := s
+	s = stripTrailingWhitespace(s)
+	separatorPosition := strings.IndexByte(s, statsdSeparator)
+	if separatorPosition < 0 {
+		s = stripTrailingWhitespace(s)
+	} else {
+		s = stripTrailingWhitespace(s[:separatorPosition])
+	}
+
+	valuesSeparatorPosition := strings.LastIndexByte(s, statsdPairsSeparator)
+
+	if valuesSeparatorPosition == 0 {
+		return tagsPool, fmt.Errorf("cannot find metric name for %q", s)
+	}
+
+	if valuesSeparatorPosition < 0 {
+		return tagsPool, fmt.Errorf("cannot find separator for %q", s)
+	}
+
+	r.Metric = s[:valuesSeparatorPosition]
+	valueStr := s[valuesSeparatorPosition+1:]
+
+	v, err := fastfloat.Parse(valueStr)
+	if err != nil {
+		return tagsPool, fmt.Errorf("cannot unmarshal value from %q: %w; original line: %q", valueStr, err, originalString)
+	}
+	r.Value = v
+
+	// parsing tags
+	tagsSeparatorPosition := strings.LastIndexByte(originalString, statsdTagsStartSeparator)
+
+	if tagsSeparatorPosition < 0 {
+		// no tags
+		return tagsPool, nil
+	}
+
+	tagsStart := len(tagsPool)
+	tagsPool = unmarshalTags(tagsPool, originalString[tagsSeparatorPosition+1:])
+	tags := tagsPool[tagsStart:]
+	r.Tags = tags[:len(tags):len(tags)]
+
+	return tagsPool, nil
+}
+
+func unmarshalRows(dst []Row, s string, tagsPool []Tag) ([]Row, []Tag) {
+	for len(s) > 0 {
+		n := strings.IndexByte(s, '\n')
+		if n < 0 {
+			// The last line.
+			return unmarshalRow(dst, s, tagsPool)
+		}
+		dst, tagsPool = unmarshalRow(dst, s[:n], tagsPool)
+		s = s[n+1:]
+	}
+	return dst, tagsPool
+}
+
+func unmarshalRow(dst []Row, s string, tagsPool []Tag) ([]Row, []Tag) {
+	if len(s) > 0 && s[len(s)-1] == '\r' {
+		s = s[:len(s)-1]
+	}
+	s = stripLeadingWhitespace(s)
+	if len(s) == 0 {
+		// Skip empty line
+		return dst, tagsPool
+	}
+	if cap(dst) > len(dst) {
+		dst = dst[:len(dst)+1]
+	} else {
+		dst = append(dst, Row{})
+	}
+	r := &dst[len(dst)-1]
+	var err error
+	tagsPool, err = r.unmarshal(s, tagsPool)
+	if err != nil {
+		dst = dst[:len(dst)-1]
+		logger.Errorf("cannot unmarshal Statsd line %q: %s", s, err)
+		invalidLines.Inc()
+	}
+	return dst, tagsPool
+}
+
+var invalidLines = metrics.NewCounter(`vm_rows_invalid_total{type="statsd"}`)
+
+func unmarshalTags(dst []Tag, s string) []Tag {
+	for {
+		if cap(dst) > len(dst) {
+			dst = dst[:len(dst)+1]
+		} else {
+			dst = append(dst, Tag{})
+		}
+		tag := &dst[len(dst)-1]
+
+		n := strings.IndexByte(s, statsdTagsSeparator)
+
+		if n < 0 {
+			// The last tag found
+			tag.unmarshal(s)
+			if len(tag.Key) == 0 || len(tag.Value) == 0 {
+				// Skip empty tag
+				dst = dst[:len(dst)-1]
+			}
+			return dst
+		}
+		tag.unmarshal(s[:n])
+		s = s[n+1:]
+		if len(tag.Key) == 0 || len(tag.Value) == 0 {
+			// Skip empty tag
+			dst = dst[:len(dst)-1]
+		}
+	}
+}
+
+// Tag is a statsd tag.
+type Tag struct {
+	Key   string
+	Value string
+}
+
+func (t *Tag) reset() {
+	t.Key = ""
+	t.Value = ""
+}
+
+func (t *Tag) unmarshal(s string) {
+	t.reset()
+	n := strings.IndexByte(s, statsdPairsSeparator)
+	if n < 0 {
+		// Empty tag value.
+		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1100
+		t.Key = s
+		t.Value = s[len(s):]
+	} else {
+		t.Key = s[:n]
+		t.Value = s[n+1:]
+	}
+}
+
+func stripTrailingWhitespace(s string) string {
+	n := len(s)
+	for {
+		n--
+		if n < 0 {
+			return ""
+		}
+		ch := s[n]
+
+		if ch != ' ' && ch != '\t' {
+			return s[:n+1]
+		}
+	}
+}
+
+func stripLeadingWhitespace(s string) string {
+	for len(s) > 0 {
+		ch := s[0]
+		if ch != ' ' && ch != '\t' {
+			return s
+		}
+		s = s[1:]
+	}
+	return ""
+}
diff --git a/lib/protoparser/statsd/stream/streamparser.go b/lib/protoparser/statsd/stream/streamparser.go
new file mode 100644
index 000000000..27d9e4027
--- /dev/null
+++ b/lib/protoparser/statsd/stream/streamparser.go
@@ -0,0 +1,218 @@
+package stream
+
+import (
+	"bufio"
+	"flag"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/statsd"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	trimTimestamp = flag.Duration("statsdTrimTimestamp", time.Second, "Trim timestamps for Statsd data to this duration. "+
+		"Minimum practical duration is 1s. Higher duration (i.e. 1m) may be used for reducing disk space usage for timestamp data")
+)
+
+// Parse parses Statsd lines from r and calls callback for the parsed rows.
+//
+// The callback can be called concurrently multiple times for streamed data from r.
+//
+// callback shouldn't hold rows after returning.
+func Parse(r io.Reader, isGzipped bool, callback func(rows []statsd.Row) error) error {
+	wcr := writeconcurrencylimiter.GetReader(r)
+	defer writeconcurrencylimiter.PutReader(wcr)
+	r = wcr
+
+	if isGzipped {
+		zr, err := common.GetGzipReader(r)
+		if err != nil {
+			return fmt.Errorf("cannot read gzipped statsd data: %w", err)
+		}
+		defer common.PutGzipReader(zr)
+		r = zr
+	}
+
+	ctx := getStreamContext(r)
+	defer putStreamContext(ctx)
+
+	for ctx.Read() {
+		uw := getUnmarshalWork()
+		uw.ctx = ctx
+		uw.callback = callback
+		uw.reqBuf, ctx.reqBuf = ctx.reqBuf, uw.reqBuf
+		ctx.wg.Add(1)
+		common.ScheduleUnmarshalWork(uw)
+		wcr.DecConcurrency()
+	}
+	ctx.wg.Wait()
+	if err := ctx.Error(); err != nil {
+		return err
+	}
+	return ctx.callbackErr
+}
+
+func (ctx *streamContext) Read() bool {
+	readCalls.Inc()
+	if ctx.err != nil || ctx.hasCallbackError() {
+		return false
+	}
+	ctx.reqBuf, ctx.tailBuf, ctx.err = common.ReadLinesBlock(ctx.br, ctx.reqBuf, ctx.tailBuf)
+	if ctx.err != nil {
+		if ctx.err != io.EOF {
+			readErrors.Inc()
+			ctx.err = fmt.Errorf("cannot read statsd plaintext protocol data: %w", ctx.err)
+		}
+		return false
+	}
+	return true
+}
+
+type streamContext struct {
+	br      *bufio.Reader
+	reqBuf  []byte
+	tailBuf []byte
+	err     error
+
+	wg              sync.WaitGroup
+	callbackErrLock sync.Mutex
+	callbackErr     error
+}
+
+func (ctx *streamContext) Error() error {
+	if ctx.err == io.EOF {
+		return nil
+	}
+	return ctx.err
+}
+
+func (ctx *streamContext) hasCallbackError() bool {
+	ctx.callbackErrLock.Lock()
+	ok := ctx.callbackErr != nil
+	ctx.callbackErrLock.Unlock()
+	return ok
+}
+
+func (ctx *streamContext) reset() {
+	ctx.br.Reset(nil)
+	ctx.reqBuf = ctx.reqBuf[:0]
+	ctx.tailBuf = ctx.tailBuf[:0]
+	ctx.err = nil
+	ctx.callbackErr = nil
+}
+
+var (
+	readCalls  = metrics.NewCounter(`vm_protoparser_read_calls_total{type="statsd"}`)
+	readErrors = metrics.NewCounter(`vm_protoparser_read_errors_total{type="statsd"}`)
+	rowsRead   = metrics.NewCounter(`vm_protoparser_rows_read_total{type="statsd"}`)
+)
+
+func getStreamContext(r io.Reader) *streamContext {
+	select {
+	case ctx := <-streamContextPoolCh:
+		ctx.br.Reset(r)
+		return ctx
+	default:
+		if v := streamContextPool.Get(); v != nil {
+			ctx := v.(*streamContext)
+			ctx.br.Reset(r)
+			return ctx
+		}
+		return &streamContext{
+			br: bufio.NewReaderSize(r, 64*1024),
+		}
+	}
+}
+
+func putStreamContext(ctx *streamContext) {
+	ctx.reset()
+	select {
+	case streamContextPoolCh <- ctx:
+	default:
+		streamContextPool.Put(ctx)
+	}
+}
+
+var streamContextPool sync.Pool
+var streamContextPoolCh = make(chan *streamContext, cgroup.AvailableCPUs())
+
+type unmarshalWork struct {
+	rows     statsd.Rows
+	ctx      *streamContext
+	callback func(rows []statsd.Row) error
+	reqBuf   []byte
+}
+
+func (uw *unmarshalWork) reset() {
+	uw.rows.Reset()
+	uw.ctx = nil
+	uw.callback = nil
+	uw.reqBuf = uw.reqBuf[:0]
+}
+
+func (uw *unmarshalWork) runCallback(rows []statsd.Row) {
+	ctx := uw.ctx
+	if err := uw.callback(rows); err != nil {
+		ctx.callbackErrLock.Lock()
+		if ctx.callbackErr == nil {
+			ctx.callbackErr = fmt.Errorf("error when processing imported data: %w", err)
+		}
+		ctx.callbackErrLock.Unlock()
+	}
+	ctx.wg.Done()
+}
+
+// Unmarshal implements common.UnmarshalWork
+func (uw *unmarshalWork) Unmarshal() {
+	uw.rows.Unmarshal(bytesutil.ToUnsafeString(uw.reqBuf))
+	rows := uw.rows.Rows
+	rowsRead.Add(len(rows))
+
+	// Fill missing timestamps with the current timestamp rounded to seconds.
+	currentTimestamp := int64(fasttime.UnixTimestamp())
+	for i := range rows {
+		r := &rows[i]
+		if r.Timestamp == 0 || r.Timestamp == -1 {
+			r.Timestamp = currentTimestamp
+		}
+	}
+
+	// Convert timestamps from seconds to milliseconds.
+	for i := range rows {
+		rows[i].Timestamp *= 1e3
+	}
+
+	// Trim timestamps if required.
+	if tsTrim := trimTimestamp.Milliseconds(); tsTrim > 1000 {
+		for i := range rows {
+			row := &rows[i]
+			row.Timestamp -= row.Timestamp % tsTrim
+		}
+	}
+
+	uw.runCallback(rows)
+	putUnmarshalWork(uw)
+}
+
+func getUnmarshalWork() *unmarshalWork {
+	v := unmarshalWorkPool.Get()
+	if v == nil {
+		return &unmarshalWork{}
+	}
+	return v.(*unmarshalWork)
+}
+
+func putUnmarshalWork(uw *unmarshalWork) {
+	uw.reset()
+	unmarshalWorkPool.Put(uw)
+}
+
+var unmarshalWorkPool sync.Pool
diff --git a/lib/streamaggr/deduplicator.go b/lib/streamaggr/deduplicator.go
index 9cc3fee22..a96d765c2 100644
--- a/lib/streamaggr/deduplicator.go
+++ b/lib/streamaggr/deduplicator.go
@@ -15,7 +15,6 @@ import (
 // Deduplicator deduplicates samples per each time series.
 type Deduplicator struct {
 	da *dedupAggr
-	lc promutils.LabelsCompressor
 
 	dropLabels []string
 
@@ -38,8 +37,7 @@ type Deduplicator struct {
 // 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 {
 	d := &Deduplicator{
-		da: newDedupAggr(),
-
+		da:         newDedupAggr(),
 		dropLabels: dropLabels,
 
 		stopCh: make(chan struct{}),
@@ -54,13 +52,6 @@ func NewDeduplicator(pushFunc PushFunc, dedupInterval time.Duration, dropLabels
 		return float64(d.da.itemsCount())
 	})
 
-	_ = ms.NewGauge(`vm_streamaggr_labels_compressor_size_bytes`, func() float64 {
-		return float64(d.lc.SizeBytes())
-	})
-	_ = ms.NewGauge(`vm_streamaggr_labels_compressor_items_count`, func() float64 {
-		return float64(d.lc.ItemsCount())
-	})
-
 	d.dedupFlushDuration = ms.GetOrCreateHistogram(`vm_streamaggr_dedup_flush_duration_seconds`)
 	d.dedupFlushTimeouts = ms.GetOrCreateCounter(`vm_streamaggr_dedup_flush_timeouts_total`)
 
@@ -103,7 +94,7 @@ func (d *Deduplicator) Push(tss []prompbmarshal.TimeSeries) {
 		}
 		labels.Sort()
 
-		buf = d.lc.Compress(buf[:0], labels.Labels)
+		buf = lc.Compress(buf[:0], labels.Labels)
 		key := bytesutil.InternBytes(buf)
 		for _, s := range ts.Samples {
 			pss = append(pss, pushSample{
@@ -155,7 +146,7 @@ func (d *Deduplicator) flush(pushFunc PushFunc, dedupInterval time.Duration) {
 		samples := ctx.samples
 		for _, ps := range pss {
 			labelsLen := len(labels)
-			labels = decompressLabels(labels, &d.lc, ps.key)
+			labels = decompressLabels(labels, ps.key)
 
 			samplesLen := len(samples)
 			samples = append(samples, prompbmarshal.Sample{
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index 5cb480aad..3e5b72ecd 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -45,6 +45,19 @@ var supportedOutputs = []string{
 	"quantiles(phi1, ..., phiN)",
 }
 
+var (
+	// lc contains information about all compressed labels for streaming aggregation
+	lc promutils.LabelsCompressor
+
+	_ = metrics.NewGauge(`vm_streamaggr_labels_compressor_size_bytes`, func() float64 {
+		return float64(lc.SizeBytes())
+	})
+
+	_ = metrics.NewGauge(`vm_streamaggr_labels_compressor_items_count`, func() float64 {
+		return float64(lc.ItemsCount())
+	})
+)
+
 // LoadFromFile loads Aggregators from the given path and uses the given pushFunc for pushing the aggregated data.
 //
 // opts can contain additional options. If opts is nil, then default options are used.
@@ -269,21 +282,6 @@ func newAggregatorsFromData(data []byte, pushFunc PushFunc, opts *Options) (*Agg
 		return float64(n)
 	})
 
-	_ = ms.NewGauge(`vm_streamaggr_labels_compressor_size_bytes`, func() float64 {
-		n := uint64(0)
-		for _, aggr := range as {
-			n += aggr.lc.SizeBytes()
-		}
-		return float64(n)
-	})
-	_ = ms.NewGauge(`vm_streamaggr_labels_compressor_items_count`, func() float64 {
-		n := uint64(0)
-		for _, aggr := range as {
-			n += aggr.lc.ItemsCount()
-		}
-		return float64(n)
-	})
-
 	metrics.RegisterSet(ms)
 	return &Aggregators{
 		as:         as,
@@ -361,9 +359,6 @@ type aggregator struct {
 	// aggrStates contains aggregate states for the given outputs
 	aggrStates []aggrState
 
-	// lc is used for compressing series keys before passing them to dedupAggr and aggrState
-	lc promutils.LabelsCompressor
-
 	// minTimestamp is used for ignoring old samples when ignoreOldSamples is set
 	minTimestamp atomic.Int64
 
@@ -822,7 +817,7 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 			outputLabels.Labels = append(outputLabels.Labels, labels.Labels...)
 		}
 
-		buf = compressLabels(buf[:0], &a.lc, inputLabels.Labels, outputLabels.Labels)
+		buf = compressLabels(buf[:0], inputLabels.Labels, outputLabels.Labels)
 		key := bytesutil.InternBytes(buf)
 		for _, sample := range ts.Samples {
 			if math.IsNaN(sample.Value) {
@@ -850,7 +845,7 @@ func (a *aggregator) Push(tss []prompbmarshal.TimeSeries, matchIdxs []byte) {
 	}
 }
 
-func compressLabels(dst []byte, lc *promutils.LabelsCompressor, inputLabels, outputLabels []prompbmarshal.Label) []byte {
+func compressLabels(dst []byte, inputLabels, outputLabels []prompbmarshal.Label) []byte {
 	bb := bbPool.Get()
 	bb.B = lc.Compress(bb.B, inputLabels)
 	dst = encoding.MarshalVarUint64(dst, uint64(len(bb.B)))
@@ -860,7 +855,7 @@ func compressLabels(dst []byte, lc *promutils.LabelsCompressor, inputLabels, out
 	return dst
 }
 
-func decompressLabels(dst []prompbmarshal.Label, lc *promutils.LabelsCompressor, key string) []prompbmarshal.Label {
+func decompressLabels(dst []prompbmarshal.Label, key string) []prompbmarshal.Label {
 	return lc.Decompress(dst, bytesutil.ToUnsafeBytes(key))
 }
 
@@ -1035,7 +1030,7 @@ func (ctx *flushCtx) flushSeries() {
 func (ctx *flushCtx) appendSeries(key, suffix string, timestamp int64, value float64) {
 	labelsLen := len(ctx.labels)
 	samplesLen := len(ctx.samples)
-	ctx.labels = decompressLabels(ctx.labels, &ctx.a.lc, key)
+	ctx.labels = decompressLabels(ctx.labels, key)
 	if !ctx.a.keepMetricNames {
 		ctx.labels = addMetricSuffix(ctx.labels, labelsLen, ctx.a.suffix, suffix)
 	}
@@ -1058,7 +1053,7 @@ func (ctx *flushCtx) appendSeries(key, suffix string, timestamp int64, value flo
 func (ctx *flushCtx) appendSeriesWithExtraLabel(key, suffix string, timestamp int64, value float64, extraName, extraValue string) {
 	labelsLen := len(ctx.labels)
 	samplesLen := len(ctx.samples)
-	ctx.labels = decompressLabels(ctx.labels, &ctx.a.lc, key)
+	ctx.labels = decompressLabels(ctx.labels, key)
 	if !ctx.a.keepMetricNames {
 		ctx.labels = addMetricSuffix(ctx.labels, labelsLen, ctx.a.suffix, suffix)
 	}


Test patch:
diff --git a/lib/promscrape/discovery/vultr/api_test.go b/lib/promscrape/discovery/vultr/api_test.go
new file mode 100644
index 000000000..06e491446
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/api_test.go
@@ -0,0 +1,16 @@
+package vultr
+
+import (
+	"testing"
+)
+
+func TestNewAPIConfig(t *testing.T) {
+
+	sdc := &SDConfig{}
+	baseDir := "."
+	_, err := newAPIConfig(sdc, baseDir)
+	if err != nil {
+		t.Errorf("newAPIConfig failed with, err: %v", err)
+		return
+	}
+}
diff --git a/lib/promscrape/discovery/vultr/instance_test.go b/lib/promscrape/discovery/vultr/instance_test.go
new file mode 100644
index 000000000..b888c1a13
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/instance_test.go
@@ -0,0 +1,301 @@
+package vultr
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+)
+
+// TestGetInstances runs general test cases for GetInstances
+func TestGetInstances(t *testing.T) {
+	testCases := []struct {
+		name           string
+		apiResponse    string
+		apiError       bool
+		expectError    bool
+		expectResponse []Instance
+	}{
+		{
+			name:           "success response",
+			apiResponse:    mockListInstanceSuccessResp,
+			apiError:       false,
+			expectError:    false,
+			expectResponse: expectSuccessInstances,
+		},
+		{
+			name:           "failed response",
+			apiResponse:    mockListInstanceFailedResp,
+			apiError:       true,
+			expectError:    true,
+			expectResponse: nil,
+		},
+	}
+
+	for _, tt := range testCases {
+		t.Run(tt.name, func(t *testing.T) {
+			// Prepare a mock Vultr server.
+			mockServer := newMockVultrServer(func() ([]byte, error) {
+				var e error
+				if tt.apiError {
+					e = errors.New("mock error")
+				}
+				return []byte(tt.apiResponse), e
+			})
+
+			// Prepare a discovery HTTP client who calls mock server.
+			client, _ := discoveryutils.NewClient(mockServer.URL, nil, nil, nil, &promauth.HTTPClientConfig{})
+			cfg := &apiConfig{
+				c: client,
+			}
+
+			// execute `getInstances`
+			instances, err := getInstances(cfg)
+
+			// evaluate test result
+			if tt.expectError != (err != nil) {
+				t.Errorf("getInstances expect (error != nil): %t, got error: %v", tt.expectError, err)
+			}
+
+			if !reflect.DeepEqual(tt.expectResponse, instances) {
+				t.Errorf("getInstances expect result: %v, got: %v", tt.expectResponse, instances)
+			}
+		})
+	}
+}
+
+// TestGetInstancesPaging run test cases for response with multiple pages.
+func TestGetInstancesPaging(t *testing.T) {
+	// Prepare a mock Vultr server.
+	// requestCount control the mock response for different page request.
+	requestCount := 0
+
+	mockServer := newMockVultrServer(func() ([]byte, error) {
+		// for the 1st request, response with `next` cursor
+		if requestCount == 0 {
+			requestCount++
+			return []byte(mockListInstanceSuccessPage0Resp), nil
+		}
+		// for the 2nd+ request, response with `prev` cursor and empty `next`.
+		return []byte(mockListInstanceSuccessPage1Resp), nil
+	})
+
+	// Prepare a discovery HTTP client who calls mock server.
+	client, _ := discoveryutils.NewClient(mockServer.URL, nil, nil, nil, &promauth.HTTPClientConfig{})
+	cfg := &apiConfig{
+		c: client,
+	}
+
+	// execute `getInstances`
+	instances, err := getInstances(cfg)
+
+	// evaluate test result
+	if err != nil {
+		t.Errorf("getInstances expect error: %v, got error: %v", nil, err)
+	}
+
+	if !reflect.DeepEqual(expectSuccessPagingInstances, instances) {
+		t.Errorf("getInstances expect result: %v, got: %v", expectSuccessPagingInstances, instances)
+	}
+}
+
+// ------------ Test dataset ------------
+var (
+	// mockListInstanceSuccessResp is crawled from a real-world response of ListInstance API
+	// with sensitive info removed/modified.
+	mockListInstanceSuccessResp = `{
+	"instances": [{
+		"id": "fake-id-07f7-4b68-88ac-fake-id",
+		"os": "Ubuntu 22.04 x64",
+		"ram": 1024,
+		"disk": 25,
+		"main_ip": "64.176.84.27",
+		"vcpu_count": 1,
+		"region": "sgp",
+		"plan": "vc2-1c-1gb",
+		"date_created": "2024-04-05T05:41:28+00:00",
+		"status": "active",
+		"allowed_bandwidth": 1,
+		"netmask_v4": "255.255.254.0",
+		"gateway_v4": "64.176.63.2",
+		"power_status": "running",
+		"server_status": "installingbooting",
+		"v6_network": "2002:18f0:4100:263a::",
+		"v6_main_ip": "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+		"v6_network_size": 64,
+		"label": "vultr-sd",
+		"internal_ip": "",
+		"kvm": "https:\/\/my.vultr.com\/subs\/vps\/novnc\/api.php?data=secret_data_string",
+		"hostname": "vultr-sd",
+		"tag": "",
+		"tags": [],
+		"os_id": 1743,
+		"app_id": 0,
+		"image_id": "",
+		"firewall_group_id": "",
+		"features": ["ipv6"],
+		"user_scheme": "root"
+	}],
+	"meta": {
+		"total": 1,
+		"links": {
+			"next": "",
+			"prev": ""
+		}
+	}
+}`
+	expectSuccessInstances = []Instance{
+		{
+			ID:               "fake-id-07f7-4b68-88ac-fake-id",
+			Os:               "Ubuntu 22.04 x64",
+			RAM:              1024,
+			Disk:             25,
+			MainIP:           "64.176.84.27",
+			VCPUCount:        1,
+			Region:           "sgp",
+			Plan:             "vc2-1c-1gb",
+			AllowedBandwidth: 1,
+			ServerStatus:     "installingbooting",
+			V6MainIP:         "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			Label:            "vultr-sd",
+			InternalIP:       "",
+			Hostname:         "vultr-sd",
+			Tags:             []string{},
+			OsID:             1743,
+			Features:         []string{"ipv6"},
+		},
+	}
+)
+
+var (
+	mockListInstanceFailedResp = `{"error":"Invalid API token.","status":401}`
+)
+
+var (
+	// mockListInstanceSuccessPage0Resp contains `next` cursor
+	mockListInstanceSuccessPage0Resp = `{
+	"instances": [{
+		"id": "fake-id-07f7-4b68-88ac-fake-id",
+		"os": "Ubuntu 22.04 x64",
+		"ram": 1024,
+		"disk": 25,
+		"main_ip": "64.176.84.27",
+		"vcpu_count": 1,
+		"region": "sgp",
+		"plan": "vc2-1c-1gb",
+		"date_created": "2024-04-05T05:41:28+00:00",
+		"status": "active",
+		"allowed_bandwidth": 1,
+		"netmask_v4": "255.255.254.0",
+		"gateway_v4": "64.176.63.2",
+		"power_status": "running",
+		"server_status": "installingbooting",
+		"v6_network": "2002:18f0:4100:263a::",
+		"v6_main_ip": "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+		"v6_network_size": 64,
+		"label": "vultr-sd",
+		"internal_ip": "",
+		"kvm": "https:\/\/my.vultr.com\/subs\/vps\/novnc\/api.php?data=secret_data_string",
+		"hostname": "vultr-sd",
+		"tag": "",
+		"tags": [],
+		"os_id": 1743,
+		"app_id": 0,
+		"image_id": "",
+		"firewall_group_id": "",
+		"features": ["ipv6"],
+		"user_scheme": "root"
+	}],
+	"meta": {
+		"total": 2,
+		"links": {
+			"next": "fake-cursor-string",
+			"prev": ""
+		}
+	}
+}`
+	// mockListInstanceSuccessPage1Resp contains `prev` cursor
+	mockListInstanceSuccessPage1Resp = `{
+	"instances": [{
+		"id": "fake-id-07f7-4b68-88ac-fake-id",
+		"os": "Ubuntu 22.04 x64",
+		"ram": 1024,
+		"disk": 25,
+		"main_ip": "64.176.84.27",
+		"vcpu_count": 1,
+		"region": "sgp",
+		"plan": "vc2-1c-1gb",
+		"date_created": "2024-04-05T05:41:28+00:00",
+		"status": "active",
+		"allowed_bandwidth": 1,
+		"netmask_v4": "255.255.254.0",
+		"gateway_v4": "64.176.63.2",
+		"power_status": "running",
+		"server_status": "installingbooting",
+		"v6_network": "2002:18f0:4100:263a::",
+		"v6_main_ip": "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+		"v6_network_size": 64,
+		"label": "vultr-sd",
+		"internal_ip": "",
+		"kvm": "https:\/\/my.vultr.com\/subs\/vps\/novnc\/api.php?data=secret_data_string",
+		"hostname": "vultr-sd",
+		"tag": "",
+		"tags": [],
+		"os_id": 1743,
+		"app_id": 0,
+		"image_id": "",
+		"firewall_group_id": "",
+		"features": ["ipv6"],
+		"user_scheme": "root"
+	}],
+	"meta": {
+		"total": 2,
+		"links": {
+			"next": "",
+			"prev": "fake-cursor-string"
+		}
+	}
+}`
+	expectSuccessPagingInstances = []Instance{
+		{
+			ID:               "fake-id-07f7-4b68-88ac-fake-id",
+			Os:               "Ubuntu 22.04 x64",
+			RAM:              1024,
+			Disk:             25,
+			MainIP:           "64.176.84.27",
+			VCPUCount:        1,
+			Region:           "sgp",
+			Plan:             "vc2-1c-1gb",
+			AllowedBandwidth: 1,
+			ServerStatus:     "installingbooting",
+			V6MainIP:         "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			Label:            "vultr-sd",
+			InternalIP:       "",
+			Hostname:         "vultr-sd",
+			Tags:             []string{},
+			OsID:             1743,
+			Features:         []string{"ipv6"},
+		},
+		{
+			ID:               "fake-id-07f7-4b68-88ac-fake-id",
+			Os:               "Ubuntu 22.04 x64",
+			RAM:              1024,
+			Disk:             25,
+			MainIP:           "64.176.84.27",
+			VCPUCount:        1,
+			Region:           "sgp",
+			Plan:             "vc2-1c-1gb",
+			AllowedBandwidth: 1,
+			ServerStatus:     "installingbooting",
+			V6MainIP:         "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			Label:            "vultr-sd",
+			InternalIP:       "",
+			Hostname:         "vultr-sd",
+			Tags:             []string{},
+			OsID:             1743,
+			Features:         []string{"ipv6"},
+		},
+	}
+)
diff --git a/lib/promscrape/discovery/vultr/mock_server_test.go b/lib/promscrape/discovery/vultr/mock_server_test.go
new file mode 100644
index 000000000..c0dd79e46
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/mock_server_test.go
@@ -0,0 +1,40 @@
+package vultr
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+)
+
+func newMockVultrServer(jsonResponse func() ([]byte, error)) *vultrServer {
+	rw := &vultrServer{}
+	rw.Server = httptest.NewServer(http.HandlerFunc(rw.handler))
+	rw.jsonResponse = jsonResponse
+	return rw
+}
+
+type vultrServer struct {
+	*httptest.Server
+	jsonResponse func() ([]byte, error)
+}
+
+func (rw *vultrServer) err(w http.ResponseWriter, err error) {
+	w.WriteHeader(http.StatusBadRequest)
+	w.Write([]byte(err.Error()))
+}
+
+func (rw *vultrServer) handler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		rw.err(w, fmt.Errorf("bad method %q", r.Method))
+		return
+	}
+
+	resp, err := rw.jsonResponse()
+	if err != nil {
+		rw.err(w, err)
+		return
+	}
+
+	w.Write(resp)
+	w.WriteHeader(http.StatusOK)
+}
diff --git a/lib/promscrape/discovery/vultr/vultr_test.go b/lib/promscrape/discovery/vultr/vultr_test.go
new file mode 100644
index 000000000..572782726
--- /dev/null
+++ b/lib/promscrape/discovery/vultr/vultr_test.go
@@ -0,0 +1,96 @@
+package vultr
+
+import (
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+)
+
+func TestGetInstanceLabels(t *testing.T) {
+	input := []Instance{
+		{
+			ID:               "fake-id-07f7-4b68-88ac-fake-id",
+			Os:               "Ubuntu 22.04 x64",
+			RAM:              1024,
+			Disk:             25,
+			MainIP:           "64.176.84.27",
+			VCPUCount:        1,
+			Region:           "sgp",
+			Plan:             "vc2-1c-1gb",
+			AllowedBandwidth: 1,
+			ServerStatus:     "installingbooting",
+			V6MainIP:         "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			Label:            "vultr-sd",
+			InternalIP:       "",
+			Hostname:         "vultr-sd",
+			Tags:             []string{"mock tags"},
+			OsID:             1743,
+			Features:         []string{"ipv6"},
+		},
+		{
+			ID:               "fake-id-07f7-4b68-88ac-fake-id",
+			Os:               "Ubuntu 22.04 x64",
+			RAM:              1024,
+			Disk:             25,
+			MainIP:           "64.176.84.27",
+			VCPUCount:        1,
+			Region:           "sgp",
+			Plan:             "vc2-1c-1gb",
+			AllowedBandwidth: 1,
+			ServerStatus:     "installingbooting",
+			V6MainIP:         "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			Label:            "vultr-sd",
+			InternalIP:       "",
+			Hostname:         "vultr-sd",
+			Tags:             []string{"mock tags"},
+			OsID:             1743,
+			Features:         []string{"ipv6"},
+		},
+	}
+
+	expect := []*promutils.Labels{
+		promutils.NewLabelsFromMap(map[string]string{
+			"__address__":                                "64.176.84.27:8080",
+			"__meta_vultr_instance_id":                   "fake-id-07f7-4b68-88ac-fake-id",
+			"__meta_vultr_instance_label":                "vultr-sd",
+			"__meta_vultr_instance_os":                   "Ubuntu 22.04 x64",
+			"__meta_vultr_instance_os_id":                "1743",
+			"__meta_vultr_instance_region":               "sgp",
+			"__meta_vultr_instance_plan":                 "vc2-1c-1gb",
+			"__meta_vultr_instance_main_ip":              "64.176.84.27",
+			"__meta_vultr_instance_internal_ip":          "",
+			"__meta_vultr_instance_main_ipv6":            "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			"__meta_vultr_instance_hostname":             "vultr-sd",
+			"__meta_vultr_instance_server_status":        "installingbooting",
+			"__meta_vultr_instance_vcpu_count":           "1",
+			"__meta_vultr_instance_ram_mb":               "1024",
+			"__meta_vultr_instance_allowed_bandwidth_gb": "1",
+			"__meta_vultr_instance_disk_gb":              "25",
+			"__meta_vultr_instance_features":             ",ipv6,",
+			"__meta_vultr_instance_tags":                 ",mock tags,",
+		}),
+		promutils.NewLabelsFromMap(map[string]string{
+			"__address__":                                "64.176.84.27:8080",
+			"__meta_vultr_instance_id":                   "fake-id-07f7-4b68-88ac-fake-id",
+			"__meta_vultr_instance_label":                "vultr-sd",
+			"__meta_vultr_instance_os":                   "Ubuntu 22.04 x64",
+			"__meta_vultr_instance_os_id":                "1743",
+			"__meta_vultr_instance_region":               "sgp",
+			"__meta_vultr_instance_plan":                 "vc2-1c-1gb",
+			"__meta_vultr_instance_main_ip":              "64.176.84.27",
+			"__meta_vultr_instance_internal_ip":          "",
+			"__meta_vultr_instance_main_ipv6":            "2002:18f0:4100:263a:5300:07ff:fdd7:691c",
+			"__meta_vultr_instance_hostname":             "vultr-sd",
+			"__meta_vultr_instance_server_status":        "installingbooting",
+			"__meta_vultr_instance_vcpu_count":           "1",
+			"__meta_vultr_instance_ram_mb":               "1024",
+			"__meta_vultr_instance_allowed_bandwidth_gb": "1",
+			"__meta_vultr_instance_disk_gb":              "25",
+			"__meta_vultr_instance_features":             ",ipv6,",
+			"__meta_vultr_instance_tags":                 ",mock tags,",
+		}),
+	}
+	labels := getInstanceLabels(input, 8080)
+	discoveryutils.TestEqualLabelss(t, labels, expect)
+}
diff --git a/lib/protoparser/statsd/parser_test.go b/lib/protoparser/statsd/parser_test.go
new file mode 100644
index 000000000..aeb498bb7
--- /dev/null
+++ b/lib/protoparser/statsd/parser_test.go
@@ -0,0 +1,367 @@
+package statsd
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestUnmarshalTagsSuccess(t *testing.T) {
+	f := func(dst []Tag, s string, tagsPoolExpected []Tag) {
+		t.Helper()
+
+		tagsPool := unmarshalTags(dst, s)
+		if !reflect.DeepEqual(tagsPool, tagsPoolExpected) {
+			t.Fatalf("unexpected tags;\ngot\n%+v;\nwant\n%+v", tagsPool, tagsPoolExpected)
+		}
+
+		// Try unmarshaling again
+		tagsPool = unmarshalTags(dst, s)
+		if !reflect.DeepEqual(tagsPool, tagsPoolExpected) {
+			t.Fatalf("unexpected tags on second unmarshal;\ngot\n%+v;\nwant\n%+v", tagsPool, tagsPoolExpected)
+		}
+	}
+
+	f([]Tag{}, "foo:bar", []Tag{
+		{
+			Key:   "foo",
+			Value: "bar",
+		},
+	})
+
+	f([]Tag{}, "foo:bar,qwe:123", []Tag{
+		{
+			Key:   "foo",
+			Value: "bar",
+		},
+		{
+			Key:   "qwe",
+			Value: "123",
+		},
+	})
+
+	f([]Tag{}, "foo.qwe:bar", []Tag{
+		{
+			Key:   "foo.qwe",
+			Value: "bar",
+		},
+	})
+
+	f([]Tag{}, "foo:10", []Tag{
+		{
+			Key:   "foo",
+			Value: "10",
+		},
+	})
+
+	f([]Tag{}, "foo: _qwe", []Tag{
+		{
+			Key:   "foo",
+			Value: " _qwe",
+		},
+	})
+
+	f([]Tag{}, "foo:qwe    ", []Tag{
+		{
+			Key:   "foo",
+			Value: "qwe    ",
+		},
+	})
+
+	f([]Tag{}, "foo  asd:qwe    ", []Tag{
+		{
+			Key:   "foo  asd",
+			Value: "qwe    ",
+		},
+	})
+
+	f([]Tag{}, "foo:var:123", []Tag{
+		{
+			Key:   "foo",
+			Value: "var:123",
+		},
+	})
+
+	// invalid tags
+	f([]Tag{}, ":bar", []Tag{})
+	f([]Tag{}, "foo:", []Tag{})
+	f([]Tag{}, "   ", []Tag{})
+}
+
+func TestRowsUnmarshalSuccess(t *testing.T) {
+	f := func(s string, rowsExpected *Rows) {
+		t.Helper()
+		var rows Rows
+		rows.Unmarshal(s)
+		if !reflect.DeepEqual(rows.Rows, rowsExpected.Rows) {
+			t.Fatalf("unexpected rows;\ngot\n%+v;\nwant\n%+v", rows.Rows, rowsExpected.Rows)
+		}
+
+		// Try unmarshaling again
+		rows.Unmarshal(s)
+		if !reflect.DeepEqual(rows.Rows, rowsExpected.Rows) {
+			t.Fatalf("unexpected rows on second unmarshal;\ngot\n%+v;\nwant\n%+v", rows.Rows, rowsExpected.Rows)
+		}
+
+		rows.Reset()
+		if len(rows.Rows) != 0 {
+			t.Fatalf("non-empty rows after reset: %+v", rows.Rows)
+		}
+	}
+
+	// Empty line
+	f("", &Rows{})
+	f("\r", &Rows{})
+	f("\n\n", &Rows{})
+	f("\n\r\n", &Rows{})
+
+	// Single line
+	f(" 123:455", &Rows{
+		Rows: []Row{{
+			Metric: "123",
+			Value:  455,
+		}},
+	})
+	f("123:455 |c", &Rows{
+		Rows: []Row{{
+			Metric: "123",
+			Value:  455,
+		}},
+	})
+	f("foobar:-123.456|c", &Rows{
+		Rows: []Row{{
+			Metric: "foobar",
+			Value:  -123.456,
+		}},
+	})
+	f("foo.bar:123.456|c\n", &Rows{
+		Rows: []Row{{
+			Metric: "foo.bar",
+			Value:  123.456,
+		}},
+	})
+
+	// with sample rate
+	f("foo.bar:1|c|@0.1", &Rows{
+		Rows: []Row{{
+			Metric: "foo.bar",
+			Value:  1,
+		}},
+	})
+
+	// without specifying metric unit
+	f("foo.bar:123", &Rows{
+		Rows: []Row{{
+			Metric: "foo.bar",
+			Value:  123,
+		}},
+	})
+	// without specifying metric unit but with tags
+	f("foo.bar:123|#foo:bar", &Rows{
+		Rows: []Row{{
+			Metric: "foo.bar",
+			Value:  123,
+			Tags: []Tag{
+				{
+					Key:   "foo",
+					Value: "bar",
+				},
+			},
+		}},
+	})
+
+	f("foo.bar:123.456|c|#foo:bar,qwe:asd", &Rows{
+		Rows: []Row{{
+			Metric: "foo.bar",
+			Value:  123.456,
+			Tags: []Tag{
+				{
+					Key:   "foo",
+					Value: "bar",
+				},
+				{
+					Key:   "qwe",
+					Value: "asd",
+				},
+			},
+		}},
+	})
+
+	// Whitespace in metric name, tag name and tag value
+	f("s a:1|c|#ta g1:aaa1,tag2:bb b2", &Rows{
+		Rows: []Row{{
+			Metric: "s a",
+			Value:  1,
+			Tags: []Tag{
+				{
+					Key:   "ta g1",
+					Value: "aaa1",
+				},
+				{
+					Key:   "tag2",
+					Value: "bb b2",
+				},
+			},
+		}},
+	})
+
+	// Tags
+	f("foo:1|c", &Rows{
+		Rows: []Row{{
+			Metric: "foo",
+			Value:  1,
+		}},
+	})
+	// Empty tag name
+	f("foo:1|#:123", &Rows{
+		Rows: []Row{{
+			Metric: "foo",
+			Tags:   []Tag{},
+			Value:  1,
+		}},
+	})
+	// Empty tag value
+	f("foo:1|#tag1:", &Rows{
+		Rows: []Row{{
+			Metric: "foo",
+			Tags:   []Tag{},
+			Value:  1,
+		}},
+	})
+	f("foo:1|#bar:baz,aa:,x:y,:z", &Rows{
+		Rows: []Row{{
+			Metric: "foo",
+			Tags: []Tag{
+				{
+					Key:   "bar",
+					Value: "baz",
+				},
+				{
+					Key:   "x",
+					Value: "y",
+				},
+			},
+			Value: 1,
+		}},
+	})
+
+	// Multi lines
+	f("foo:0.3|c\naaa:3|g\nbar.baz:0.34|c\n", &Rows{
+		Rows: []Row{
+			{
+				Metric: "foo",
+				Value:  0.3,
+			},
+			{
+				Metric: "aaa",
+				Value:  3,
+			},
+			{
+				Metric: "bar.baz",
+				Value:  0.34,
+			},
+		},
+	})
+
+	f("foo:0.3|c|#tag1:1,tag2:2\naaa:3|g|#tag3:3,tag4:4", &Rows{
+		Rows: []Row{
+			{
+				Metric: "foo",
+				Value:  0.3,
+				Tags: []Tag{
+					{
+						Key:   "tag1",
+						Value: "1",
+					},
+					{
+						Key:   "tag2",
+						Value: "2",
+					},
+				},
+			},
+			{
+				Metric: "aaa",
+				Value:  3,
+				Tags: []Tag{
+					{
+						Key:   "tag3",
+						Value: "3",
+					},
+					{
+						Key:   "tag4",
+						Value: "4",
+					},
+				},
+			},
+		},
+	})
+
+	// Multi lines with invalid line
+	f("foo:0.3|c\naaa\nbar.baz:0.34\n", &Rows{
+		Rows: []Row{
+			{
+				Metric: "foo",
+				Value:  0.3,
+			},
+			{
+				Metric: "bar.baz",
+				Value:  0.34,
+			},
+		},
+	})
+
+	// Whitespace after at the end
+	f("foo.baz:125|c\na:1.34\t  ", &Rows{
+		Rows: []Row{
+			{
+				Metric: "foo.baz",
+				Value:  125,
+			},
+			{
+				Metric: "a",
+				Value:  1.34,
+			},
+		},
+	})
+
+	// ignores sample rate
+	f("foo.baz:125|c|@0.5#tag1:12", &Rows{
+		Rows: []Row{
+			{
+				Metric: "foo.baz",
+				Value:  125,
+				Tags: []Tag{
+					{
+						Key:   "tag1",
+						Value: "12",
+					},
+				},
+			},
+		},
+	})
+}
+
+func TestRowsUnmarshalFailure(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+		var rows Rows
+		rows.Unmarshal(s)
+		if len(rows.Rows) != 0 {
+			t.Fatalf("unexpected number of rows parsed; got %d; want 0", len(rows.Rows))
+		}
+
+		// Try again
+		rows.Unmarshal(s)
+		if len(rows.Rows) != 0 {
+			t.Fatalf("unexpected number of rows parsed; got %d; want 0", len(rows.Rows))
+		}
+	}
+
+	// random string
+	f("aaa")
+
+	// empty value
+	f("foo:")
+
+	// empty metric name
+	f(":12")
+}
diff --git a/lib/protoparser/statsd/parser_timing_test.go b/lib/protoparser/statsd/parser_timing_test.go
new file mode 100644
index 000000000..d9e3c6479
--- /dev/null
+++ b/lib/protoparser/statsd/parser_timing_test.go
@@ -0,0 +1,25 @@
+package statsd
+
+import (
+	"fmt"
+	"testing"
+)
+
+func BenchmarkRowsUnmarshal(b *testing.B) {
+	s := `cpu.usage_user:1.23|c
+cpu.usage_system:23.344|c
+cpu.usage_iowait:3.3443|c
+cpu.usage_irq:0.34432|c
+`
+	b.SetBytes(int64(len(s)))
+	b.ReportAllocs()
+	b.RunParallel(func(pb *testing.PB) {
+		var rows Rows
+		for pb.Next() {
+			rows.Unmarshal(s)
+			if len(rows.Rows) != 4 {
+				panic(fmt.Errorf("unexpected number of rows unmarshaled: got %d; want 4", len(rows.Rows)))
+			}
+		}
+	})
+}
diff --git a/lib/protoparser/statsd/stream/streamparser_test.go b/lib/protoparser/statsd/stream/streamparser_test.go
new file mode 100644
index 000000000..8800dfd45
--- /dev/null
+++ b/lib/protoparser/statsd/stream/streamparser_test.go
@@ -0,0 +1,60 @@
+package stream
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/statsd"
+)
+
+func Test_streamContext_Read(t *testing.T) {
+	f := func(s string, rowsExpected *statsd.Rows) {
+		t.Helper()
+		ctx := getStreamContext(strings.NewReader(s))
+		if !ctx.Read() {
+			t.Fatalf("expecting successful read")
+		}
+		uw := getUnmarshalWork()
+		callbackCalls := 0
+		uw.ctx = ctx
+		uw.callback = func(rows []statsd.Row) error {
+			callbackCalls++
+			if len(rows) != len(rowsExpected.Rows) {
+				t.Fatalf("different len of expected rows;\ngot\n%+v;\nwant\n%+v", rows, rowsExpected.Rows)
+			}
+			if !reflect.DeepEqual(rows, rowsExpected.Rows) {
+				t.Fatalf("unexpected rows;\ngot\n%+v;\nwant\n%+v", rows, rowsExpected.Rows)
+			}
+			return nil
+		}
+		uw.reqBuf = append(uw.reqBuf[:0], ctx.reqBuf...)
+		ctx.wg.Add(1)
+		uw.Unmarshal()
+		if callbackCalls != 1 {
+			t.Fatalf("unexpected number of callback calls; got %d; want 1", callbackCalls)
+		}
+	}
+
+	// Full line without tags
+	f("aaa:1123|c", &statsd.Rows{
+		Rows: []statsd.Row{{
+			Metric:    "aaa",
+			Value:     1123,
+			Timestamp: int64(fasttime.UnixTimestamp()) * 1000,
+		}},
+	})
+	// Full line with tags
+	f("aaa:1123|c|#x:y", &statsd.Rows{
+		Rows: []statsd.Row{{
+			Metric: "aaa",
+			Tags: []statsd.Tag{{
+				Key:   "x",
+				Value: "y",
+			}},
+			Value:     1123,
+			Timestamp: int64(fasttime.UnixTimestamp()) * 1000,
+		}},
+	})
+}
diff --git a/lib/streamaggr/dedup_timing_test.go b/lib/streamaggr/dedup_timing_test.go
index 2b906fe39..abb1d0451 100644
--- a/lib/streamaggr/dedup_timing_test.go
+++ b/lib/streamaggr/dedup_timing_test.go
@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
 func BenchmarkDedupAggr(b *testing.B) {
@@ -50,7 +49,6 @@ func benchmarkDedupAggr(b *testing.B, samplesPerPush int) {
 }
 
 func newBenchSamples(count int) []pushSample {
-	var lc promutils.LabelsCompressor
 	labels := []prompbmarshal.Label{
 		{
 			Name:  "app",
@@ -82,7 +80,7 @@ func newBenchSamples(count int) []pushSample {
 			Name:  "app",
 			Value: fmt.Sprintf("instance-%d", i),
 		})
-		keyBuf = compressLabels(keyBuf[:0], &lc, labels[:labelsLen], labels[labelsLen:])
+		keyBuf = compressLabels(keyBuf[:0], labels[:labelsLen], labels[labelsLen:])
 		sample.key = string(keyBuf)
 		sample.value = float64(i)
 	}