Repository: VictoriaMetrics/VictoriaMetrics
Pull request #4676 (2023-07-20 11:49:39)

vmui: improvements for the Active Queries
- The `id` column has been removed
- The `from`, `to`, and `step` columns have been combined into a single `args` column.
- The `remote_addr` column has been renamed to `client address`.
- The columns have been reordered as follows: `duration`, `client address`, `query`, `args`.
- The table has been sorted in descending order based on the `duration` column.

https://github.com/VictoriaMetrics/VictoriaMetrics/pull/4653#issuecomment-1642888619


Patch:
diff --git a/.gitignore b/.gitignore
index f06c7eeea..96d97c523 100644
--- a/.gitignore
+++ b/.gitignore
@@ -8,11 +8,11 @@
 *.test
 *.swp
 /gocache-for-docker
+/victoria-logs-data
 /victoria-metrics-data
 /vmagent-remotewrite-data
 /vmstorage-data
 /vmselect-cache
-/victoria-logs-data
 /package/temp-deb-*
 /package/temp-rpm-*
 /package/*.deb
diff --git a/README.md b/README.md
index df089f4e9..9ce300325 100644
--- a/README.md
+++ b/README.md
@@ -401,6 +401,11 @@ matching the specified [series selector](https://prometheus.io/docs/prometheus/l
 
 Cardinality explorer is built on top of [/api/v1/status/tsdb](#tsdb-stats).
 
+In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html) each vmstorage tracks the stored time series individually.
+vmselect requests stats via [/api/v1/status/tsdb](#tsdb-stats) API from each vmstorage node and merges the results by summing per-series stats.
+This may lead to inflated values when samples for the same time series are spread across multiple vmstorage nodes
+due to [replication](#replication) or [rerouting](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html?highlight=re-routes#cluster-availability).
+
 See [cardinality explorer playground](https://play.victoriametrics.com/select/accounting/1/6a716b0f-38bc-4856-90ce-448fd713e3fe/prometheus/graph/#/cardinality).
 See the example of using the cardinality explorer [here](https://victoriametrics.com/blog/cardinality-explorer/).
 
@@ -1668,7 +1673,8 @@ Retention filters can be evaluated for free by downloading and using enterprise
 
 * `-downsampling.period=30d:5m,180d:1h` instructs VictoriaMetrics to deduplicate samples older than 30 days with 5 minutes interval and to deduplicate samples older than 180 days with 1 hour interval.
 
-Downsampling is applied independently per each time series. It can reduce disk space usage and improve query performance if it is applied to time series with big number of samples per each series. The downsampling doesn't improve query performance if the database contains big number of time series with small number of samples per each series (aka [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate)), since downsampling doesn't reduce the number of time series. So the majority of time is spent on searching for the matching time series. It is possible to use recording rules in [vmalert](https://docs.victoriametrics.com/vmalert.html) in order to reduce the number of time series. See [these docs](https://docs.victoriametrics.com/vmalert.html#downsampling-and-aggregation-via-vmalert).
+Downsampling is applied independently per each time series. It can reduce disk space usage and improve query performance if it is applied to time series with big number of samples per each series. The downsampling doesn't improve query performance if the database contains big number of time series with small number of samples per each series (aka [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate)), since downsampling doesn't reduce the number of time series. So the majority of time is spent on searching for the matching time series. 
+It is possible to use [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html) in vmagent or recording rules in [vmalert](https://docs.victoriametrics.com/vmalert.html) in order to [reduce the number of time series](https://docs.victoriametrics.com/vmalert.html#downsampling-and-aggregation-via-vmalert).
 
 Downsampling happens during [background merges](https://docs.victoriametrics.com/#storage) 
 and can't be performed if there is not enough of free disk space or if vmstorage 
@@ -1783,6 +1789,11 @@ VictoriaMetrics returns TSDB stats at `/api/v1/status/tsdb` page in the way simi
 * `match[]=SELECTOR` where `SELECTOR` is an arbitrary [time series selector](https://prometheus.io/docs/prometheus/latest/querying/basics/#time-series-selectors) for series to take into account during stats calculation. By default all the series are taken into account.
 * `extra_label=LABEL=VALUE`. See [these docs](#prometheus-querying-api-enhancements) for more details.
 
+In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html) each vmstorage tracks the stored time series individually.
+vmselect requests stats via [/api/v1/status/tsdb](#tsdb-stats) API from each vmstorage node and merges the results by summing per-series stats.
+This may lead to inflated values when samples for the same time series are spread across multiple vmstorage nodes
+due to [replication](#replication) or [rerouting](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html?highlight=re-routes#cluster-availability).
+
 VictoriaMetrics provides an UI on top of `/api/v1/status/tsdb` - see [cardinality explorer docs](#cardinality-explorer).
 
 ## Query tracing
diff --git a/app/vlinsert/elasticsearch/elasticsearch.go b/app/vlinsert/elasticsearch/elasticsearch.go
index 7f1a22956..c11d8b398 100644
--- a/app/vlinsert/elasticsearch/elasticsearch.go
+++ b/app/vlinsert/elasticsearch/elasticsearch.go
@@ -199,12 +199,15 @@ func readBulkLine(sc *bufio.Scanner, timeField, msgField string,
 		return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
 	}
 
-	timestamp, err := extractTimestampFromFields(timeField, p.Fields)
+	ts, err := extractTimestampFromFields(timeField, p.Fields)
 	if err != nil {
 		return false, fmt.Errorf("cannot parse timestamp: %w", err)
 	}
+	if ts == 0 {
+		ts = time.Now().UnixNano()
+	}
 	p.RenameField(msgField, "_msg")
-	processLogMessage(timestamp, p.Fields)
+	processLogMessage(ts, p.Fields)
 	logjson.PutParser(p)
 	return true, nil
 }
@@ -222,10 +225,15 @@ func extractTimestampFromFields(timeField string, fields []logstorage.Field) (in
 		f.Value = ""
 		return timestamp, nil
 	}
-	return time.Now().UnixNano(), nil
+	return 0, nil
 }
 
 func parseElasticsearchTimestamp(s string) (int64, error) {
+	if s == "0" || s == "" {
+		// Special case - zero or empty timestamp must be substituted
+		// with the current time by the caller.
+		return 0, nil
+	}
 	if len(s) < len("YYYY-MM-DD") || s[len("YYYY")] != '-' {
 		// Try parsing timestamp in milliseconds
 		n, err := strconv.ParseInt(s, 10, 64)
diff --git a/app/vlinsert/jsonline/jsonline.go b/app/vlinsert/jsonline/jsonline.go
index 373bc0129..1b901db36 100644
--- a/app/vlinsert/jsonline/jsonline.go
+++ b/app/vlinsert/jsonline/jsonline.go
@@ -99,12 +99,15 @@ func readLine(sc *bufio.Scanner, timeField, msgField string, processLogMessage f
 	if err := p.ParseLogMessage(line); err != nil {
 		return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
 	}
-	timestamp, err := extractTimestampFromFields(timeField, p.Fields)
+	ts, err := extractTimestampFromFields(timeField, p.Fields)
 	if err != nil {
 		return false, fmt.Errorf("cannot parse timestamp: %w", err)
 	}
+	if ts == 0 {
+		ts = time.Now().UnixNano()
+	}
 	p.RenameField(msgField, "_msg")
-	processLogMessage(timestamp, p.Fields)
+	processLogMessage(ts, p.Fields)
 	logjson.PutParser(p)
 	return true, nil
 }
@@ -122,10 +125,15 @@ func extractTimestampFromFields(timeField string, fields []logstorage.Field) (in
 		f.Value = ""
 		return timestamp, nil
 	}
-	return time.Now().UnixNano(), nil
+	return 0, nil
 }
 
 func parseISO8601Timestamp(s string) (int64, error) {
+	if s == "0" || s == "" {
+		// Special case for returning the current timestamp.
+		// It must be automatically converted to the current timestamp by the caller.
+		return 0, nil
+	}
 	t, err := time.Parse(time.RFC3339, s)
 	if err != nil {
 		return 0, fmt.Errorf("cannot parse timestamp %q: %w", s, err)
diff --git a/app/vlinsert/loki/loki.go b/app/vlinsert/loki/loki.go
index b7450b0b5..878616328 100644
--- a/app/vlinsert/loki/loki.go
+++ b/app/vlinsert/loki/loki.go
@@ -4,35 +4,32 @@ import (
 	"net/http"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/metrics"
 )
 
-const msgField = "_msg"
-
 var (
-	lokiRequestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push"}`)
+	lokiRequestsJSONTotal     = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="json"}`)
+	lokiRequestsProtobufTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
 )
 
-// RequestHandler processes ElasticSearch insert requests
+// RequestHandler processes Loki insert requests
+//
+// See https://grafana.com/docs/loki/latest/api/#push-log-entries-to-loki
 func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
-	switch path {
-	case "/api/v1/push":
-		contentType := r.Header.Get("Content-Type")
-		lokiRequestsTotal.Inc()
-		switch contentType {
-		case "application/x-protobuf":
-			return handleProtobuf(r, w)
-		case "application/json", "gzip":
-			return handleJSON(r, w)
-		default:
-			logger.Warnf("unsupported Content-Type=%q for %q request; skipping it", contentType, path)
-			return false
-		}
-	default:
+	if path != "/api/v1/push" {
 		return false
 	}
+	contentType := r.Header.Get("Content-Type")
+	switch contentType {
+	case "application/json":
+		lokiRequestsJSONTotal.Inc()
+		return handleJSON(r, w)
+	default:
+		// Protobuf request body should be handled by default accoring to https://grafana.com/docs/loki/latest/api/#push-log-entries-to-loki
+		lokiRequestsProtobufTotal.Inc()
+		return handleProtobuf(r, w)
+	}
 }
 
 func getCommonParams(r *http.Request) (*insertutils.CommonParams, error) {
diff --git a/app/vlinsert/loki/loki_json.go b/app/vlinsert/loki/loki_json.go
index 874f92cf4..5ace832b0 100644
--- a/app/vlinsert/loki/loki_json.go
+++ b/app/vlinsert/loki/loki_json.go
@@ -6,127 +6,185 @@ import (
 	"math"
 	"net/http"
 	"strconv"
+	"time"
 
-	"github.com/valyala/fastjson"
-
-	"github.com/VictoriaMetrics/metrics"
-
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/metrics"
+	"github.com/valyala/fastjson"
 )
 
 var (
-	rowsIngestedTotalJSON = metrics.NewCounter(`vl_rows_ingested_total{type="loki", format="json"}`)
+	rowsIngestedJSONTotal = metrics.NewCounter(`vl_rows_ingested_total{type="loki",format="json"}`)
 	parserPool            fastjson.ParserPool
 )
 
 func handleJSON(r *http.Request, w http.ResponseWriter) bool {
-	contentType := r.Header.Get("Content-Type")
 	reader := r.Body
-	if contentType == "gzip" {
+	if r.Header.Get("Content-Encoding") == "gzip" {
 		zr, err := common.GetGzipReader(reader)
 		if err != nil {
-			httpserver.Errorf(w, r, "cannot read gzipped request: %s", err)
+			httpserver.Errorf(w, r, "cannot initialize gzip reader: %s", err)
 			return true
 		}
 		defer common.PutGzipReader(zr)
 		reader = zr
 	}
 
+	wcr := writeconcurrencylimiter.GetReader(reader)
+	data, err := io.ReadAll(wcr)
+	writeconcurrencylimiter.PutReader(wcr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot read request body: %s", err)
+		return true
+	}
+
 	cp, err := getCommonParams(r)
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot parse request: %s", err)
+		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return true
 	}
 	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields)
-	defer logstorage.PutLogRows(lr)
-
 	processLogMessage := cp.GetProcessLogMessageFunc(lr)
-	n, err := processJSONRequest(reader, processLogMessage)
+	n, err := parseJSONRequest(data, processLogMessage)
+	vlstorage.MustAddRows(lr)
+	logstorage.PutLogRows(lr)
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot decode loki request: %s", err)
+		httpserver.Errorf(w, r, "cannot parse Loki request: %s", err)
 		return true
 	}
-	rowsIngestedTotalJSON.Add(n)
+	rowsIngestedJSONTotal.Add(n)
 	return true
 }
 
-func processJSONRequest(r io.Reader, processLogMessage func(timestamp int64, fields []logstorage.Field)) (int, error) {
-	wcr := writeconcurrencylimiter.GetReader(r)
-	defer writeconcurrencylimiter.PutReader(wcr)
-
-	bytes, err := io.ReadAll(wcr)
+func parseJSONRequest(data []byte, processLogMessage func(timestamp int64, fields []logstorage.Field)) (int, error) {
+	p := parserPool.Get()
+	defer parserPool.Put(p)
+	v, err := p.ParseBytes(data)
 	if err != nil {
-		return 0, fmt.Errorf("cannot read request body: %w", err)
+		return 0, fmt.Errorf("cannot parse JSON request body: %w", err)
 	}
 
-	p := parserPool.Get()
-	defer parserPool.Put(p)
-	v, err := p.ParseBytes(bytes)
+	streamsV := v.Get("streams")
+	if streamsV == nil {
+		return 0, fmt.Errorf("missing `streams` item in the parsed JSON: %q", v)
+	}
+	streams, err := streamsV.Array()
 	if err != nil {
-		return 0, fmt.Errorf("cannot parse request body: %w", err)
+		return 0, fmt.Errorf("`streams` item in the parsed JSON must contain an array; got %q", streamsV)
 	}
 
+	currentTimestamp := time.Now().UnixNano()
 	var commonFields []logstorage.Field
 	rowsIngested := 0
-	for stIdx, st := range v.GetArray("streams") {
-		// `stream` contains labels for the stream.
-		// Labels are same for all entries in the stream.
-		logFields := st.GetObject("stream")
-		if logFields == nil {
-			logger.Warnf("missing streams field from %q", st)
-			logFields = &fastjson.Object{}
+	for _, stream := range streams {
+		// populate common labels from `stream` dict
+		commonFields = commonFields[:0]
+		labelsV := stream.Get("stream")
+		var labels *fastjson.Object
+		if labelsV != nil {
+			o, err := labelsV.Object()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("`stream` item in the parsed JSON must contain an object; got %q", labelsV)
+			}
+			labels = o
 		}
-		commonFields = slicesutil.ResizeNoCopyMayOverallocate(commonFields, logFields.Len()+1)
-		i := 0
-		logFields.Visit(func(k []byte, v *fastjson.Value) {
-			sfName := bytesutil.ToUnsafeString(k)
-			sfValue := bytesutil.ToUnsafeString(v.GetStringBytes())
-			commonFields[i].Name = sfName
-			commonFields[i].Value = sfValue
-			i++
+		labels.Visit(func(k []byte, v *fastjson.Value) {
+			if err != nil {
+				return
+			}
+			vStr, errLocal := v.StringBytes()
+			if errLocal != nil {
+				err = fmt.Errorf("unexpected label value type for %q:%q; want string", k, v)
+				return
+			}
+			commonFields = append(commonFields, logstorage.Field{
+				Name:  bytesutil.ToUnsafeString(k),
+				Value: bytesutil.ToUnsafeString(vStr),
+			})
 		})
-		msgFieldIdx := logFields.Len()
-		commonFields[msgFieldIdx].Name = msgField
+		if err != nil {
+			return rowsIngested, fmt.Errorf("error when parsing `stream` object: %w", err)
+		}
 
-		for idx, v := range st.GetArray("values") {
-			vs := v.GetArray()
-			if len(vs) != 2 {
-				return rowsIngested, fmt.Errorf("unexpected number of values in stream %d line %d: %q; got %d; want %d", stIdx, idx, v, len(vs), 2)
+		// populate messages from `values` array
+		linesV := stream.Get("values")
+		if linesV == nil {
+			return rowsIngested, fmt.Errorf("missing `values` item in the parsed JSON %q", stream)
+		}
+		lines, err := linesV.Array()
+		if err != nil {
+			return rowsIngested, fmt.Errorf("`values` item in the parsed JSON must contain an array; got %q", linesV)
+		}
+
+		fields := commonFields
+		for _, line := range lines {
+			lineA, err := line.Array()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("unexpected contents of `values` item; want array; got %q", line)
+			}
+			if len(lineA) != 2 {
+				return rowsIngested, fmt.Errorf("unexpected number of values in `values` item array %q; got %d want 2", line, len(lineA))
 			}
 
-			tsString := bytesutil.ToUnsafeString(vs[0].GetStringBytes())
-			ts, err := parseLokiTimestamp(tsString)
+			// parse timestamp
+			timestamp, err := lineA[0].StringBytes()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("unexpected log timestamp type for %q; want string", lineA[0])
+			}
+			ts, err := parseLokiTimestamp(bytesutil.ToUnsafeString(timestamp))
 			if err != nil {
-				return rowsIngested, fmt.Errorf("cannot parse timestamp in stream %d line %d: %q: %s", stIdx, idx, vs, err)
+				return rowsIngested, fmt.Errorf("cannot parse log timestamp %q: %w", timestamp, err)
+			}
+			if ts == 0 {
+				ts = currentTimestamp
 			}
 
-			commonFields[msgFieldIdx].Value = bytesutil.ToUnsafeString(vs[1].GetStringBytes())
-			processLogMessage(ts, commonFields)
+			// parse log message
+			msg, err := lineA[1].StringBytes()
+			if err != nil {
+				return rowsIngested, fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
+			}
+
+			fields = append(fields[:len(commonFields)], logstorage.Field{
+				Name:  "_msg",
+				Value: bytesutil.ToUnsafeString(msg),
+			})
+			processLogMessage(ts, fields)
 
-			rowsIngested++
 		}
+		rowsIngested += len(lines)
 	}
 
 	return rowsIngested, nil
 }
 
 func parseLokiTimestamp(s string) (int64, error) {
-	// Parsing timestamp in nanoseconds
+	if s == "" {
+		// Special case - an empty timestamp must be substituted with the current time by the caller.
+		return 0, nil
+	}
 	n, err := strconv.ParseInt(s, 10, 64)
 	if err != nil {
-		return 0, fmt.Errorf("cannot parse timestamp in nanoseconds from %q: %w", s, err)
-	}
-	if n > int64(math.MaxInt64) {
-		return 0, fmt.Errorf("too big timestamp in nanoseconds: %d; mustn't exceed %d", n, math.MaxInt64)
+		// Fall back to parsing floating-point value
+		f, err := strconv.ParseFloat(s, 64)
+		if err != nil {
+			return 0, err
+		}
+		if f > math.MaxInt64 {
+			return 0, fmt.Errorf("too big timestamp in nanoseconds: %v; mustn't exceed %v", f, math.MaxInt64)
+		}
+		if f < math.MinInt64 {
+			return 0, fmt.Errorf("too small timestamp in nanoseconds: %v; must be bigger or equal to %v", f, math.MinInt64)
+		}
+		n = int64(f)
 	}
 	if n < 0 {
-		return 0, fmt.Errorf("too small timestamp in nanoseconds: %d; must be bigger than %d", n, 0)
+		return 0, fmt.Errorf("too small timestamp in nanoseconds: %d; must be bigger than 0", n)
 	}
 	return n, nil
 }
diff --git a/app/vlinsert/loki/loki_protobuf.go b/app/vlinsert/loki/loki_protobuf.go
index db88a40c0..a75bf93d0 100644
--- a/app/vlinsert/loki/loki_protobuf.go
+++ b/app/vlinsert/loki/loki_protobuf.go
@@ -4,67 +4,66 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
+	"strings"
 	"sync"
+	"time"
 
-	"github.com/golang/snappy"
-
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
-	"github.com/VictoriaMetrics/metricsql"
+	"github.com/golang/snappy"
 )
 
 var (
-	rowsIngestedTotalProtobuf = metrics.NewCounter(`vl_rows_ingested_total{type="loki", format="protobuf"}`)
+	rowsIngestedProtobufTotal = metrics.NewCounter(`vl_rows_ingested_total{type="loki",format="protobuf"}`)
 	bytesBufPool              bytesutil.ByteBufferPool
 	pushReqsPool              sync.Pool
 )
 
 func handleProtobuf(r *http.Request, w http.ResponseWriter) bool {
 	wcr := writeconcurrencylimiter.GetReader(r.Body)
-	defer writeconcurrencylimiter.PutReader(wcr)
+	data, err := io.ReadAll(wcr)
+	writeconcurrencylimiter.PutReader(wcr)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot read request body: %s", err)
+		return true
+	}
 
 	cp, err := getCommonParams(r)
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot parse request: %s", err)
+		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return true
 	}
 	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields)
-	defer logstorage.PutLogRows(lr)
-
 	processLogMessage := cp.GetProcessLogMessageFunc(lr)
-	n, err := processProtobufRequest(wcr, processLogMessage)
+	n, err := parseProtobufRequest(data, processLogMessage)
+	vlstorage.MustAddRows(lr)
+	logstorage.PutLogRows(lr)
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot decode loki request: %s", err)
+		httpserver.Errorf(w, r, "cannot parse loki request: %s", err)
 		return true
 	}
-
-	rowsIngestedTotalProtobuf.Add(n)
-
+	rowsIngestedProtobufTotal.Add(n)
 	return true
 }
 
-func processProtobufRequest(r io.Reader, processLogMessage func(timestamp int64, fields []logstorage.Field)) (int, error) {
-	wcr := writeconcurrencylimiter.GetReader(r)
-	defer writeconcurrencylimiter.PutReader(wcr)
-
-	bytes, err := io.ReadAll(wcr)
-	if err != nil {
-		return 0, fmt.Errorf("cannot read request body: %s", err)
-	}
-
+func parseProtobufRequest(data []byte, processLogMessage func(timestamp int64, fields []logstorage.Field)) (int, error) {
 	bb := bytesBufPool.Get()
 	defer bytesBufPool.Put(bb)
-	bb.B, err = snappy.Decode(bb.B[:cap(bb.B)], bytes)
+
+	buf, err := snappy.Decode(bb.B[:cap(bb.B)], data)
 	if err != nil {
-		return 0, fmt.Errorf("cannot decode snappy from request body: %s", err)
+		return 0, fmt.Errorf("cannot decode snappy-encoded request body: %w", err)
 	}
+	bb.B = buf
+
+	req := getPushRequest()
+	defer putPushRequest(req)
 
-	req := getPushReq()
-	defer putPushReq(req)
 	err = req.Unmarshal(bb.B)
 	if err != nil {
 		return 0, fmt.Errorf("cannot parse request body: %s", err)
@@ -72,54 +71,93 @@ func processProtobufRequest(r io.Reader, processLogMessage func(timestamp int64,
 
 	var commonFields []logstorage.Field
 	rowsIngested := 0
-	for stIdx, st := range req.Streams {
+	streams := req.Streams
+	currentTimestamp := time.Now().UnixNano()
+	for i := range streams {
+		stream := &streams[i]
 		// st.Labels contains labels for the stream.
 		// Labels are same for all entries in the stream.
-		commonFields, err = parseLogFields(st.Labels, commonFields)
+		commonFields, err = parsePromLabels(commonFields[:0], stream.Labels)
 		if err != nil {
-			return rowsIngested, fmt.Errorf("failed to unmarshal labels in stream %d: %q; %s", stIdx, st.Labels, err)
+			return rowsIngested, fmt.Errorf("cannot parse stream labels %q: %s", stream.Labels, err)
 		}
-		msgFieldIDx := len(commonFields) - 1
-		commonFields[msgFieldIDx].Name = msgField
-
-		for _, v := range st.Entries {
-			commonFields[msgFieldIDx].Value = v.Line
-			processLogMessage(v.Timestamp.UnixNano(), commonFields)
-			rowsIngested++
+		fields := commonFields
+
+		entries := stream.Entries
+		for j := range entries {
+			entry := &entries[j]
+			fields = append(fields[:len(commonFields)], logstorage.Field{
+				Name:  "_msg",
+				Value: entry.Line,
+			})
+			ts := entry.Timestamp.UnixNano()
+			if ts == 0 {
+				ts = currentTimestamp
+			}
+			processLogMessage(ts, fields)
 		}
+		rowsIngested += len(stream.Entries)
 	}
 	return rowsIngested, nil
 }
 
-// Parses logs fields s and returns the corresponding log fields.
-// Cannot use searchutils.ParseMetricSelector here because its dependencies
-// bring flags which clashes with logstorage flags.
+// parsePromLabels parses log fields in Prometheus text exposition format from s, appends them to dst and returns the result.
 //
-// Loki encodes labels in the PromQL labels format.
 // See test data of promtail for examples: https://github.com/grafana/loki/blob/a24ef7b206e0ca63ee74ca6ecb0a09b745cd2258/pkg/push/types_test.go
-func parseLogFields(s string, dst []logstorage.Field) ([]logstorage.Field, error) {
-	expr, err := metricsql.Parse(s)
-	if err != nil {
-		return nil, err
+func parsePromLabels(dst []logstorage.Field, s string) ([]logstorage.Field, error) {
+	// Make sure s is wrapped into `{...}`
+	s = strings.TrimSpace(s)
+	if len(s) < 2 {
+		return nil, fmt.Errorf("too short string to parse: %q", s)
 	}
-
-	me, ok := expr.(*metricsql.MetricExpr)
-	if !ok {
-		return nil, fmt.Errorf("failed to parse stream labels; got %q", expr.AppendString(nil))
+	if s[0] != '{' {
+		return nil, fmt.Errorf("missing `{` at the beginning of %q", s)
 	}
-
-	// Allocate space for labels + msg field.
-	// Msg field is added by caller.
-	dst = slicesutil.ResizeNoCopyMayOverallocate(dst, len(me.LabelFilters)+1)
-	for i, l := range me.LabelFilters {
-		dst[i].Name = l.Label
-		dst[i].Value = l.Value
+	if s[len(s)-1] != '}' {
+		return nil, fmt.Errorf("missing `}` at the end of %q", s)
 	}
+	s = s[1 : len(s)-1]
 
+	for len(s) > 0 {
+		// Parse label name
+		n := strings.IndexByte(s, '=')
+		if n < 0 {
+			return nil, fmt.Errorf("cannot find `=` char for label value at %s", s)
+		}
+		name := s[:n]
+		s = s[n+1:]
+
+		// Parse label value
+		qs, err := strconv.QuotedPrefix(s)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse value for label %q at %s: %w", name, s, err)
+		}
+		s = s[len(qs):]
+		value, err := strconv.Unquote(qs)
+		if err != nil {
+			return nil, fmt.Errorf("cannot unquote value %q for label %q: %w", qs, name, err)
+		}
+
+		// Append the found field to dst.
+		dst = append(dst, logstorage.Field{
+			Name:  name,
+			Value: value,
+		})
+
+		// Check whether there are other labels remaining
+		if len(s) == 0 {
+			break
+		}
+		if !strings.HasPrefix(s, ",") {
+			return nil, fmt.Errorf("missing `,` char at %s", s)
+		}
+		s = s[1:]
+		s = strings.TrimPrefix(s, " ")
+	}
 	return dst, nil
 }
 
-func getPushReq() *PushRequest {
+func getPushRequest() *PushRequest {
 	v := pushReqsPool.Get()
 	if v == nil {
 		return &PushRequest{}
@@ -127,7 +165,7 @@ func getPushReq() *PushRequest {
 	return v.(*PushRequest)
 }
 
-func putPushReq(reqs *PushRequest) {
-	reqs.Reset()
-	pushReqsPool.Put(reqs)
+func putPushRequest(req *PushRequest) {
+	req.Reset()
+	pushReqsPool.Put(req)
 }
diff --git a/app/vmalert/Makefile b/app/vmalert/Makefile
index e3a970967..4a27eac1c 100644
--- a/app/vmalert/Makefile
+++ b/app/vmalert/Makefile
@@ -74,6 +74,7 @@ test-vmalert:
 	go test -v -race -cover ./app/vmalert/config
 	go test -v -race -cover ./app/vmalert/remotewrite
 	go test -v -race -cover ./app/vmalert/utils
+	go test -v -race -cover ./app/vmalert/unittest
 
 run-vmalert: vmalert
 	./bin/vmalert -rule=app/vmalert/config/testdata/rules/rules2-good.rules \
@@ -102,6 +103,10 @@ replay-vmalert: vmalert
 		-replay.timeFrom=2021-05-11T07:21:43Z \
 		-replay.timeTo=2021-05-29T18:40:43Z
 
+unittest-vmalert: vmalert
+	./bin/vmalert -unittestFile=app/vmalert/unittest/testdata/test1.yaml \
+ 		-unittestFile=app/vmalert/unittest/testdata/test2.yaml
+
 vmalert-linux-amd64:
 	APP_NAME=vmalert CGO_ENABLED=1 GOOS=linux GOARCH=amd64 $(MAKE) app-local-goos-goarch
 
diff --git a/app/vmalert/README.md b/app/vmalert/README.md
index b0b3bdebd..544576d97 100644
--- a/app/vmalert/README.md
+++ b/app/vmalert/README.md
@@ -732,6 +732,248 @@ See full description for these flags in `./vmalert -help`.
 * `query` template function is disabled for performance reasons (might be changed in future);
 * `limit` group's param has no effect during replay (might be changed in future);
 
+## Unit Testing for Rules
+
+> Unit testing is available from v1.92.0. 
+> Unit tests do not respect `-clusterMode` for now.
+
+You can use `vmalert` to run unit tests for alerting and recording rules.
+In unit test mode vmalert performs the following actions:
+* sets up an isolated VictoriaMetrics instance;
+* simulates the periodic ingestion of time series;
+* queries the ingested data for recording and alerting rules evaluation;
+* tests whether the firing alerts or resulting recording rules match the expected results. 
+
+See how to run vmalert in unit test mode below:
+```
+# Run vmalert with one or multiple test files via -unittestFile cmd-line flag
+./vmalert -unittestFile=test1.yaml -unittestFile=test2.yaml
+```
+
+vmalert is compatible with [Prometheus config format for tests](https://prometheus.io/docs/prometheus/latest/configuration/unit_testing_rules/#test-file-format)
+except `promql_expr_test` field. Use `metricsql_expr_test` field name instead. The name is different because vmalert 
+validates and executes [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html) expressions, 
+which aren't always backward compatible with [PromQL](https://prometheus.io/docs/prometheus/latest/querying/basics/).
+
+### Test file format
+
+The configuration format for files specified in `-unittestFile` cmd-line flag is the following:
+```
+# Path to the files or http url containing [rule groups](https://docs.victoriametrics.com/vmalert.html#groups) configuration.
+# Enterprise version of vmalert supports S3 and GCS paths to rules.
+rule_files:
+  [ - <string> ]
+
+# The evaluation interval for rules specified in `rule_files`
+[ evaluation_interval: <duration> | default = 1m ]
+
+# Groups listed below will be evaluated by order.
+# Not All the groups need not be mentioned, if not, they will be evaluated by define order in rule_files.
+group_eval_order:
+  [ - <string> ]
+
+# The list of unit test files to be checked during evaluation.
+tests:
+  [ - <test_group> ]
+```
+
+#### `<test_group>`
+
+```
+# Interval between samples for input series
+interval: <duration>
+# Time series to persist into the database according to configured <interval> before running tests.
+input_series:
+  [ - <series> ]
+
+# Name of the test group, optional
+[ name: <string> ]
+
+# Unit tests for alerting rules
+alert_rule_test:
+  [ - <alert_test_case> ]
+
+# Unit tests for Metricsql expressions.
+metricsql_expr_test:
+  [ - <metricsql_expr_test> ]
+
+# External labels accessible for templating.
+external_labels:
+  [ <labelname>: <string> ... ]
+
+```
+
+#### `<series>`
+
+```
+# series in the following format '<metric name>{<label name>=<label value>, ...}'
+# Examples:
+#      series_name{label1="value1", label2="value2"}
+#      go_goroutines{job="prometheus", instance="localhost:9090"}
+series: <string>
+
+# values support several special equations:
+#     'a+bxc' becomes 'a a+b a+(2*b) a+(3*b) … a+(c*b)'
+#     Read this as series starts at a, then c further samples incrementing by b.
+#     'a-bxc' becomes 'a a-b a-(2*b) a-(3*b) … a-(c*b)'
+#     Read this as series starts at a, then c further samples decrementing by b (or incrementing by negative b).
+#    '_' represents a missing sample from scrape
+# Examples:
+#     1. '-2+4x3' becomes '-2 2 6 10' - series starts at -2, then 3 further samples incrementing by 4.
+#     2. ' 1-2x4' becomes '1 -1 -3 -5 -7' - series starts at 1, then 4 further samples decrementing by 2.
+#     3. ' 1x4' becomes '1 1 1 1 1' - shorthand for '1+0x4', series starts at 1, then 4 further samples incrementing by 0.
+#     4. ' 1 _x3' becomes '1 _ _ _ ' - the missing sample cannot increment, so 3 missing samples are produced by the '_x3' expression.
+values: <string>
+```
+
+#### `<alert_test_case>`
+
+vmalert by default adds `alertgroup` and `alertname` to the generated alerts and time series. 
+So you will need to specify both `groupname` and `alertname` under a single `<alert_test_case>`, 
+but no need to add them under `exp_alerts`.
+You can also pass `--disableAlertgroupLabel` to prevent vmalert from adding `alertgroup` label.
+
+```
+# The time elapsed from time=0s when this alerting rule should be checked.
+# Means this rule should be firing at this point, or shouldn't be firing if 'exp_alerts' is empty.
+eval_time: <duration>
+
+# Name of the group name to be tested.
+groupname: <string>
+
+# Name of the alert to be tested.
+alertname: <string>
+
+# List of the expected alerts that are firing under the given alertname at
+# the given evaluation time. If you want to test if an alerting rule should
+# not be firing, then you can mention only the fields above and leave 'exp_alerts' empty.
+exp_alerts:
+  [ - <alert> ]
+```
+
+#### `<alert>`
+
+```
+# These are the expanded labels and annotations of the expected alert.
+# Note: labels also include the labels of the sample associated with the alert
+exp_labels:
+  [ <labelname>: <string> ]
+exp_annotations:
+  [ <labelname>: <string> ]
+```
+
+#### `<metricsql_expr_test>`
+
+```
+# Expression to evaluate
+expr: <string>
+
+# The time elapsed from time=0s when this expression be evaluated.
+eval_time: <duration>
+
+# Expected samples at the given evaluation time.
+exp_samples:
+  [ - <sample> ]
+```
+
+#### `<sample>`
+
+```
+# Labels of the sample in usual series notation '<metric name>{<label name>=<label value>, ...}'
+# Examples:
+#      series_name{label1="value1", label2="value2"}
+#      go_goroutines{job="prometheus", instance="localhost:9090"}
+labels: <string>
+
+# The expected value of the Metricsql expression.
+value: <number>
+```
+
+### Example
+
+This is an example input file for unit testing which will pass.
+`test.yaml` is the test file which follows the syntax above and `alerts.yaml` contains the alerting rules.
+
+With `rules.yaml` in the same directory, run `./vmalert -unittestFile=./unittest/testdata/test.yaml`.
+
+#### `test.yaml`
+
+```
+rule_files:
+  - rules.yaml
+
+evaluation_interval: 1m
+
+tests:
+  - interval: 1m
+    input_series:
+      - series: 'up{job="prometheus", instance="localhost:9090"}'
+        values: "0+0x1440"
+
+    metricsql_expr_test:
+      - expr: suquery_interval_test
+        eval_time: 4m
+        exp_samples:
+          - labels: '{__name__="suquery_interval_test", datacenter="dc-123", instance="localhost:9090", job="prometheus"}'
+            value: 1
+
+    alert_rule_test:
+      - eval_time: 2h
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts:
+          - exp_labels:
+              job: prometheus
+              severity: page
+              instance: localhost:9090
+              datacenter: dc-123
+            exp_annotations:
+              summary: "Instance localhost:9090 down"
+              description: "localhost:9090 of job prometheus has been down for more than 5 minutes."
+
+      - eval_time: 0
+        groupname: group1
+        alertname: AlwaysFiring
+        exp_alerts:
+          - exp_labels:
+              datacenter: dc-123
+
+      - eval_time: 0
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts: []
+
+    external_labels:
+      datacenter: dc-123
+```
+
+#### `alerts.yaml`
+
+```
+# This is the rules file.
+
+groups:
+  - name: group1
+    rules:
+      - alert: InstanceDown
+        expr: up == 0
+        for: 5m
+        labels:
+          severity: page
+        annotations:
+          summary: "Instance {{ $labels.instance }} down"
+          description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 5 minutes."
+      - alert: AlwaysFiring
+        expr: 1
+
+  - name: group2
+    rules:
+      - record: job:test:count_over_time1m
+        expr: sum without(instance) (count_over_time(test[1m]))
+      - record: suquery_interval_test
+        expr: count_over_time(up[5m:])
+```
+
 ## Monitoring
 
 `vmalert` exports various metrics in Prometheus exposition format at `http://vmalert-host:8880/metrics` page.
@@ -1282,6 +1524,11 @@ The shortlist of configuration flags is the following:
      Path to file with TLS key if -tls is set. The provided key file is automatically re-read every second, so it can be dynamically updated
   -tlsMinVersion string
      Optional minimum TLS version to use for incoming requests over HTTPS if -tls is set. Supported values: TLS10, TLS11, TLS12, TLS13
+  -unittestFile array
+     Path to the unit test files. When set, vmalert starts in unit test mode and performs only tests on configured files.
+     Examples:
+      -unittestFile="./unittest/testdata/test1.yaml,./unittest/testdata/test2.yaml".
+     See more information here https://docs.victoriametrics.com/vmalert.html#unit-testing-for-rules.
   -version
      Show VictoriaMetrics version
 ```
diff --git a/app/vmalert/datasource/datasource.go b/app/vmalert/datasource/datasource.go
index bae112d14..2e90e1848 100644
--- a/app/vmalert/datasource/datasource.go
+++ b/app/vmalert/datasource/datasource.go
@@ -1,10 +1,14 @@
 package datasource
 
 import (
+	"bytes"
 	"context"
 	"net/http"
 	"net/url"
+	"strconv"
 	"time"
+
+	"golang.org/x/exp/slices"
 )
 
 // Querier interface wraps Query and QueryRange methods
@@ -108,3 +112,69 @@ type Label struct {
 	Name  string
 	Value string
 }
+
+// Labels is collection of Label
+type Labels []Label
+
+func (ls Labels) Len() int           { return len(ls) }
+func (ls Labels) Swap(i, j int)      { ls[i], ls[j] = ls[j], ls[i] }
+func (ls Labels) Less(i, j int) bool { return ls[i].Name < ls[j].Name }
+
+func (ls Labels) String() string {
+	var b bytes.Buffer
+
+	b.WriteByte('{')
+	for i, l := range ls {
+		if i > 0 {
+			b.WriteByte(',')
+			b.WriteByte(' ')
+		}
+		b.WriteString(l.Name)
+		b.WriteByte('=')
+		b.WriteString(strconv.Quote(l.Value))
+	}
+	b.WriteByte('}')
+	return b.String()
+}
+
+// LabelCompare return negative if a is less than b, return 0 if they are the same
+// eg.
+// a=[]Label{{Name: "a", Value: "1"}},b=[]Label{{Name: "b", Value: "1"}}, return -1
+// a=[]Label{{Name: "a", Value: "2"}},b=[]Label{{Name: "a", Value: "1"}}, return 1
+// a=[]Label{{Name: "a", Value: "1"}},b=[]Label{{Name: "a", Value: "1"}}, return 0
+func LabelCompare(a, b Labels) int {
+	l := len(a)
+	if len(b) < l {
+		l = len(b)
+	}
+
+	for i := 0; i < l; i++ {
+		if a[i].Name != b[i].Name {
+			if a[i].Name < b[i].Name {
+				return -1
+			}
+			return 1
+		}
+		if a[i].Value != b[i].Value {
+			if a[i].Value < b[i].Value {
+				return -1
+			}
+			return 1
+		}
+	}
+	// if all labels so far were in common, the set with fewer labels comes first.
+	return len(a) - len(b)
+}
+
+// ConvertToLabels convert map to Labels
+func ConvertToLabels(m map[string]string) (labelset Labels) {
+	for k, v := range m {
+		labelset = append(labelset, Label{
+			Name:  k,
+			Value: v,
+		})
+	}
+	// sort label
+	slices.SortFunc(labelset, func(a, b Label) bool { return a.Name < b.Name })
+	return
+}
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index 3ccc01269..8ee17f78b 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -274,7 +274,7 @@ func (g *Group) close() {
 
 var skipRandSleepOnGroupStart bool
 
-func (g *Group) start(ctx context.Context, nts func() []notifier.Notifier, rw *remotewrite.Client, rr datasource.QuerierBuilder) {
+func (g *Group) start(ctx context.Context, nts func() []notifier.Notifier, rw remotewrite.RWClient, rr datasource.QuerierBuilder) {
 	defer func() { close(g.finishedCh) }()
 
 	// Spread group rules evaluation over time in order to reduce load on VictoriaMetrics.
@@ -422,7 +422,7 @@ type executor struct {
 	notifiers       func() []notifier.Notifier
 	notifierHeaders map[string]string
 
-	rw *remotewrite.Client
+	rw remotewrite.RWClient
 
 	previouslySentSeriesToRWMu sync.Mutex
 	// previouslySentSeriesToRW stores series sent to RW on previous iteration
diff --git a/app/vmalert/main.go b/app/vmalert/main.go
index 912daefa4..8c7c0bd7e 100644
--- a/app/vmalert/main.go
+++ b/app/vmalert/main.go
@@ -91,7 +91,12 @@ absolute path to all .tpl files in root.
 
 	disableAlertGroupLabel = flag.Bool("disableAlertgroupLabel", false, "Whether to disable adding group's Name as label to generated alerts and time series.")
 
-	dryRun = flag.Bool("dryRun", false, "Whether to check only config files without running vmalert. The rules file are validated. The -rule flag must be specified.")
+	dryRun        = flag.Bool("dryRun", false, "Whether to check only config files without running vmalert. The rules file are validated. The -rule flag must be specified.")
+	unitTestFiles = flagutil.NewArrayString("unittestFile", `Path to the unit test files. When set, vmalert starts in unit test mode and performs only tests on configured files.
+Examples:
+ -unittestFile="./unittest/testdata/test1.yaml,./unittest/testdata/test2.yaml".
+See more information here https://docs.victoriametrics.com/vmalert.html#unit-testing-for-rules.
+`)
 )
 
 var alertURLGeneratorFn notifier.AlertURLGenerator
@@ -117,6 +122,13 @@ func main() {
 		logger.Fatalf("failed to parse %q: %s", *ruleTemplatesPath, err)
 	}
 
+	if len(*unitTestFiles) > 0 {
+		if unitRule(*unitTestFiles...) {
+			os.Exit(1)
+		}
+		os.Exit(0)
+	}
+
 	if *dryRun {
 		groups, err := config.Parse(*rulePath, notifier.ValidateTemplates, true)
 		if err != nil {
@@ -399,7 +411,7 @@ func configsEqual(a, b []config.Group) bool {
 // setConfigSuccess sets config reload status to 1.
 func setConfigSuccess(at uint64) {
 	configSuccess.Set(1)
-	configTimestamp.Set(fasttime.UnixTimestamp())
+	configTimestamp.Set(at)
 	// reset the error if any
 	setConfigErr(nil)
 }
diff --git a/app/vmalert/manager.go b/app/vmalert/manager.go
index 26e909ddf..d97a96e2c 100644
--- a/app/vmalert/manager.go
+++ b/app/vmalert/manager.go
@@ -19,7 +19,7 @@ type manager struct {
 	querierBuilder datasource.QuerierBuilder
 	notifiers      func() []notifier.Notifier
 
-	rw *remotewrite.Client
+	rw remotewrite.RWClient
 	// remote read builder.
 	rr datasource.QuerierBuilder
 
diff --git a/app/vmalert/remotewrite/client.go b/app/vmalert/remotewrite/client.go
new file mode 100644
index 000000000..8d4f935c1
--- /dev/null
+++ b/app/vmalert/remotewrite/client.go
@@ -0,0 +1,320 @@
+package remotewrite
+
+import (
+	"bytes"
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"path"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/golang/snappy"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	disablePathAppend = flag.Bool("remoteWrite.disablePathAppend", false, "Whether to disable automatic appending of '/api/v1/write' path to the configured -remoteWrite.url.")
+	sendTimeout       = flag.Duration("remoteWrite.sendTimeout", 30*time.Second, "Timeout for sending data to the configured -remoteWrite.url.")
+	retryMinInterval  = flag.Duration("remoteWrite.retryMinInterval", time.Second, "The minimum delay between retry attempts. Every next retry attempt will double the delay to prevent hammering of remote database. See also -remoteWrite.retryMaxInterval")
+	retryMaxTime      = flag.Duration("remoteWrite.retryMaxTime", time.Second*30, "The max time spent on retry attempts for the failed remote-write request. Change this value if it is expected for remoteWrite.url to be unreachable for more than -remoteWrite.retryMaxTime. See also -remoteWrite.retryMinInterval")
+)
+
+// Client is an asynchronous HTTP client for writing
+// timeseries via remote write protocol.
+type Client struct {
+	addr          string
+	c             *http.Client
+	authCfg       *promauth.Config
+	input         chan prompbmarshal.TimeSeries
+	flushInterval time.Duration
+	maxBatchSize  int
+	maxQueueSize  int
+
+	wg     sync.WaitGroup
+	doneCh chan struct{}
+}
+
+// Config is config for remote write.
+type Config struct {
+	// Addr of remote storage
+	Addr    string
+	AuthCfg *promauth.Config
+
+	// Concurrency defines number of readers that
+	// concurrently read from the queue and flush data
+	Concurrency int
+	// MaxBatchSize defines max number of timeseries
+	// to be flushed at once
+	MaxBatchSize int
+	// MaxQueueSize defines max length of input queue
+	// populated by Push method.
+	// Push will be rejected once queue is full.
+	MaxQueueSize int
+	// FlushInterval defines time interval for flushing batches
+	FlushInterval time.Duration
+	// Transport will be used by the underlying http.Client
+	Transport *http.Transport
+}
+
+const (
+	defaultConcurrency   = 4
+	defaultMaxBatchSize  = 1e3
+	defaultMaxQueueSize  = 1e5
+	defaultFlushInterval = 5 * time.Second
+	defaultWriteTimeout  = 30 * time.Second
+)
+
+// NewClient returns asynchronous client for
+// writing timeseries via remotewrite protocol.
+func NewClient(ctx context.Context, cfg Config) (*Client, error) {
+	if cfg.Addr == "" {
+		return nil, fmt.Errorf("config.Addr can't be empty")
+	}
+	if cfg.MaxBatchSize == 0 {
+		cfg.MaxBatchSize = defaultMaxBatchSize
+	}
+	if cfg.MaxQueueSize == 0 {
+		cfg.MaxQueueSize = defaultMaxQueueSize
+	}
+	if cfg.FlushInterval == 0 {
+		cfg.FlushInterval = defaultFlushInterval
+	}
+	if cfg.Transport == nil {
+		cfg.Transport = http.DefaultTransport.(*http.Transport).Clone()
+	}
+	cc := defaultConcurrency
+	if cfg.Concurrency > 0 {
+		cc = cfg.Concurrency
+	}
+	c := &Client{
+		c: &http.Client{
+			Timeout:   *sendTimeout,
+			Transport: cfg.Transport,
+		},
+		addr:          strings.TrimSuffix(cfg.Addr, "/"),
+		authCfg:       cfg.AuthCfg,
+		flushInterval: cfg.FlushInterval,
+		maxBatchSize:  cfg.MaxBatchSize,
+		maxQueueSize:  cfg.MaxQueueSize,
+		doneCh:        make(chan struct{}),
+		input:         make(chan prompbmarshal.TimeSeries, cfg.MaxQueueSize),
+	}
+
+	for i := 0; i < cc; i++ {
+		c.run(ctx)
+	}
+	return c, nil
+}
+
+// Push adds timeseries into queue for writing into remote storage.
+// Push returns and error if client is stopped or if queue is full.
+func (c *Client) Push(s prompbmarshal.TimeSeries) error {
+	select {
+	case <-c.doneCh:
+		return fmt.Errorf("client is closed")
+	case c.input <- s:
+		return nil
+	default:
+		return fmt.Errorf("failed to push timeseries - queue is full (%d entries). "+
+			"Queue size is controlled by -remoteWrite.maxQueueSize flag",
+			c.maxQueueSize)
+	}
+}
+
+// Close stops the client and waits for all goroutines
+// to exit.
+func (c *Client) Close() error {
+	if c.doneCh == nil {
+		return fmt.Errorf("client is already closed")
+	}
+	close(c.input)
+	close(c.doneCh)
+	c.wg.Wait()
+	return nil
+}
+
+func (c *Client) run(ctx context.Context) {
+	ticker := time.NewTicker(c.flushInterval)
+	wr := &prompbmarshal.WriteRequest{}
+	shutdown := func() {
+		for ts := range c.input {
+			wr.Timeseries = append(wr.Timeseries, ts)
+		}
+		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
+		logger.Infof("shutting down remote write client and flushing remained %d series", len(wr.Timeseries))
+		c.flush(lastCtx, wr)
+		cancel()
+	}
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		defer ticker.Stop()
+		for {
+			select {
+			case <-c.doneCh:
+				shutdown()
+				return
+			case <-ctx.Done():
+				shutdown()
+				return
+			case <-ticker.C:
+				c.flush(ctx, wr)
+			case ts, ok := <-c.input:
+				if !ok {
+					continue
+				}
+				wr.Timeseries = append(wr.Timeseries, ts)
+				if len(wr.Timeseries) >= c.maxBatchSize {
+					c.flush(ctx, wr)
+				}
+			}
+		}
+	}()
+}
+
+var (
+	sentRows            = metrics.NewCounter(`vmalert_remotewrite_sent_rows_total`)
+	sentBytes           = metrics.NewCounter(`vmalert_remotewrite_sent_bytes_total`)
+	sendDuration        = metrics.NewFloatCounter(`vmalert_remotewrite_send_duration_seconds_total`)
+	droppedRows         = metrics.NewCounter(`vmalert_remotewrite_dropped_rows_total`)
+	droppedBytes        = metrics.NewCounter(`vmalert_remotewrite_dropped_bytes_total`)
+	bufferFlushDuration = metrics.NewHistogram(`vmalert_remotewrite_flush_duration_seconds`)
+
+	_ = metrics.NewGauge(`vmalert_remotewrite_concurrency`, func() float64 {
+		return float64(*concurrency)
+	})
+)
+
+// flush is a blocking function that marshals WriteRequest and sends
+// it to remote-write endpoint. Flush performs limited amount of retries
+// if request fails.
+func (c *Client) flush(ctx context.Context, wr *prompbmarshal.WriteRequest) {
+	if len(wr.Timeseries) < 1 {
+		return
+	}
+	defer prompbmarshal.ResetWriteRequest(wr)
+	defer bufferFlushDuration.UpdateDuration(time.Now())
+
+	data, err := wr.Marshal()
+	if err != nil {
+		logger.Errorf("failed to marshal WriteRequest: %s", err)
+		return
+	}
+
+	b := snappy.Encode(nil, data)
+
+	retryInterval, maxRetryInterval := *retryMinInterval, *retryMaxTime
+	if retryInterval > maxRetryInterval {
+		retryInterval = maxRetryInterval
+	}
+	timeStart := time.Now()
+	defer sendDuration.Add(time.Since(timeStart).Seconds())
+L:
+	for attempts := 0; ; attempts++ {
+		err := c.send(ctx, b)
+		if err == nil {
+			sentRows.Add(len(wr.Timeseries))
+			sentBytes.Add(len(b))
+			return
+		}
+
+		_, isNotRetriable := err.(*nonRetriableError)
+		logger.Warnf("attempt %d to send request failed: %s (retriable: %v)", attempts+1, err, !isNotRetriable)
+
+		if isNotRetriable {
+			// exit fast if error isn't retriable
+			break
+		}
+
+		// check if request has been cancelled before backoff
+		select {
+		case <-ctx.Done():
+			logger.Errorf("interrupting retry attempt %d: context cancelled", attempts+1)
+			break L
+		default:
+		}
+
+		timeLeftForRetries := maxRetryInterval - time.Since(timeStart)
+		if timeLeftForRetries < 0 {
+			// the max retry time has passed, so we give up
+			break
+		}
+
+		if retryInterval > timeLeftForRetries {
+			retryInterval = timeLeftForRetries
+		}
+		// sleeping to prevent remote db hammering
+		time.Sleep(retryInterval)
+		retryInterval *= 2
+
+	}
+
+	droppedRows.Add(len(wr.Timeseries))
+	droppedBytes.Add(len(b))
+	logger.Errorf("attempts to send remote-write request failed - dropping %d time series",
+		len(wr.Timeseries))
+}
+
+func (c *Client) send(ctx context.Context, data []byte) error {
+	r := bytes.NewReader(data)
+	req, err := http.NewRequest(http.MethodPost, c.addr, r)
+	if err != nil {
+		return fmt.Errorf("failed to create new HTTP request: %w", err)
+	}
+
+	// RFC standard compliant headers
+	req.Header.Set("Content-Encoding", "snappy")
+	req.Header.Set("Content-Type", "application/x-protobuf")
+
+	// Prometheus compliant headers
+	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+
+	if c.authCfg != nil {
+		c.authCfg.SetHeaders(req, true)
+	}
+	if !*disablePathAppend {
+		req.URL.Path = path.Join(req.URL.Path, "/api/v1/write")
+	}
+	resp, err := c.c.Do(req.WithContext(ctx))
+	if err != nil {
+		return fmt.Errorf("error while sending request to %s: %w; Data len %d(%d)",
+			req.URL.Redacted(), err, len(data), r.Size())
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	body, _ := io.ReadAll(resp.Body)
+
+	// according to https://prometheus.io/docs/concepts/remote_write_spec/
+	// Prometheus remote Write compatible receivers MUST
+	switch resp.StatusCode / 100 {
+	case 2:
+		// respond with a HTTP 2xx status code when the write is successful.
+		return nil
+	case 4:
+		if resp.StatusCode != http.StatusTooManyRequests {
+			// MUST NOT retry write requests on HTTP 4xx responses other than 429
+			return &nonRetriableError{fmt.Errorf("unexpected response code %d for %s. Response body %q",
+				resp.StatusCode, req.URL.Redacted(), body)}
+		}
+		fallthrough
+	default:
+		return fmt.Errorf("unexpected response code %d for %s. Response body %q",
+			resp.StatusCode, req.URL.Redacted(), body)
+	}
+}
+
+type nonRetriableError struct {
+	err error
+}
+
+func (e *nonRetriableError) Error() string {
+	return e.err.Error()
+}
diff --git a/app/vmalert/remotewrite/remotewrite_test.go b/app/vmalert/remotewrite/client_test.go
similarity index 100%
rename from app/vmalert/remotewrite/remotewrite_test.go
rename to app/vmalert/remotewrite/client_test.go
diff --git a/app/vmalert/remotewrite/debug_client.go b/app/vmalert/remotewrite/debug_client.go
new file mode 100644
index 000000000..a3cd17282
--- /dev/null
+++ b/app/vmalert/remotewrite/debug_client.go
@@ -0,0 +1,97 @@
+package remotewrite
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"path"
+	"strings"
+	"sync"
+
+	"github.com/golang/snappy"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+)
+
+// DebugClient won't push series periodically, but will write data to remote endpoint
+// immediately when Push() is called
+type DebugClient struct {
+	addr string
+	c    *http.Client
+
+	wg sync.WaitGroup
+}
+
+// NewDebugClient initiates and returns a new DebugClient
+func NewDebugClient() (*DebugClient, error) {
+	if *addr == "" {
+		return nil, nil
+	}
+
+	t, err := utils.Transport(*addr, *tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transport: %w", err)
+	}
+	c := &DebugClient{
+		c: &http.Client{
+			Timeout:   *sendTimeout,
+			Transport: t,
+		},
+		addr: strings.TrimSuffix(*addr, "/"),
+	}
+	return c, nil
+}
+
+// Push sends the given timeseries to the remote storage.
+func (c *DebugClient) Push(s prompbmarshal.TimeSeries) error {
+	c.wg.Add(1)
+	defer c.wg.Done()
+	wr := &prompbmarshal.WriteRequest{Timeseries: []prompbmarshal.TimeSeries{s}}
+	data, err := wr.Marshal()
+	if err != nil {
+		return fmt.Errorf("failed to marshal the given time series: %w", err)
+	}
+
+	return c.send(data)
+}
+
+// Close stops the DebugClient
+func (c *DebugClient) Close() error {
+	c.wg.Wait()
+	return nil
+}
+
+func (c *DebugClient) send(data []byte) error {
+	b := snappy.Encode(nil, data)
+	r := bytes.NewReader(b)
+	req, err := http.NewRequest(http.MethodPost, c.addr, r)
+	if err != nil {
+		return fmt.Errorf("failed to create new HTTP request: %w", err)
+	}
+
+	// RFC standard compliant headers
+	req.Header.Set("Content-Encoding", "snappy")
+	req.Header.Set("Content-Type", "application/x-protobuf")
+
+	// Prometheus compliant headers
+	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+
+	if !*disablePathAppend {
+		req.URL.Path = path.Join(req.URL.Path, "/api/v1/write")
+	}
+	resp, err := c.c.Do(req)
+	if err != nil {
+		return fmt.Errorf("error while sending request to %s: %w; Data len %d(%d)",
+			req.URL.Redacted(), err, len(data), r.Size())
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode/100 == 2 {
+		return nil
+	}
+	body, _ := io.ReadAll(resp.Body)
+	return fmt.Errorf("unexpected response code %d for %s. Response body %q",
+		resp.StatusCode, req.URL.Redacted(), body)
+}
diff --git a/app/vmalert/remotewrite/remotewrite.go b/app/vmalert/remotewrite/remotewrite.go
index 8d4f935c1..27d505e48 100644
--- a/app/vmalert/remotewrite/remotewrite.go
+++ b/app/vmalert/remotewrite/remotewrite.go
@@ -1,320 +1,13 @@
 package remotewrite
 
 import (
-	"bytes"
-	"context"
-	"flag"
-	"fmt"
-	"io"
-	"net/http"
-	"path"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/golang/snappy"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/metrics"
-)
-
-var (
-	disablePathAppend = flag.Bool("remoteWrite.disablePathAppend", false, "Whether to disable automatic appending of '/api/v1/write' path to the configured -remoteWrite.url.")
-	sendTimeout       = flag.Duration("remoteWrite.sendTimeout", 30*time.Second, "Timeout for sending data to the configured -remoteWrite.url.")
-	retryMinInterval  = flag.Duration("remoteWrite.retryMinInterval", time.Second, "The minimum delay between retry attempts. Every next retry attempt will double the delay to prevent hammering of remote database. See also -remoteWrite.retryMaxInterval")
-	retryMaxTime      = flag.Duration("remoteWrite.retryMaxTime", time.Second*30, "The max time spent on retry attempts for the failed remote-write request. Change this value if it is expected for remoteWrite.url to be unreachable for more than -remoteWrite.retryMaxTime. See also -remoteWrite.retryMinInterval")
-)
-
-// Client is an asynchronous HTTP client for writing
-// timeseries via remote write protocol.
-type Client struct {
-	addr          string
-	c             *http.Client
-	authCfg       *promauth.Config
-	input         chan prompbmarshal.TimeSeries
-	flushInterval time.Duration
-	maxBatchSize  int
-	maxQueueSize  int
-
-	wg     sync.WaitGroup
-	doneCh chan struct{}
-}
-
-// Config is config for remote write.
-type Config struct {
-	// Addr of remote storage
-	Addr    string
-	AuthCfg *promauth.Config
-
-	// Concurrency defines number of readers that
-	// concurrently read from the queue and flush data
-	Concurrency int
-	// MaxBatchSize defines max number of timeseries
-	// to be flushed at once
-	MaxBatchSize int
-	// MaxQueueSize defines max length of input queue
-	// populated by Push method.
-	// Push will be rejected once queue is full.
-	MaxQueueSize int
-	// FlushInterval defines time interval for flushing batches
-	FlushInterval time.Duration
-	// Transport will be used by the underlying http.Client
-	Transport *http.Transport
-}
-
-const (
-	defaultConcurrency   = 4
-	defaultMaxBatchSize  = 1e3
-	defaultMaxQueueSize  = 1e5
-	defaultFlushInterval = 5 * time.Second
-	defaultWriteTimeout  = 30 * time.Second
 )
 
-// NewClient returns asynchronous client for
-// writing timeseries via remotewrite protocol.
-func NewClient(ctx context.Context, cfg Config) (*Client, error) {
-	if cfg.Addr == "" {
-		return nil, fmt.Errorf("config.Addr can't be empty")
-	}
-	if cfg.MaxBatchSize == 0 {
-		cfg.MaxBatchSize = defaultMaxBatchSize
-	}
-	if cfg.MaxQueueSize == 0 {
-		cfg.MaxQueueSize = defaultMaxQueueSize
-	}
-	if cfg.FlushInterval == 0 {
-		cfg.FlushInterval = defaultFlushInterval
-	}
-	if cfg.Transport == nil {
-		cfg.Transport = http.DefaultTransport.(*http.Transport).Clone()
-	}
-	cc := defaultConcurrency
-	if cfg.Concurrency > 0 {
-		cc = cfg.Concurrency
-	}
-	c := &Client{
-		c: &http.Client{
-			Timeout:   *sendTimeout,
-			Transport: cfg.Transport,
-		},
-		addr:          strings.TrimSuffix(cfg.Addr, "/"),
-		authCfg:       cfg.AuthCfg,
-		flushInterval: cfg.FlushInterval,
-		maxBatchSize:  cfg.MaxBatchSize,
-		maxQueueSize:  cfg.MaxQueueSize,
-		doneCh:        make(chan struct{}),
-		input:         make(chan prompbmarshal.TimeSeries, cfg.MaxQueueSize),
-	}
-
-	for i := 0; i < cc; i++ {
-		c.run(ctx)
-	}
-	return c, nil
-}
-
-// Push adds timeseries into queue for writing into remote storage.
-// Push returns and error if client is stopped or if queue is full.
-func (c *Client) Push(s prompbmarshal.TimeSeries) error {
-	select {
-	case <-c.doneCh:
-		return fmt.Errorf("client is closed")
-	case c.input <- s:
-		return nil
-	default:
-		return fmt.Errorf("failed to push timeseries - queue is full (%d entries). "+
-			"Queue size is controlled by -remoteWrite.maxQueueSize flag",
-			c.maxQueueSize)
-	}
-}
-
-// Close stops the client and waits for all goroutines
-// to exit.
-func (c *Client) Close() error {
-	if c.doneCh == nil {
-		return fmt.Errorf("client is already closed")
-	}
-	close(c.input)
-	close(c.doneCh)
-	c.wg.Wait()
-	return nil
-}
-
-func (c *Client) run(ctx context.Context) {
-	ticker := time.NewTicker(c.flushInterval)
-	wr := &prompbmarshal.WriteRequest{}
-	shutdown := func() {
-		for ts := range c.input {
-			wr.Timeseries = append(wr.Timeseries, ts)
-		}
-		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
-		logger.Infof("shutting down remote write client and flushing remained %d series", len(wr.Timeseries))
-		c.flush(lastCtx, wr)
-		cancel()
-	}
-	c.wg.Add(1)
-	go func() {
-		defer c.wg.Done()
-		defer ticker.Stop()
-		for {
-			select {
-			case <-c.doneCh:
-				shutdown()
-				return
-			case <-ctx.Done():
-				shutdown()
-				return
-			case <-ticker.C:
-				c.flush(ctx, wr)
-			case ts, ok := <-c.input:
-				if !ok {
-					continue
-				}
-				wr.Timeseries = append(wr.Timeseries, ts)
-				if len(wr.Timeseries) >= c.maxBatchSize {
-					c.flush(ctx, wr)
-				}
-			}
-		}
-	}()
-}
-
-var (
-	sentRows            = metrics.NewCounter(`vmalert_remotewrite_sent_rows_total`)
-	sentBytes           = metrics.NewCounter(`vmalert_remotewrite_sent_bytes_total`)
-	sendDuration        = metrics.NewFloatCounter(`vmalert_remotewrite_send_duration_seconds_total`)
-	droppedRows         = metrics.NewCounter(`vmalert_remotewrite_dropped_rows_total`)
-	droppedBytes        = metrics.NewCounter(`vmalert_remotewrite_dropped_bytes_total`)
-	bufferFlushDuration = metrics.NewHistogram(`vmalert_remotewrite_flush_duration_seconds`)
-
-	_ = metrics.NewGauge(`vmalert_remotewrite_concurrency`, func() float64 {
-		return float64(*concurrency)
-	})
-)
-
-// flush is a blocking function that marshals WriteRequest and sends
-// it to remote-write endpoint. Flush performs limited amount of retries
-// if request fails.
-func (c *Client) flush(ctx context.Context, wr *prompbmarshal.WriteRequest) {
-	if len(wr.Timeseries) < 1 {
-		return
-	}
-	defer prompbmarshal.ResetWriteRequest(wr)
-	defer bufferFlushDuration.UpdateDuration(time.Now())
-
-	data, err := wr.Marshal()
-	if err != nil {
-		logger.Errorf("failed to marshal WriteRequest: %s", err)
-		return
-	}
-
-	b := snappy.Encode(nil, data)
-
-	retryInterval, maxRetryInterval := *retryMinInterval, *retryMaxTime
-	if retryInterval > maxRetryInterval {
-		retryInterval = maxRetryInterval
-	}
-	timeStart := time.Now()
-	defer sendDuration.Add(time.Since(timeStart).Seconds())
-L:
-	for attempts := 0; ; attempts++ {
-		err := c.send(ctx, b)
-		if err == nil {
-			sentRows.Add(len(wr.Timeseries))
-			sentBytes.Add(len(b))
-			return
-		}
-
-		_, isNotRetriable := err.(*nonRetriableError)
-		logger.Warnf("attempt %d to send request failed: %s (retriable: %v)", attempts+1, err, !isNotRetriable)
-
-		if isNotRetriable {
-			// exit fast if error isn't retriable
-			break
-		}
-
-		// check if request has been cancelled before backoff
-		select {
-		case <-ctx.Done():
-			logger.Errorf("interrupting retry attempt %d: context cancelled", attempts+1)
-			break L
-		default:
-		}
-
-		timeLeftForRetries := maxRetryInterval - time.Since(timeStart)
-		if timeLeftForRetries < 0 {
-			// the max retry time has passed, so we give up
-			break
-		}
-
-		if retryInterval > timeLeftForRetries {
-			retryInterval = timeLeftForRetries
-		}
-		// sleeping to prevent remote db hammering
-		time.Sleep(retryInterval)
-		retryInterval *= 2
-
-	}
-
-	droppedRows.Add(len(wr.Timeseries))
-	droppedBytes.Add(len(b))
-	logger.Errorf("attempts to send remote-write request failed - dropping %d time series",
-		len(wr.Timeseries))
-}
-
-func (c *Client) send(ctx context.Context, data []byte) error {
-	r := bytes.NewReader(data)
-	req, err := http.NewRequest(http.MethodPost, c.addr, r)
-	if err != nil {
-		return fmt.Errorf("failed to create new HTTP request: %w", err)
-	}
-
-	// RFC standard compliant headers
-	req.Header.Set("Content-Encoding", "snappy")
-	req.Header.Set("Content-Type", "application/x-protobuf")
-
-	// Prometheus compliant headers
-	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
-
-	if c.authCfg != nil {
-		c.authCfg.SetHeaders(req, true)
-	}
-	if !*disablePathAppend {
-		req.URL.Path = path.Join(req.URL.Path, "/api/v1/write")
-	}
-	resp, err := c.c.Do(req.WithContext(ctx))
-	if err != nil {
-		return fmt.Errorf("error while sending request to %s: %w; Data len %d(%d)",
-			req.URL.Redacted(), err, len(data), r.Size())
-	}
-	defer func() { _ = resp.Body.Close() }()
-
-	body, _ := io.ReadAll(resp.Body)
-
-	// according to https://prometheus.io/docs/concepts/remote_write_spec/
-	// Prometheus remote Write compatible receivers MUST
-	switch resp.StatusCode / 100 {
-	case 2:
-		// respond with a HTTP 2xx status code when the write is successful.
-		return nil
-	case 4:
-		if resp.StatusCode != http.StatusTooManyRequests {
-			// MUST NOT retry write requests on HTTP 4xx responses other than 429
-			return &nonRetriableError{fmt.Errorf("unexpected response code %d for %s. Response body %q",
-				resp.StatusCode, req.URL.Redacted(), body)}
-		}
-		fallthrough
-	default:
-		return fmt.Errorf("unexpected response code %d for %s. Response body %q",
-			resp.StatusCode, req.URL.Redacted(), body)
-	}
-}
-
-type nonRetriableError struct {
-	err error
-}
-
-func (e *nonRetriableError) Error() string {
-	return e.err.Error()
+// RWClient represents an HTTP client for pushing data via remote write protocol
+type RWClient interface {
+	// Push pushes the give time series to remote storage
+	Push(s prompbmarshal.TimeSeries) error
+	// Close stops the client. Client can't be reused after Close call.
+	Close() error
 }
diff --git a/app/vmalert/replay.go b/app/vmalert/replay.go
index 1f3763a24..1151c17bf 100644
--- a/app/vmalert/replay.go
+++ b/app/vmalert/replay.go
@@ -33,7 +33,7 @@ var (
 		"Progress bar rendering might be verbose or break the logs parsing, so it is recommended to be disabled when not used in interactive mode.")
 )
 
-func replay(groupsCfg []config.Group, qb datasource.QuerierBuilder, rw *remotewrite.Client) error {
+func replay(groupsCfg []config.Group, qb datasource.QuerierBuilder, rw remotewrite.RWClient) error {
 	if *replayMaxDatapoints < 1 {
 		return fmt.Errorf("replay.maxDatapointsPerQuery can't be lower than 1")
 	}
@@ -78,7 +78,7 @@ func replay(groupsCfg []config.Group, qb datasource.QuerierBuilder, rw *remotewr
 	return nil
 }
 
-func (g *Group) replay(start, end time.Time, rw *remotewrite.Client) int {
+func (g *Group) replay(start, end time.Time, rw remotewrite.RWClient) int {
 	var total int
 	step := g.Interval * time.Duration(*replayMaxDatapoints)
 	ri := rangeIterator{start: start, end: end, step: step}
@@ -119,7 +119,7 @@ func (g *Group) replay(start, end time.Time, rw *remotewrite.Client) int {
 	return total
 }
 
-func replayRule(rule Rule, start, end time.Time, rw *remotewrite.Client) (int, error) {
+func replayRule(rule Rule, start, end time.Time, rw remotewrite.RWClient) (int, error) {
 	var err error
 	var tss []prompbmarshal.TimeSeries
 	for i := 0; i < *replayRuleRetryAttempts; i++ {
diff --git a/app/vmctl/README.md b/app/vmctl/README.md
index d51b490a9..ff8c91752 100644
--- a/app/vmctl/README.md
+++ b/app/vmctl/README.md
@@ -725,6 +725,9 @@ requires an Authentication header like `X-Scope-OrgID`. You can define it via th
 
 ## Migrating data from VictoriaMetrics
 
+The simplest way to migrate data between VM instances is to copy data.
+See more details [here](https://docs.victoriametrics.com/Single-server-VictoriaMetrics.html#data-migration).
+
 vmctl uses [native binary protocol](https://docs.victoriametrics.com/#how-to-export-data-in-native-format)
 (available since [1.42.0 release](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.42.0))
 o migrate data between VM instances: single to single, cluster to cluster, single to cluster and vice versa.
diff --git a/app/vminsert/common/streamaggr.go b/app/vminsert/common/streamaggr.go
index af4fa4609..0a9da9451 100644
--- a/app/vminsert/common/streamaggr.go
+++ b/app/vminsert/common/streamaggr.go
@@ -30,7 +30,7 @@ var (
 )
 
 var (
-	saCfgReloaderStopCh = make(chan struct{})
+	saCfgReloaderStopCh chan struct{}
 	saCfgReloaderWG     sync.WaitGroup
 
 	saCfgReloads   = metrics.NewCounter(`vminsert_streamagg_config_reloads_total`)
@@ -59,6 +59,7 @@ func CheckStreamAggrConfig() error {
 //
 // MustStopStreamAggr must be called when stream aggr is no longer needed.
 func InitStreamAggr() {
+	saCfgReloaderStopCh = make(chan struct{})
 	if *streamAggrConfig == "" {
 		return
 	}
diff --git a/app/vmselect/main.go b/app/vmselect/main.go
index 2b3fc74c7..7f10095dd 100644
--- a/app/vmselect/main.go
+++ b/app/vmselect/main.go
@@ -329,7 +329,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	case "/api/v1/status/active_queries":
 		statusActiveQueriesRequests.Inc()
 		httpserver.EnableCORS(w, r)
-		promql.WriteActiveQueries(w, r)
+		promql.ActiveQueriesHandler(w, r)
 		return true
 	case "/api/v1/status/top_queries":
 		topQueriesRequests.Inc()
diff --git a/app/vmselect/promql/active_queries.go b/app/vmselect/promql/active_queries.go
index 8e0e22eb9..d66058e53 100644
--- a/app/vmselect/promql/active_queries.go
+++ b/app/vmselect/promql/active_queries.go
@@ -9,10 +9,10 @@ import (
 	"time"
 )
 
-// WriteActiveQueries writes active queries to w.
+// ActiveQueriesHandler returns response to /api/v1/status/active_queries
 //
-// The written active queries are sorted in descending order of their exeuction duration.
-func WriteActiveQueries(w http.ResponseWriter, r *http.Request) {
+// It writes a JSON with active queries to w.
+func ActiveQueriesHandler(w http.ResponseWriter, r *http.Request) {
 	aqes := activeQueriesV.GetAll()
 
 	w.Header().Set("Content-Type", "application/json")
diff --git a/app/vmui/packages/vmui/package-lock.json b/app/vmui/packages/vmui/package-lock.json
index ea99020c0..402e850f4 100644
--- a/app/vmui/packages/vmui/package-lock.json
+++ b/app/vmui/packages/vmui/package-lock.json
@@ -8,6 +8,7 @@
       "name": "vmui",
       "version": "0.1.0",
       "dependencies": {
+        "@monaco-editor/react": "^4.5.1",
         "@types/lodash.debounce": "^4.0.6",
         "@types/lodash.get": "^4.4.6",
         "@types/lodash.throttle": "^4.1.6",
@@ -26,7 +27,7 @@
         "preact": "^10.7.1",
         "qs": "^6.10.3",
         "react-input-mask": "^2.0.4",
-        "react-router-dom": "^6.3.0",
+        "react-router-dom": "^6.10.0",
         "sass": "^1.56.0",
         "source-map-explorer": "^2.5.3",
         "typescript": "~4.6.2",
@@ -3432,6 +3433,30 @@
       "dev": true,
       "peer": true
     },
+    "node_modules/@monaco-editor/loader": {
+      "version": "1.3.3",
+      "resolved": "https://registry.npmjs.org/@monaco-editor/loader/-/loader-1.3.3.tgz",
+      "integrity": "sha512-6KKF4CTzcJiS8BJwtxtfyYt9shBiEv32ateQ9T4UVogwn4HM/uPo9iJd2Dmbkpz8CM6Y0PDUpjnZzCwC+eYo2Q==",
+      "dependencies": {
+        "state-local": "^1.0.6"
+      },
+      "peerDependencies": {
+        "monaco-editor": ">= 0.21.0 < 1"
+      }
+    },
+    "node_modules/@monaco-editor/react": {
+      "version": "4.5.1",
+      "resolved": "https://registry.npmjs.org/@monaco-editor/react/-/react-4.5.1.tgz",
+      "integrity": "sha512-NNDFdP+2HojtNhCkRfE6/D6ro6pBNihaOzMbGK84lNWzRu+CfBjwzGt4jmnqimLuqp5yE5viHS2vi+QOAnD5FQ==",
+      "dependencies": {
+        "@monaco-editor/loader": "^1.3.3"
+      },
+      "peerDependencies": {
+        "monaco-editor": ">= 0.25.0 < 1",
+        "react": "^16.8.0 || ^17.0.0 || ^18.0.0",
+        "react-dom": "^16.8.0 || ^17.0.0 || ^18.0.0"
+      }
+    },
     "node_modules/@nicolo-ribaudo/eslint-scope-5-internals": {
       "version": "5.1.1-v1",
       "resolved": "https://registry.npmjs.org/@nicolo-ribaudo/eslint-scope-5-internals/-/eslint-scope-5-internals-5.1.1-v1.tgz",
@@ -13171,6 +13196,12 @@
         "mkdirp": "bin/cmd.js"
       }
     },
+    "node_modules/monaco-editor": {
+      "version": "0.40.0",
+      "resolved": "https://registry.npmjs.org/monaco-editor/-/monaco-editor-0.40.0.tgz",
+      "integrity": "sha512-1wymccLEuFSMBvCk/jT1YDW/GuxMLYwnFwF9CDyYCxoTw2Pt379J3FUhwy9c43j51JdcxVPjwk0jm0EVDsBS2g==",
+      "peer": true
+    },
     "node_modules/ms": {
       "version": "2.1.2",
       "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.2.tgz",
@@ -17066,6 +17097,11 @@
       "dev": true,
       "peer": true
     },
+    "node_modules/state-local": {
+      "version": "1.0.7",
+      "resolved": "https://registry.npmjs.org/state-local/-/state-local-1.0.7.tgz",
+      "integrity": "sha512-HTEHMNieakEnoe33shBYcZ7NX83ACUjCu8c40iOGEZsngj9zRnkqS9j1pqQPXwobB0ZcVTk27REb7COQ0UR59w=="
+    },
     "node_modules/statuses": {
       "version": "2.0.1",
       "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.1.tgz",
diff --git a/app/vmui/packages/vmui/package.json b/app/vmui/packages/vmui/package.json
index ff05078f7..12463f74f 100644
--- a/app/vmui/packages/vmui/package.json
+++ b/app/vmui/packages/vmui/package.json
@@ -4,6 +4,7 @@
   "private": true,
   "homepage": "./",
   "dependencies": {
+    "@monaco-editor/react": "^4.5.1",
     "@types/lodash.debounce": "^4.0.6",
     "@types/lodash.get": "^4.4.6",
     "@types/lodash.throttle": "^4.1.6",
diff --git a/app/vmui/packages/vmui/src/App.tsx b/app/vmui/packages/vmui/src/App.tsx
index 9736b0e9d..a5087f34d 100644
--- a/app/vmui/packages/vmui/src/App.tsx
+++ b/app/vmui/packages/vmui/src/App.tsx
@@ -14,7 +14,7 @@ import PreviewIcons from "./components/Main/Icons/PreviewIcons";
 import WithTemplate from "./pages/WithTemplate";
 import Relabel from "./pages/Relabel";
 import ExploreLogs from "./pages/ExploreLogs/ExploreLogs";
-import from "./pages/ActiveQueries";
+import ActiveQueries from "./pages/ActiveQueries";
 
 const App: FC = () => {
   const [loadedTheme, setLoadedTheme] = useState(false);
diff --git a/app/vmui/packages/vmui/src/api/expand-with-exprs.ts b/app/vmui/packages/vmui/src/api/expand-with-exprs.ts
index 8b0da7497..beb433a9c 100644
--- a/app/vmui/packages/vmui/src/api/expand-with-exprs.ts
+++ b/app/vmui/packages/vmui/src/api/expand-with-exprs.ts
@@ -1,2 +1,2 @@
 export const getExpandWithExprUrl = (server: string, query: string): string =>
-  `${server}/expand-with-exprs?query=${query}&format=json`;
+  `${server}/expand-with-exprs?query=${encodeURIComponent(query)}&format=json`;
diff --git a/app/vmui/packages/vmui/src/components/Configurators/QueryEditor/QueryEditor.tsx b/app/vmui/packages/vmui/src/components/Configurators/QueryEditor/QueryEditor.tsx
index cd77cc059..f0ef16c60 100644
--- a/app/vmui/packages/vmui/src/components/Configurators/QueryEditor/QueryEditor.tsx
+++ b/app/vmui/packages/vmui/src/components/Configurators/QueryEditor/QueryEditor.tsx
@@ -48,6 +48,9 @@ const QueryEditor: FC<QueryEditorProps> = ({
   const handleKeyDown = (e: KeyboardEvent) => {
     const { key, ctrlKey, metaKey, shiftKey } = e;
 
+    const value = (e.target as HTMLTextAreaElement).value || "";
+    const isMultiline = value.split("\n").length > 1;
+
     const ctrlMetaKey = ctrlKey || metaKey;
     const arrowUp = key === "ArrowUp";
     const arrowDown = key === "ArrowDown";
@@ -65,12 +68,21 @@ const QueryEditor: FC<QueryEditorProps> = ({
       onArrowDown();
     }
 
+    if (enter && openAutocomplete) {
+      e.preventDefault();
+    }
+
     // execute query
-    if (enter && !shiftKey && !openAutocomplete) {
+    if (enter && !shiftKey && (!isMultiline || ctrlMetaKey) && !openAutocomplete) {
+      e.preventDefault();
       onEnter();
     }
   };
 
+  const handleChangeFoundOptions = (val: string[]) => {
+    setOpenAutocomplete(!!val.length);
+  };
+
   return <div
     className="vm-query-editor"
     ref={autocompleteAnchorEl}
@@ -93,7 +105,7 @@ const QueryEditor: FC<QueryEditorProps> = ({
         options={options}
         anchor={autocompleteAnchorEl}
         onSelect={handleSelect}
-        onOpenAutocomplete={setOpenAutocomplete}
+        onFoundOptions={handleChangeFoundOptions}
       />
     )}
     {showSeriesFetchedWarning && (
diff --git a/app/vmui/packages/vmui/src/components/Main/Autocomplete/Autocomplete.tsx b/app/vmui/packages/vmui/src/components/Main/Autocomplete/Autocomplete.tsx
index a53feacf1..5f9a7d45f 100644
--- a/app/vmui/packages/vmui/src/components/Main/Autocomplete/Autocomplete.tsx
+++ b/app/vmui/packages/vmui/src/components/Main/Autocomplete/Autocomplete.tsx
@@ -21,6 +21,7 @@ interface AutocompleteProps {
   disabledFullScreen?: boolean
   onSelect: (val: string) => void
   onOpenAutocomplete?: (val: boolean) => void
+  onFoundOptions?: (val: string[]) => void
 }
 
 const Autocomplete: FC<AutocompleteProps> = ({
@@ -36,7 +37,8 @@ const Autocomplete: FC<AutocompleteProps> = ({
   label,
   disabledFullScreen,
   onSelect,
-  onOpenAutocomplete
+  onOpenAutocomplete,
+  onFoundOptions
 }) => {
   const { isMobile } = useDeviceDetect();
   const wrapperEl = useRef<HTMLDivElement>(null);
@@ -120,6 +122,10 @@ const Autocomplete: FC<AutocompleteProps> = ({
     onOpenAutocomplete && onOpenAutocomplete(openAutocomplete);
   }, [openAutocomplete]);
 
+  useEffect(() => {
+    onFoundOptions && onFoundOptions(foundOptions);
+  }, [foundOptions]);
+
   return (
     <Popper
       open={openAutocomplete}
diff --git a/app/vmui/packages/vmui/src/components/Main/TextField/TextField.tsx b/app/vmui/packages/vmui/src/components/Main/TextField/TextField.tsx
index e2ccdaf30..935b2ace0 100644
--- a/app/vmui/packages/vmui/src/components/Main/TextField/TextField.tsx
+++ b/app/vmui/packages/vmui/src/components/Main/TextField/TextField.tsx
@@ -59,9 +59,13 @@ const TextField: FC<TextFieldProps> = ({
 
   const handleKeyDown = (e: KeyboardEvent<HTMLInputElement | HTMLTextAreaElement>) => {
     onKeyDown && onKeyDown(e);
-    if (e.key === "Enter" && !e.shiftKey) {
+
+    const { key, ctrlKey, metaKey } = e;
+    const isEnter = key === "Enter";
+    const runByEnter = type !== "textarea" ? isEnter : isEnter && (metaKey || ctrlKey);
+    if (runByEnter && onEnter) {
       e.preventDefault();
-      onEnter && onEnter();
+      onEnter();
     }
   };
 
@@ -139,7 +143,8 @@ const TextField: FC<TextFieldProps> = ({
         {helperText}
       </span>
     )}
-  </label>;
+  </label>
+  ;
 };
 
 export default TextField;
diff --git a/app/vmui/packages/vmui/src/components/Main/TextField/style.scss b/app/vmui/packages/vmui/src/components/Main/TextField/style.scss
index fe686e015..70b463645 100644
--- a/app/vmui/packages/vmui/src/components/Main/TextField/style.scss
+++ b/app/vmui/packages/vmui/src/components/Main/TextField/style.scss
@@ -128,4 +128,13 @@
     left: auto;
     right: $padding-small;
   }
+
+  &__controls-info {
+    position: absolute;
+    bottom: $padding-small;
+    right: $padding-global;
+    color: $color-text-secondary;
+    font-size: $font-size-small;
+    opacity: 0.8;
+  }
 }
diff --git a/app/vmui/packages/vmui/src/components/MonacoEditor/MonacoEditor.tsx b/app/vmui/packages/vmui/src/components/MonacoEditor/MonacoEditor.tsx
new file mode 100644
index 000000000..cee3c1b84
--- /dev/null
+++ b/app/vmui/packages/vmui/src/components/MonacoEditor/MonacoEditor.tsx
@@ -0,0 +1,52 @@
+import React, { FC } from "preact/compat";
+import Editor, { useMonaco } from "@monaco-editor/react";
+import useMonacoTheme from "./hooks/useMonacoTheme";
+import useLabelsSyntax from "./hooks/useLabelsSyntax";
+import useKeybindings from "./hooks/useKeybindings";
+import "./style.scss";
+import classNames from "classnames";
+
+interface MonacoEditorProps {
+  value: string;
+  label?: string;
+  language?: string;
+  disabled?: boolean;
+  resize?: "vertical" | "horizontal" | "both" | "none";
+  onChange: (val: string | undefined) => void;
+  onEnter?: (val: string) => void;
+}
+
+const MonacoEditor: FC<MonacoEditorProps> = ({ value, label, language, disabled, resize = "none", onChange, onEnter }) => {
+  const monaco = useMonaco();
+  useMonacoTheme(monaco);
+  useLabelsSyntax(monaco);
+  useKeybindings(monaco, onEnter);
+
+  return (
+    <div className="vm-text-field vm-monaco-editor">
+      <Editor
+        className={classNames({
+          "vm-text-field__input": true,
+          "vm-monaco-editor__input": true,
+          [`vm-monaco-editor__input_resize-${resize}`]: resize,
+        })}
+        defaultLanguage={language}
+        value={value}
+        theme={"vm-theme"}
+        options={{
+          readOnly: disabled,
+          scrollBeyondLastLine: false,
+          automaticLayout: true,
+          lineNumbers: "off",
+          minimap: {
+            enabled: false
+          },
+        }}
+        onChange={onChange}
+      />
+      {label && <span className="vm-text-field__label">{label}</span>}
+    </div>
+  );
+};
+
+export default MonacoEditor;
diff --git a/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useKeybindings.ts b/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useKeybindings.ts
new file mode 100644
index 000000000..9a7a2fd40
--- /dev/null
+++ b/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useKeybindings.ts
@@ -0,0 +1,22 @@
+import { Monaco } from "@monaco-editor/react";
+import { useEffect } from "preact/compat";
+import * as monaco from "monaco-editor";
+
+const useKeybindings = (monaco: Monaco | null, onEnter?: (val: string) => void) => {
+
+  const handleRunEnter = (e: monaco.editor.ICodeEditor) => {
+    onEnter && onEnter(e.getValue() || "");
+  };
+
+  useEffect(() => {
+    if (!monaco) return;
+    monaco.editor.addEditorAction({
+      id: "execute-ctrl-enter",
+      label: "Execute",
+      keybindings: [monaco.KeyMod.CtrlCmd | monaco.KeyCode.Enter],
+      run: handleRunEnter
+    });
+  }, [monaco, onEnter]);
+};
+
+export default useKeybindings;
diff --git a/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useLabelsSyntax.ts b/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useLabelsSyntax.ts
new file mode 100644
index 000000000..967d65fdf
--- /dev/null
+++ b/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useLabelsSyntax.ts
@@ -0,0 +1,84 @@
+import { useEffect } from "preact/compat";
+import { Monaco } from "@monaco-editor/react";
+import * as monaco from "monaco-editor";
+
+const languageId = "vm-labels";
+
+export const language = {
+  ignoreCase: false,
+  defaultToken: "",
+  tokenizer: {
+    root: [
+      // labels
+      [ /[a-z_]\w*(?=\s*(=|!=|=~|!~))/, "tag" ],
+
+      // strings
+      [ /"([^"\\]|\\.)*$/, "string.invalid" ],
+      [ /'([^'\\]|\\.)*$/, "string.invalid" ],
+      [ /"/, "string", "@string_double" ],
+      [ /'/, "string", "@string_single" ],
+      [ /`/, "string", "@string_backtick" ],
+
+      // delimiters and operators
+      [ /[{}()[]]/, "@brackets" ],
+    ],
+
+    string_double: [
+      [ /[^\\"]+/, "string" ],
+      [ /\\./, "string.escape.invalid" ],
+      [ /"/, "string", "@pop" ]
+    ],
+
+    string_single: [
+      [ /[^\\']+/, "string" ],
+      [ /\\./, "string.escape.invalid" ],
+      [ /'/, "string", "@pop" ]
+    ],
+
+    string_backtick: [
+      [ /[^\\`$]+/, "string" ],
+      [ /\\./, "string.escape.invalid" ],
+      [ /`/, "string", "@pop" ]
+    ],
+  },
+} as monaco.languages.IMonarchLanguage;
+
+export const languageConfiguration: monaco.languages.LanguageConfiguration = {
+  wordPattern: /(-?\d*\.\d\w*)|([^`~!#%^&*()\-=+[{\]}\\|;:'",.<>/?\s]+)/g,
+  comments: {
+    lineComment: "#",
+  },
+  brackets: [
+    [ "{", "}" ],
+    [ "[", "]" ],
+    [ "(", ")" ],
+  ],
+  autoClosingPairs: [
+    { open: "{", close: "}" },
+    { open: "[", close: "]" },
+    { open: "(", close: ")" },
+    { open: "\"", close: "\"" },
+    { open: "'", close: "'" },
+  ],
+  surroundingPairs: [
+    { open: "{", close: "}" },
+    { open: "[", close: "]" },
+    { open: "(", close: ")" },
+    { open: "\"", close: "\"" },
+    { open: "'", close: "'" },
+    { open: "<", close: ">" },
+  ],
+  folding: {}
+};
+
+const useLabelsSyntax = (monaco: Monaco | null) => {
+
+  useEffect(() => {
+    if (!monaco) return;
+    monaco.languages.register({ id: languageId });
+    monaco.languages.setMonarchTokensProvider(languageId, language);
+    monaco.languages.setLanguageConfiguration(languageId, languageConfiguration);
+  }, [monaco]);
+};
+
+export default useLabelsSyntax;
diff --git a/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useMonacoTheme.ts b/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useMonacoTheme.ts
new file mode 100644
index 000000000..e09101371
--- /dev/null
+++ b/app/vmui/packages/vmui/src/components/MonacoEditor/hooks/useMonacoTheme.ts
@@ -0,0 +1,25 @@
+import { useEffect } from "preact/compat";
+import { useAppState } from "../../../state/common/StateContext";
+import { Monaco } from "@monaco-editor/react";
+
+const useMonacoTheme = (monaco: Monaco | null) => {
+  const { isDarkTheme } = useAppState();
+
+  useEffect(() => {
+    if (!monaco) return;
+    monaco.editor.defineTheme("vm-theme", {
+      base: isDarkTheme ? "vs-dark" : "vs",
+      inherit: true,
+      rules: [],
+      colors: {
+        // #00000000 - for transparent
+        "editor.background": "#00000000",
+        "editor.lineHighlightBackground": "#00000000",
+        "editor.lineHighlightBorder": "#00000000"
+      }
+    });
+    monaco.editor.setTheme("vm-theme");
+  }, [monaco, isDarkTheme]);
+};
+
+export default useMonacoTheme;
diff --git a/app/vmui/packages/vmui/src/components/MonacoEditor/style.scss b/app/vmui/packages/vmui/src/components/MonacoEditor/style.scss
new file mode 100644
index 000000000..2297189be
--- /dev/null
+++ b/app/vmui/packages/vmui/src/components/MonacoEditor/style.scss
@@ -0,0 +1,27 @@
+@use "src/styles/variables" as *;
+
+.vm-monaco-editor {
+  display: grid;
+  height: 100%;
+  min-height: inherit;
+
+  &__input {
+    height: 100%;
+    padding: $padding-small 0;
+    resize: none;
+
+    &_resize {
+      &-vertical {
+        resize: vertical;
+      }
+
+      &-horizontal {
+        resize: horizontal;
+      }
+
+      &-both {
+        resize: both;
+      }
+    }
+  }
+}
diff --git a/app/vmui/packages/vmui/src/pages/ActiveQueries/index.tsx b/app/vmui/packages/vmui/src/pages/ActiveQueries/index.tsx
index fc6ebed81..6fd581ca7 100644
--- a/app/vmui/packages/vmui/src/pages/ActiveQueries/index.tsx
+++ b/app/vmui/packages/vmui/src/pages/ActiveQueries/index.tsx
@@ -11,6 +11,8 @@ import classNames from "classnames";
 import Button from "../../components/Main/Button/Button";
 import { RefreshIcon } from "../../components/Main/Icons";
 import "./style.scss";
+import { DATE_TIME_FORMAT } from "../../constants/date";
+import { roundStep } from "../../utils/time";
 
 const ActiveQueries: FC = () => {
   const { isMobile } = useDeviceDetect();
@@ -18,29 +20,31 @@ const ActiveQueries: FC = () => {
 
   const { data, lastUpdated, isLoading, error, fetchData } = useFetchActiveQueries();
 
-  const activeQueries = useMemo(() => data.map(({ duration, ...item }: ActiveQueriesType) => ({
-    duration,
-    data: JSON.stringify(item, null, 2),
-    from: dayjs(item.start).tz().format("MMM DD, YYYY \nHH:mm:ss.SSS"),
-    to: dayjs(item.end).tz().format("MMM DD, YYYY \nHH:mm:ss.SSS"),
-    ...item,
-  })), [data, timezone]);
+  const activeQueries = useMemo(() => data.map((item: ActiveQueriesType) => {
+    const from = dayjs(item.start).tz().format(DATE_TIME_FORMAT);
+    const to = dayjs(item.end).tz().format(DATE_TIME_FORMAT);
+    return {
+      duration: item.duration,
+      remote_addr: item.remote_addr,
+      query: item.query,
+      args: `${from} to ${to}, step=${roundStep(item.step)}`,
+      data: JSON.stringify(item, null, 2),
+    } as ActiveQueriesType;
+  }), [data, timezone]);
 
   const columns = useMemo(() => {
     if (!activeQueries?.length) return [];
-    const hideColumns = ["end", "start", "data"];
-    const keys = new Set<string>();
-    for (const item of activeQueries) {
-      for (const key in item) {
-        keys.add(key);
-      }
-    }
-    return Array.from(keys)
-      .filter((col) => !hideColumns.includes(col))
-      .map((key) => ({
-        key: key as keyof ActiveQueriesType,
-        title: key,
-      }));
+    const keys = Object.keys(activeQueries[0]) as (keyof ActiveQueriesType)[];
+
+    const titles: Partial<Record<keyof ActiveQueriesType, string>> = {
+      remote_addr: "client address",
+    };
+    const hideColumns = ["data"];
+
+    return keys.filter((col) => !hideColumns.includes(col)).map((key) => ({
+      key: key,
+      title: titles[key] || key,
+    }));
   }, [activeQueries]);
 
   const handleRefresh = async () => {
@@ -76,7 +80,7 @@ const ActiveQueries: FC = () => {
           <Table
             rows={activeQueries}
             columns={columns}
-            defaultOrderBy={"from"}
+            defaultOrderBy={"duration"}
             copyToClipboard={"data"}
             paginationOffset={{ startIndex: 0, endIndex: Infinity }}
           />
diff --git a/app/vmui/packages/vmui/src/pages/Relabel/index.tsx b/app/vmui/packages/vmui/src/pages/Relabel/index.tsx
index 620fd3714..595db152d 100644
--- a/app/vmui/packages/vmui/src/pages/Relabel/index.tsx
+++ b/app/vmui/packages/vmui/src/pages/Relabel/index.tsx
@@ -1,6 +1,5 @@
-import React, { FC, useEffect } from "preact/compat";
+import React, { FC, useCallback, useEffect } from "preact/compat";
 import "./style.scss";
-import TextField from "../../components/Main/TextField/TextField";
 import Button from "../../components/Main/Button/Button";
 import { InfoIcon, PlayIcon, WikiIcon } from "../../components/Main/Icons";
 import "./style.scss";
@@ -9,6 +8,7 @@ import Spinner from "../../components/Main/Spinner/Spinner";
 import Alert from "../../components/Main/Alert/Alert";
 import { useSearchParams } from "react-router-dom";
 import useStateSearchParams from "../../hooks/useStateSearchParams";
+import MonacoEditor from "../../components/MonacoEditor/MonacoEditor";
 
 const example = {
   config: `- if: '{bar_label=~"b.*"}'
@@ -30,20 +30,20 @@ const Relabel: FC = () => {
   const [config, setConfig] = useStateSearchParams("", "config");
   const [labels, setLabels] = useStateSearchParams("", "labels");
 
-  const handleChangeConfig = (val: string) => {
-    setConfig(val);
+  const handleChangeConfig = (val?: string) => {
+    setConfig(val || "");
   };
 
-  const handleChangeLabels = (val: string) => {
-    setLabels(val);
+  const handleChangeLabels = (val?: string) => {
+    setLabels(val || "");
   };
 
-  const handleRunQuery = () => {
+  const handleRunQuery = useCallback(() => {
     fetchData(config, labels);
     searchParams.set("config", config);
     searchParams.set("labels", labels);
     setSearchParams(searchParams);
-  };
+  }, [config, labels]);
 
   const handleRunExample = () => {
     const { config, labels } = example;
@@ -69,21 +69,24 @@ const Relabel: FC = () => {
     <section className="vm-relabeling">
       {loading && <Spinner/>}
       <div className="vm-relabeling-header vm-block">
-        <div className="vm-relabeling-header__configs">
-          <TextField
-            type="textarea"
+        <div className="vm-relabeling-header-configs">
+          <MonacoEditor
             label="Relabel configs"
             value={config}
-            autofocus
+            language={"yaml"}
+            resize={"vertical"}
             onChange={handleChangeConfig}
+            onEnter={handleRunQuery}
           />
         </div>
         <div className="vm-relabeling-header__labels">
-          <TextField
-            type="textarea"
+          <MonacoEditor
             label="Labels"
             value={labels}
+            language={"vm-labels"}
+            resize={"vertical"}
             onChange={handleChangeLabels}
+            onEnter={handleRunQuery}
           />
         </div>
         <div className="vm-relabeling-header-bottom">
diff --git a/app/vmui/packages/vmui/src/pages/Relabel/style.scss b/app/vmui/packages/vmui/src/pages/Relabel/style.scss
index 893454bbb..5a14aaf40 100644
--- a/app/vmui/packages/vmui/src/pages/Relabel/style.scss
+++ b/app/vmui/packages/vmui/src/pages/Relabel/style.scss
@@ -10,22 +10,12 @@
     align-items: flex-start;
     width: 100%;
 
-    &__configs {
-      textarea {
-        min-height: 200px;
-      }
+    &-configs {
+      min-height: 200px;
     }
 
     &__labels {
-      textarea {
-        min-height: 60px;
-      }
-    }
-
-    textarea {
-      overflow: auto;
-      width: 100%;
-      height: 100%;
+      min-height: 60px;
     }
 
     &-bottom {
diff --git a/app/vmui/packages/vmui/src/pages/TracePage/JsonForm/JsonForm.tsx b/app/vmui/packages/vmui/src/pages/TracePage/JsonForm/JsonForm.tsx
index b34764138..9c3933d71 100644
--- a/app/vmui/packages/vmui/src/pages/TracePage/JsonForm/JsonForm.tsx
+++ b/app/vmui/packages/vmui/src/pages/TracePage/JsonForm/JsonForm.tsx
@@ -99,6 +99,7 @@ const JsonForm: FC<JsonFormProps> = ({
         error={error}
         autofocus
         onChange={handleChangeJson}
+        onEnter={handleApply}
         disabled={!editable}
       />
       <div className="vm-json-form-footer">
diff --git a/app/vmui/packages/vmui/src/pages/WithTemplate/hooks/useExpandWithExprs.ts b/app/vmui/packages/vmui/src/pages/WithTemplate/hooks/useExpandWithExprs.ts
index 28e13db94..f13e52e36 100644
--- a/app/vmui/packages/vmui/src/pages/WithTemplate/hooks/useExpandWithExprs.ts
+++ b/app/vmui/packages/vmui/src/pages/WithTemplate/hooks/useExpandWithExprs.ts
@@ -2,15 +2,19 @@ import { useAppState } from "../../../state/common/StateContext";
 import { useState } from "react";
 import { ErrorTypes } from "../../../types";
 import { getExpandWithExprUrl } from "../../../api/expand-with-exprs";
+import { useSearchParams } from "react-router-dom";
 
 export const useExpandWithExprs = () => {
   const { serverUrl } = useAppState();
+  const [searchParams, setSearchParams] = useSearchParams();
 
   const [data, setData] = useState("");
   const [loading, setLoading] = useState(false);
   const [error, setError] = useState<ErrorTypes | string>();
 
   const fetchData = async (query: string) => {
+    searchParams.set("expr", query);
+    setSearchParams(searchParams);
     const fetchUrl = getExpandWithExprUrl(serverUrl, query);
     setLoading(true);
     try {
diff --git a/app/vmui/packages/vmui/src/pages/WithTemplate/index.tsx b/app/vmui/packages/vmui/src/pages/WithTemplate/index.tsx
index 27097a174..602df8362 100644
--- a/app/vmui/packages/vmui/src/pages/WithTemplate/index.tsx
+++ b/app/vmui/packages/vmui/src/pages/WithTemplate/index.tsx
@@ -1,16 +1,19 @@
 import React, { FC } from "preact/compat";
 import "./style.scss";
 import TextField from "../../components/Main/TextField/TextField";
-import { useState } from "react";
+import { useEffect, useState } from "react";
 import Button from "../../components/Main/Button/Button";
 import { PlayIcon } from "../../components/Main/Icons";
 import WithTemplateTutorial from "./WithTemplateTutorial/WithTemplateTutorial";
 import { useExpandWithExprs } from "./hooks/useExpandWithExprs";
 import Spinner from "../../components/Main/Spinner/Spinner";
+import { useSearchParams } from "react-router-dom";
 
 const WithTemplate: FC = () => {
+  const [searchParams] = useSearchParams();
+
   const { data, loading, error, expand } = useExpandWithExprs();
-  const [expr, setExpr] = useState("");
+  const [expr, setExpr] = useState(searchParams.get("expr") || "");
 
   const handleChangeInput = (val: string) => {
     setExpr(val);
@@ -20,6 +23,10 @@ const WithTemplate: FC = () => {
     expand(expr);
   };
 
+  useEffect(() => {
+    if (expr) expand(expr);
+  }, []);
+
   return (
     <section className="vm-with-template">
       {loading && <Spinner />}
@@ -32,6 +39,7 @@ const WithTemplate: FC = () => {
             value={expr}
             error={error}
             autofocus
+            onEnter={handleRunQuery}
             onChange={handleChangeInput}
           />
         </div>
diff --git a/app/vmui/packages/vmui/src/pages/WithTemplate/style.scss b/app/vmui/packages/vmui/src/pages/WithTemplate/style.scss
index 71a79e785..323d02ffd 100644
--- a/app/vmui/packages/vmui/src/pages/WithTemplate/style.scss
+++ b/app/vmui/packages/vmui/src/pages/WithTemplate/style.scss
@@ -26,6 +26,7 @@
     }
 
     textarea {
+      font-family: $font-family-monospace;
       overflow: auto;
       width: 100%;
       height: 100%;
diff --git a/app/vmui/packages/vmui/src/types/index.ts b/app/vmui/packages/vmui/src/types/index.ts
index c199808a9..33235ea1c 100644
--- a/app/vmui/packages/vmui/src/types/index.ts
+++ b/app/vmui/packages/vmui/src/types/index.ts
@@ -147,7 +147,6 @@ export interface ActiveQueriesType {
   query: string;
   remote_addr: string;
   step: number;
-  from?: string;
-  to?: string;
+  args?: string;
   data?: string;
 }
diff --git a/app/vmui/packages/vmui/src/utils/time.ts b/app/vmui/packages/vmui/src/utils/time.ts
index 3f4c1c59d..fb964e574 100644
--- a/app/vmui/packages/vmui/src/utils/time.ts
+++ b/app/vmui/packages/vmui/src/utils/time.ts
@@ -30,7 +30,7 @@ const shortDurations = supportedDurations.map(d => d.short);
 
 export const roundToMilliseconds = (num: number): number => Math.round(num*1000)/1000;
 
-const roundStep = (step: number) => {
+export const roundStep = (step: number) => {
   let result = roundToMilliseconds(step);
   const integerStep = Math.round(step);
 
diff --git a/deployment/docker/victorialogs/promtail/config.yml b/deployment/docker/victorialogs/promtail/config.yml
index 3587c8dae..4a5974a7d 100644
--- a/deployment/docker/victorialogs/promtail/config.yml
+++ b/deployment/docker/victorialogs/promtail/config.yml
@@ -6,7 +6,7 @@ positions:
   filename: /tmp/positions.yaml
 
 clients:
-  - url: http://vlogs:9428/insert/loki/api/v1/push?_stream_fields=filename,job,stream,host,app,pid
+  - url: http://vlogs:9428/insert/loki/api/v1/push?_stream_fields=instance,job,host,app,pid
     tenant_id: "0:0"
 
 scrape_configs:
diff --git a/deployment/docker/victorialogs/promtail/docker-compose.yml b/deployment/docker/victorialogs/promtail/docker-compose.yml
index cd9e6c0ad..1e05a2f8f 100644
--- a/deployment/docker/victorialogs/promtail/docker-compose.yml
+++ b/deployment/docker/victorialogs/promtail/docker-compose.yml
@@ -13,7 +13,7 @@ services:
 
   # Run `make package-victoria-logs` to build victoria-logs image
   vlogs:
-    image: docker.io/victoriametrics/victoria-logs:latest
+    image: docker.io/victoriametrics/victoria-logs:v0.2.0-victorialogs
     volumes:
       - victorialogs-promtail-docker:/vlogs
     ports:
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index af92a1365..306db79bf 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -46,9 +46,11 @@ The following `tip` changes can be tested by building VictoriaMetrics components
 * FEATUTE: [vmalert](https://docs.victoriametrics.com/vmalert.html): display the error message received during unsuccessful config reload in vmalert's UI. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4076) for details.
 * FEATUTE: [vmalert](https://docs.victoriametrics.com/vmalert.html): allow disabling of `step` param attached to [instant queries](https://docs.victoriametrics.com/keyConcepts.html#instant-query). This might be useful for using vmalert with datasources that to not support this param, unlike VictoriaMetrics. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4573) for details.
 * FEATUTE: [vmalert](https://docs.victoriametrics.com/vmalert.html): support option for "blackholing" alerting notifications if `-notifier.blackhole` cmd-line flag is set. Enable this flag if you want vmalert to evaluate alerting rules without sending any notifications to external receivers (eg. alertmanager). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4122) for details. Thanks to @venkatbvc for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/4639).
+* FEATURE: [vmalert](https://docs.victoriametrics.com/vmalert.html): add unit test for alerting and recording rules, see more [details](https://docs.victoriametrics.com/vmalert.html#unit-testing-for-rules) here. Thanks to @Haleygo for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/4596).
 * FEATURE: [vmauth](https://docs.victoriametrics.com/vmauth.html): expose `vmauth_user_request_duration_seconds` and `vmauth_unauthorized_user_request_duration_seconds` summary metrics for measuring requests latency per user.
 * FEATURE: [vmbackup](https://docs.victoriametrics.com/vmbackup.html): show backup progress percentage in log during backup uploading. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4460).
 * FEATURE: [vmrestore](https://docs.victoriametrics.com/vmrestore.html): show restoring progress percentage in log during backup downloading. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4460).
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): allow configuring staleness interval in [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html) config. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4667) for details.
 * FEATURE: add ability to fine-tune Graphite API limits via the following command-line flags:
   `-search.maxGraphiteTagKeys` for limiting the number of tag keys returned from [Graphite API for tags](https://docs.victoriametrics.com/#graphite-tags-api-usage)
   `-search.maxGraphiteTagValues` for limiting the number of tag values returned from [Graphite API for tag values](https://docs.victoriametrics.com/#graphite-tags-api-usage)
diff --git a/docs/README.md b/docs/README.md
index e03318e8b..e9479f84e 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -404,6 +404,11 @@ matching the specified [series selector](https://prometheus.io/docs/prometheus/l
 
 Cardinality explorer is built on top of [/api/v1/status/tsdb](#tsdb-stats).
 
+In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html) each vmstorage tracks the stored time series individually.
+vmselect requests stats via [/api/v1/status/tsdb](#tsdb-stats) API from each vmstorage node and merges the results by summing per-series stats.
+This may lead to inflated values when samples for the same time series are spread across multiple vmstorage nodes
+due to [replication](#replication) or [rerouting](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html?highlight=re-routes#cluster-availability).
+
 See [cardinality explorer playground](https://play.victoriametrics.com/select/accounting/1/6a716b0f-38bc-4856-90ce-448fd713e3fe/prometheus/graph/#/cardinality).
 See the example of using the cardinality explorer [here](https://victoriametrics.com/blog/cardinality-explorer/).
 
@@ -1671,7 +1676,8 @@ Retention filters can be evaluated for free by downloading and using enterprise
 
 * `-downsampling.period=30d:5m,180d:1h` instructs VictoriaMetrics to deduplicate samples older than 30 days with 5 minutes interval and to deduplicate samples older than 180 days with 1 hour interval.
 
-Downsampling is applied independently per each time series. It can reduce disk space usage and improve query performance if it is applied to time series with big number of samples per each series. The downsampling doesn't improve query performance if the database contains big number of time series with small number of samples per each series (aka [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate)), since downsampling doesn't reduce the number of time series. So the majority of time is spent on searching for the matching time series. It is possible to use recording rules in [vmalert](https://docs.victoriametrics.com/vmalert.html) in order to reduce the number of time series. See [these docs](https://docs.victoriametrics.com/vmalert.html#downsampling-and-aggregation-via-vmalert).
+Downsampling is applied independently per each time series. It can reduce disk space usage and improve query performance if it is applied to time series with big number of samples per each series. The downsampling doesn't improve query performance if the database contains big number of time series with small number of samples per each series (aka [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate)), since downsampling doesn't reduce the number of time series. So the majority of time is spent on searching for the matching time series. 
+It is possible to use [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html) in vmagent or recording rules in [vmalert](https://docs.victoriametrics.com/vmalert.html) in order to [reduce the number of time series](https://docs.victoriametrics.com/vmalert.html#downsampling-and-aggregation-via-vmalert).
 
 Downsampling happens during [background merges](https://docs.victoriametrics.com/#storage) 
 and can't be performed if there is not enough of free disk space or if vmstorage 
@@ -1786,6 +1792,11 @@ VictoriaMetrics returns TSDB stats at `/api/v1/status/tsdb` page in the way simi
 * `match[]=SELECTOR` where `SELECTOR` is an arbitrary [time series selector](https://prometheus.io/docs/prometheus/latest/querying/basics/#time-series-selectors) for series to take into account during stats calculation. By default all the series are taken into account.
 * `extra_label=LABEL=VALUE`. See [these docs](#prometheus-querying-api-enhancements) for more details.
 
+In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html) each vmstorage tracks the stored time series individually.
+vmselect requests stats via [/api/v1/status/tsdb](#tsdb-stats) API from each vmstorage node and merges the results by summing per-series stats.
+This may lead to inflated values when samples for the same time series are spread across multiple vmstorage nodes
+due to [replication](#replication) or [rerouting](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html?highlight=re-routes#cluster-availability).
+
 VictoriaMetrics provides an UI on top of `/api/v1/status/tsdb` - see [cardinality explorer docs](#cardinality-explorer).
 
 ## Query tracing
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index faac96326..096a1fbe1 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -412,6 +412,11 @@ matching the specified [series selector](https://prometheus.io/docs/prometheus/l
 
 Cardinality explorer is built on top of [/api/v1/status/tsdb](#tsdb-stats).
 
+In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html) each vmstorage tracks the stored time series individually.
+vmselect requests stats via [/api/v1/status/tsdb](#tsdb-stats) API from each vmstorage node and merges the results by summing per-series stats.
+This may lead to inflated values when samples for the same time series are spread across multiple vmstorage nodes
+due to [replication](#replication) or [rerouting](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html?highlight=re-routes#cluster-availability).
+
 See [cardinality explorer playground](https://play.victoriametrics.com/select/accounting/1/6a716b0f-38bc-4856-90ce-448fd713e3fe/prometheus/graph/#/cardinality).
 See the example of using the cardinality explorer [here](https://victoriametrics.com/blog/cardinality-explorer/).
 
@@ -1679,7 +1684,8 @@ Retention filters can be evaluated for free by downloading and using enterprise
 
 * `-downsampling.period=30d:5m,180d:1h` instructs VictoriaMetrics to deduplicate samples older than 30 days with 5 minutes interval and to deduplicate samples older than 180 days with 1 hour interval.
 
-Downsampling is applied independently per each time series. It can reduce disk space usage and improve query performance if it is applied to time series with big number of samples per each series. The downsampling doesn't improve query performance if the database contains big number of time series with small number of samples per each series (aka [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate)), since downsampling doesn't reduce the number of time series. So the majority of time is spent on searching for the matching time series. It is possible to use recording rules in [vmalert](https://docs.victoriametrics.com/vmalert.html) in order to reduce the number of time series. See [these docs](https://docs.victoriametrics.com/vmalert.html#downsampling-and-aggregation-via-vmalert).
+Downsampling is applied independently per each time series. It can reduce disk space usage and improve query performance if it is applied to time series with big number of samples per each series. The downsampling doesn't improve query performance if the database contains big number of time series with small number of samples per each series (aka [high churn rate](https://docs.victoriametrics.com/FAQ.html#what-is-high-churn-rate)), since downsampling doesn't reduce the number of time series. So the majority of time is spent on searching for the matching time series. 
+It is possible to use [stream aggregation](https://docs.victoriametrics.com/stream-aggregation.html) in vmagent or recording rules in [vmalert](https://docs.victoriametrics.com/vmalert.html) in order to [reduce the number of time series](https://docs.victoriametrics.com/vmalert.html#downsampling-and-aggregation-via-vmalert).
 
 Downsampling happens during [background merges](https://docs.victoriametrics.com/#storage) 
 and can't be performed if there is not enough of free disk space or if vmstorage 
@@ -1794,6 +1800,11 @@ VictoriaMetrics returns TSDB stats at `/api/v1/status/tsdb` page in the way simi
 * `match[]=SELECTOR` where `SELECTOR` is an arbitrary [time series selector](https://prometheus.io/docs/prometheus/latest/querying/basics/#time-series-selectors) for series to take into account during stats calculation. By default all the series are taken into account.
 * `extra_label=LABEL=VALUE`. See [these docs](#prometheus-querying-api-enhancements) for more details.
 
+In [cluster version of VictoriaMetrics](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html) each vmstorage tracks the stored time series individually.
+vmselect requests stats via [/api/v1/status/tsdb](#tsdb-stats) API from each vmstorage node and merges the results by summing per-series stats.
+This may lead to inflated values when samples for the same time series are spread across multiple vmstorage nodes
+due to [replication](#replication) or [rerouting](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html?highlight=re-routes#cluster-availability).
+
 VictoriaMetrics provides an UI on top of `/api/v1/status/tsdb` - see [cardinality explorer docs](#cardinality-explorer).
 
 ## Query tracing
diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index 8d6223127..54e84ff97 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -5,6 +5,8 @@ according to [these docs](https://docs.victoriametrics.com/VictoriaLogs/QuickSta
 
 ## tip
 
+* FEATURE: add support for data ingestion via Promtail (aka default log shipper for Grafana Loki). See [these](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/Promtail.html) and [these](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/#loki-json-api) docs.
+
 ## [v0.2.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.2.0-victorialogs)
 
 Released at 2023-07-17
diff --git a/docs/VictoriaLogs/data-ingestion/Promtail.md b/docs/VictoriaLogs/data-ingestion/Promtail.md
index 7aa3850b4..7d72e9a99 100644
--- a/docs/VictoriaLogs/data-ingestion/Promtail.md
+++ b/docs/VictoriaLogs/data-ingestion/Promtail.md
@@ -1,16 +1,20 @@
 # Promtail setup
 
+[Promtail](https://grafana.com/docs/loki/latest/clients/promtail/) is a default log shipper for Grafana Loki.
+Promtail can be configured to send the collected logs to VictoriaLogs according to the following docs.
+
 Specify [`clients`](https://grafana.com/docs/loki/latest/clients/promtail/configuration/#clients) section in the configuration file
 for sending the collected logs to [VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/):
 
 ```yaml
 clients:
-  - url: http://vlogs:9428/insert/loki/api/v1/push?_stream_fields=filename,job,stream,host,app,pid
+  - url: http://localhost:9428/insert/loki/api/v1/push?_stream_fields=instance,job,host,app,pid
 ```
 
-Substitute `vlogs:9428` address inside `clients` with the real TCP address of VictoriaLogs.
+Substitute `localhost:9428` address inside `clients` with the real TCP address of VictoriaLogs.
 
 See [these docs](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/#http-parameters) for details on the used URL query parameter section.
+There is no need in specifying `_msg_field` and `_time_field` query args, since VictoriaLogs automatically extracts log message and timestamp from the ingested Loki data.
 
 It is recommended verifying whether the initial setup generates the needed [log fields](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#data-model)
 and uses the correct [stream fields](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#stream-fields).
@@ -19,27 +23,28 @@ and inspecting VictoriaLogs logs then:
 
 ```yaml
 clients:
-  - url: http://vlogs:9428/insert/loki/api/v1/push?_stream_fields=filename,job,stream,host,app,pid&debug=1
+  - url: http://localhost:9428/insert/loki/api/v1/push?_stream_fields=instance,job,host,app,pid&debug=1
 ```
 
 If some [log fields](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#data-model) must be skipped
 during data ingestion, then they can be put into `ignore_fields` [parameter](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/#http-parameters).
-For example, the following config instructs VictoriaLogs to ignore `log.offset` and `event.original` fields in the ingested logs:
+For example, the following config instructs VictoriaLogs to ignore `filename` and `stream` fields in the ingested logs:
 
 ```yaml
 clients:
-  - url: http://vlogs:9428/insert/loki/api/v1/push?_stream_fields=filename,job,stream,host,app,pid&debug=1
+  - url: http://localhost:9428/insert/loki/api/v1/push?_stream_fields=instance,job,host,app,pid&ignore_fields=filename,stream
 ```
 
 By default the ingested logs are stored in the `(AccountID=0, ProjectID=0)` [tenant](https://docs.victoriametrics.com/VictoriaLogs/#multitenancy).
-If you need storing logs in other tenant, then It is possible to either use `tenant_id` provided by Loki configuration, or to use `headers` and provide
-`AccountID` and `ProjectID` headers. Format for `tenant_id` is `AccountID:ProjectID`. 
-For example, the following config instructs VictoriaLogs to store logs in the `(AccountID=12, ProjectID=12)` tenant:
+If you need storing logs in other tenant, then specify the needed tenant via `tenant_id` field
+in the [Loki client configuration](https://grafana.com/docs/loki/latest/clients/promtail/configuration/#clients)
+The `tenant_id` must have `AccountID:ProjectID` format, where `AccountID` and `ProjectID` are arbitrary uint32 numbers.
+For example, the following config instructs VictoriaLogs to store logs in the `(AccountID=12, ProjectID=34)` [tenant](https://docs.victoriametrics.com/VictoriaLogs/#multitenancy):
 
 ```yaml
 clients:
-  - url: http://vlogs:9428/insert/loki/api/v1/push?_stream_fields=filename,job,stream,host,app,pid&debug=1
-    tenant_id: "12:12"
+  - url: http://localhost:9428/insert/loki/api/v1/push?_stream_fields=instance,job,host,app,pid&debug=1
+    tenant_id: "12:34"
 ```
 
 The ingested log entries can be queried according to [these docs](https://docs.victoriametrics.com/VictoriaLogs/querying/).
diff --git a/docs/VictoriaLogs/data-ingestion/README.md b/docs/VictoriaLogs/data-ingestion/README.md
index 6b02566cd..925106100 100644
--- a/docs/VictoriaLogs/data-ingestion/README.md
+++ b/docs/VictoriaLogs/data-ingestion/README.md
@@ -17,7 +17,7 @@ menu:
 - Fluentbit. See [how to setup Fluentbit for sending logs to VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/Fluentbit.html).
 - Logstash. See [how to setup Logstash for sending logs to VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/Logstash.html).
 - Vector. See [how to setup Vector for sending logs to VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/Vector.html).
-- Promtail. See [how to setup Promtail for sending logs to VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/Promtail.html).
+- Promtail (aka Grafana Loki). See [how to setup Promtail for sending logs to VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/Promtail.html).
 
 The ingested logs can be queried according to [these docs](https://docs.victoriametrics.com/VictoriaLogs/querying/).
 
@@ -33,7 +33,7 @@ VictoriaLogs supports the following data ingestion HTTP APIs:
 
 - Elasticsearch bulk API. See [these docs](#elasticsearch-bulk-api).
 - JSON stream API aka [ndjson](http://ndjson.org/). See [these docs](#json-stream-api).
-- [Loki JSON API](https://grafana.com/docs/loki/latest/api/#push-log-entries-to-lokiq). See [these docs](#loki-json-api).
+- Loki JSON API. See [these docs](#loki-json-api).
 
 VictoriaLogs accepts optional [HTTP parameters](#http-parameters) at data ingestion HTTP APIs.
 
@@ -47,12 +47,18 @@ The following command pushes a single log line to VictoriaLogs:
 
 ```bash
 echo '{"create":{}}
-{"_msg":"cannot open file","_time":"2023-06-21T04:24:24Z","host.name":"host123"}
+{"_msg":"cannot open file","_time":"0","host.name":"host123"}
 ' | curl -X POST -H 'Content-Type: application/json' --data-binary @- http://localhost:9428/insert/elasticsearch/_bulk
 ```
 
 It is possible to push thousands of log lines in a single request to this API.
 
+If the [timestamp field](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#time-field) is set to `"0"`,
+then the current timestamp at VictoriaLogs side is used per each ingested log line.
+Otherwise the timestamp field must be in the [ISO8601](https://en.wikipedia.org/wiki/ISO_8601) format. For example, `2023-06-20T15:32:10Z`.
+Optional fractional part of seconds can be specified after the dot - `2023-06-20T15:32:10.123Z`.
+Timezone can be specified instead of `Z` suffix - `2023-06-20T15:32:10+02:00`.
+
 See [these docs](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#data-model) for details on fields,
 which must be present in the ingested log messages.
 
@@ -88,17 +94,18 @@ VictoriaLogs accepts JSON line stream aka [ndjson](http://ndjson.org/) at `http:
 The following command pushes multiple log lines to VictoriaLogs:
 
  ```bash
-echo '{ "log": { "level": "info", "message": "hello world" }, "date": "2023-06-20T15:31:23Z", "stream": "stream1" }
-{ "log": { "level": "error", "message": "oh no!" }, "date": "2023-06-20T15:32:10.567Z", "stream": "stream1" }
-{ "log": { "level": "info", "message": "hello world" }, "date": "2023-06-20T15:35:11.567890+02:00", "stream": "stream2" }
+echo '{ "log": { "level": "info", "message": "hello world" }, "date": "0", "stream": "stream1" }
+{ "log": { "level": "error", "message": "oh no!" }, "date": "0", "stream": "stream1" }
+{ "log": { "level": "info", "message": "hello world" }, "date": "0", "stream": "stream2" }
 ' | curl -X POST -H 'Content-Type: application/stream+json' --data-binary @- \
   'http://localhost:9428/insert/jsonline?_stream_fields=stream&_time_field=date&_msg_field=log.message'
 ```
 
 It is possible to push unlimited number of log lines in a single request to this API.
 
-The [timestamp field](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#time-field) must be
-in the [ISO8601](https://en.wikipedia.org/wiki/ISO_8601) format. For example, `2023-06-20T15:32:10Z`.
+If the [timestamp field](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#time-field) is set to `"0"`,
+then the current timestamp at VictoriaLogs side is used per each ingested log line.
+Otherwise the timestamp field must be in the [ISO8601](https://en.wikipedia.org/wiki/ISO_8601) format. For example, `2023-06-20T15:32:10Z`.
 Optional fractional part of seconds can be specified after the dot - `2023-06-20T15:32:10.123Z`.
 Timezone can be specified instead of `Z` suffix - `2023-06-20T15:32:10+02:00`.
 
@@ -134,15 +141,42 @@ See also:
 
 ### Loki JSON API
 
-VictoriaLogs accepts logs in [Loki JSON API](https://grafana.com/docs/loki/latest/api/#push-log-entries-to-lokiq) format at `http://localhost:9428/insert/loki/api/v1/push` endpoint.
+VictoriaLogs accepts logs in [Loki JSON API](https://grafana.com/docs/loki/latest/api/#push-log-entries-to-loki) format at `http://localhost:9428/insert/loki/api/v1/push` endpoint.
 
 The following command pushes a single log line to Loki JSON API at VictoriaLogs:
 
 ```bash
-curl -v -H "Content-Type: application/json" -XPOST -s "http://localhost:9428/insert/loki/api/v1/push?_stream_fields=foo" --data-raw \
-  '{"streams": [{ "stream": { "foo": "bar2" }, "values": [ [ "1570818238000000000", "fizzbuzz" ] ] }]}'
+curl -H "Content-Type: application/json" -XPOST "http://localhost:9428/insert/loki/api/v1/push?_stream_fields=instance,job" --data-raw \
+  '{"streams": [{ "stream": { "instance": "host123", "job": "app42" }, "values": [ [ "0", "foo fizzbuzz bar" ] ] }]}'
+```
+
+It is possible to push thousands of log streams and log lines in a single request to this API.
+
+The API accepts various http parameters, which can change the data ingestion behavior - [these docs](#http-parameters) for details.
+
+The following command verifies that the data has been successfully ingested into VictoriaLogs by [querying](https://docs.victoriametrics.com/VictoriaLogs/querying/) it:
+
+```bash
+curl http://localhost:9428/select/logsql/query -d 'query=fizzbuzz'
 ```
 
+The command should return the following response:
+
+```bash
+{"_msg":"foo fizzbuzz bar","_stream":"{instance=\"host123\",job=\"app42\"}","_time":"2023-07-20T23:01:19.288676497Z"}
+```
+
+The response by default contains [`_msg`](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#message-field),
+[`_stream`](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#stream-fields) and
+[`_time`](https://docs.victoriametrics.com/VictoriaLogs/keyConcepts.html#time-field) fields plus the explicitly mentioned fields.
+See [these docs](https://docs.victoriametrics.com/VictoriaLogs/LogsQL.html#querying-specific-fields) for details.
+
+See also:
+
+- [How to debug data ingestion](#troubleshooting).
+- [HTTP parameters, which can be passed to the API](#http-parameters).
+- [How to query VictoriaLogs](https://docs.victoriametrics.com/VictoriaLogs/querying.html).
+
 ### HTTP parameters
 
 VictoriaLogs accepts the following parameters at [data ingestion HTTP APIs](#http-apis):
diff --git a/docs/enterprise.md b/docs/enterprise.md
index be2458904..6920f9074 100644
--- a/docs/enterprise.md
+++ b/docs/enterprise.md
@@ -27,7 +27,7 @@ The use of VictoriaMetrics enterprise components is permitted in the following c
 - Production use if you have a valid enterprise contract or valid permit from VictoriaMetrics company.
   [Contact us](mailto:info@victoriametrics.com) if you need such contract.
 
-- [Managed VictoriaMetrics](https://docs.victoriametrics.com/managed_victoriametrics/) is built on top of enterprise binaries of VictoriaMetrics.
+- [Managed VictoriaMetrics](https://docs.victoriametrics.com/managed-victoriametrics/) is built on top of enterprise binaries of VictoriaMetrics.
 
 All the enterprise apps require `-eula` command-line flag to be passed to them. This flag acknowledges that your usage fits one of the cases listed above.
 
diff --git a/docs/guides/understand-your-setup-size.md b/docs/guides/understand-your-setup-size.md
index ead8a08a4..9a41540b7 100644
--- a/docs/guides/understand-your-setup-size.md
+++ b/docs/guides/understand-your-setup-size.md
@@ -10,7 +10,7 @@ aliases:
 ---
 # Understand Your Setup Size
 
-The docs provide a simple and high-level overview of Ingestion Rate, Active Time Series, and Query per Second. These terms are a part of capacity planning ([Single-Node](https://docs.victoriametrics.com/Single-server-VictoriaMetrics.html#capacity-planning), [Cluster](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#capacity-planning)) and [Managed VictoriaMetrics](https://docs.victoriametrics.com/managed_victoriametrics/) pricing.
+The docs provide a simple and high-level overview of Ingestion Rate, Active Time Series, and Query per Second. These terms are a part of capacity planning ([Single-Node](https://docs.victoriametrics.com/Single-server-VictoriaMetrics.html#capacity-planning), [Cluster](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#capacity-planning)) and [Managed VictoriaMetrics](https://docs.victoriametrics.com/managed-victoriametrics/) pricing.
 
 ## Terminology
 
diff --git a/docs/stream-aggregation.md b/docs/stream-aggregation.md
index 5f320d1a9..dc81fcc3d 100644
--- a/docs/stream-aggregation.md
+++ b/docs/stream-aggregation.md
@@ -358,7 +358,7 @@ Below are aggregation functions that can be put in the `outputs` list at [stream
 ### total
 
 `total` generates output [counter](https://docs.victoriametrics.com/keyConcepts.html#counter) by summing the input counters.
-`total` only makes sense for aggregating [counter](https://docs.victoriametrics.com/keyConcepts.html#counter) type metrics.
+`total` only makes sense for aggregating [counter](https://docs.victoriametrics.com/keyConcepts.html#counter) metrics.
 
 The results of `total` is equal to the `sum(some_counter)` query.
 
@@ -383,7 +383,7 @@ This changes a label with pod's name in the series, but `total` account for such
 ### increase
 
 `increase` returns the increase of input [counters](https://docs.victoriametrics.com/keyConcepts.html#counter).
-`increase` only makes sense for aggregating [counter](https://docs.victoriametrics.com/keyConcepts.html#counter) type metrics.
+`increase` only makes sense for aggregating [counter](https://docs.victoriametrics.com/keyConcepts.html#counter) metrics.
 
 The results of `increase` with aggregation interval of `1m` is equal to the `increase(some_counter[1m])` query.
 
@@ -406,6 +406,7 @@ The results of `count_samples` with aggregation interval of `1m` is equal to the
 ### sum_samples
 
 `sum_samples` sums input [sample values](https://docs.victoriametrics.com/keyConcepts.html#raw-samples).
+`sum_samples` makes sense only for aggregating [gauge](https://docs.victoriametrics.com/keyConcepts.html#gauge) metrics.
 
 The results of `sum_samples` with aggregation interval of `1m` is equal to the `sum_over_time(some_metric[1m])` query.
 
@@ -455,12 +456,14 @@ For example, see below time series produced by config with aggregation interval
 ### stddev
 
 `stddev` returns [standard deviation](https://en.wikipedia.org/wiki/Standard_deviation) for the input [sample values](https://docs.victoriametrics.com/keyConcepts.html#raw-samples).
+`stddev` makes sense only for aggregating [gauge](https://docs.victoriametrics.com/keyConcepts.html#gauge) metrics.
 
 The results of `stddev` with aggregation interval of `1m` is equal to the `stddev_over_time(some_metric[1m])` query.
 
 ### stdvar
 
 `stdvar` returns [standard variance](https://en.wikipedia.org/wiki/Variance) for the input [sample values](https://docs.victoriametrics.com/keyConcepts.html#raw-samples).
+`stdvar` makes sense only for aggregating [gauge](https://docs.victoriametrics.com/keyConcepts.html#gauge) metrics.
 
 The results of `stdvar` with aggregation interval of `1m` is equal to the `stdvar_over_time(some_metric[1m])` query.
 
@@ -472,6 +475,7 @@ For example, see below time series produced by config with aggregation interval
 
 `histogram_bucket` returns [VictoriaMetrics histogram buckets](https://valyala.medium.com/improving-histogram-usability-for-prometheus-and-grafana-bc7e5df0e350)
   for the input [sample values](https://docs.victoriametrics.com/keyConcepts.html#raw-samples).
+`histogram_bucket` makes sense only for aggregating [gauge](https://docs.victoriametrics.com/keyConcepts.html#gauge) metrics.
 
 The results of `histogram_bucket` with aggregation interval of `1m` is equal to the `histogram_over_time(some_histogram_bucket[1m])` query.
 
@@ -480,6 +484,7 @@ The results of `histogram_bucket` with aggregation interval of `1m` is equal to
 `quantiles(phi1, ..., phiN)` returns [percentiles](https://en.wikipedia.org/wiki/Percentile) for the given `phi*`
 over the input [sample values](https://docs.victoriametrics.com/keyConcepts.html#raw-samples). 
 The `phi` must be in the range `[0..1]`, where `0` means `0th` percentile, while `1` means `100th` percentile.
+`quantiles(...)` makes sense only for aggregating [gauge](https://docs.victoriametrics.com/keyConcepts.html#gauge) metrics.
 
 The results of `quantiles(phi1, ..., phiN)` with aggregation interval of `1m` 
 is equal to the `quantiles_over_time("quantile", phi1, ..., phiN, some_histogram_bucket[1m])` query.
@@ -541,6 +546,16 @@ at [single-node VictoriaMetrics](https://docs.victoriametrics.com/Single-server-
   # The aggregated stats is sent to remote storage once per interval.
   interval: 1m
 
+  # staleness_interval defines an interval after which the series state will be reset if no samples have been sent during it.
+  # It means that:
+  # - no data point will be written for a resulting time series if it didn't receive any updates during configured interval,
+  # - if the series receives updates after the configured interval again, then the time series will be calculated from the initial state
+  #   (it's like this series didn't exist until now).
+  # Increase this parameter if it is expected for matched metrics to be delayed or collected with irregular intervals exceeding the `interval` value.
+  # By default, is equal to x2 of the `interval` field.
+  # The parameter is only relevant for outputs: total, increase and histogram_bucket.
+  # staleness_interval: 2m
+
   # without is an optional list of labels, which must be removed from the output aggregation.
   # See https://docs.victoriametrics.com/stream-aggregation.html#aggregating-by-labels
   without: [instance]
diff --git a/docs/vmalert.md b/docs/vmalert.md
index 27e4d05b8..085f3c3cb 100644
--- a/docs/vmalert.md
+++ b/docs/vmalert.md
@@ -743,6 +743,248 @@ See full description for these flags in `./vmalert -help`.
 * `query` template function is disabled for performance reasons (might be changed in future);
 * `limit` group's param has no effect during replay (might be changed in future);
 
+## Unit Testing for Rules
+
+> Unit testing is available from v1.92.0. 
+> Unit tests do not respect `-clusterMode` for now.
+
+You can use `vmalert` to run unit tests for alerting and recording rules.
+In unit test mode vmalert performs the following actions:
+* sets up an isolated VictoriaMetrics instance;
+* simulates the periodic ingestion of time series;
+* queries the ingested data for recording and alerting rules evaluation;
+* tests whether the firing alerts or resulting recording rules match the expected results. 
+
+See how to run vmalert in unit test mode below:
+```
+# Run vmalert with one or multiple test files via -unittestFile cmd-line flag
+./vmalert -unittestFile=test1.yaml -unittestFile=test2.yaml
+```
+
+vmalert is compatible with [Prometheus config format for tests](https://prometheus.io/docs/prometheus/latest/configuration/unit_testing_rules/#test-file-format)
+except `promql_expr_test` field. Use `metricsql_expr_test` field name instead. The name is different because vmalert 
+validates and executes [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html) expressions, 
+which aren't always backward compatible with [PromQL](https://prometheus.io/docs/prometheus/latest/querying/basics/).
+
+### Test file format
+
+The configuration format for files specified in `-unittestFile` cmd-line flag is the following:
+```
+# Path to the files or http url containing [rule groups](https://docs.victoriametrics.com/vmalert.html#groups) configuration.
+# Enterprise version of vmalert supports S3 and GCS paths to rules.
+rule_files:
+  [ - <string> ]
+
+# The evaluation interval for rules specified in `rule_files`
+[ evaluation_interval: <duration> | default = 1m ]
+
+# Groups listed below will be evaluated by order.
+# Not All the groups need not be mentioned, if not, they will be evaluated by define order in rule_files.
+group_eval_order:
+  [ - <string> ]
+
+# The list of unit test files to be checked during evaluation.
+tests:
+  [ - <test_group> ]
+```
+
+#### `<test_group>`
+
+```
+# Interval between samples for input series
+interval: <duration>
+# Time series to persist into the database according to configured <interval> before running tests.
+input_series:
+  [ - <series> ]
+
+# Name of the test group, optional
+[ name: <string> ]
+
+# Unit tests for alerting rules
+alert_rule_test:
+  [ - <alert_test_case> ]
+
+# Unit tests for Metricsql expressions.
+metricsql_expr_test:
+  [ - <metricsql_expr_test> ]
+
+# External labels accessible for templating.
+external_labels:
+  [ <labelname>: <string> ... ]
+
+```
+
+#### `<series>`
+
+```
+# series in the following format '<metric name>{<label name>=<label value>, ...}'
+# Examples:
+#      series_name{label1="value1", label2="value2"}
+#      go_goroutines{job="prometheus", instance="localhost:9090"}
+series: <string>
+
+# values support several special equations:
+#     'a+bxc' becomes 'a a+b a+(2*b) a+(3*b) … a+(c*b)'
+#     Read this as series starts at a, then c further samples incrementing by b.
+#     'a-bxc' becomes 'a a-b a-(2*b) a-(3*b) … a-(c*b)'
+#     Read this as series starts at a, then c further samples decrementing by b (or incrementing by negative b).
+#    '_' represents a missing sample from scrape
+# Examples:
+#     1. '-2+4x3' becomes '-2 2 6 10' - series starts at -2, then 3 further samples incrementing by 4.
+#     2. ' 1-2x4' becomes '1 -1 -3 -5 -7' - series starts at 1, then 4 further samples decrementing by 2.
+#     3. ' 1x4' becomes '1 1 1 1 1' - shorthand for '1+0x4', series starts at 1, then 4 further samples incrementing by 0.
+#     4. ' 1 _x3' becomes '1 _ _ _ ' - the missing sample cannot increment, so 3 missing samples are produced by the '_x3' expression.
+values: <string>
+```
+
+#### `<alert_test_case>`
+
+vmalert by default adds `alertgroup` and `alertname` to the generated alerts and time series. 
+So you will need to specify both `groupname` and `alertname` under a single `<alert_test_case>`, 
+but no need to add them under `exp_alerts`.
+You can also pass `--disableAlertgroupLabel` to prevent vmalert from adding `alertgroup` label.
+
+```
+# The time elapsed from time=0s when this alerting rule should be checked.
+# Means this rule should be firing at this point, or shouldn't be firing if 'exp_alerts' is empty.
+eval_time: <duration>
+
+# Name of the group name to be tested.
+groupname: <string>
+
+# Name of the alert to be tested.
+alertname: <string>
+
+# List of the expected alerts that are firing under the given alertname at
+# the given evaluation time. If you want to test if an alerting rule should
+# not be firing, then you can mention only the fields above and leave 'exp_alerts' empty.
+exp_alerts:
+  [ - <alert> ]
+```
+
+#### `<alert>`
+
+```
+# These are the expanded labels and annotations of the expected alert.
+# Note: labels also include the labels of the sample associated with the alert
+exp_labels:
+  [ <labelname>: <string> ]
+exp_annotations:
+  [ <labelname>: <string> ]
+```
+
+#### `<metricsql_expr_test>`
+
+```
+# Expression to evaluate
+expr: <string>
+
+# The time elapsed from time=0s when this expression be evaluated.
+eval_time: <duration>
+
+# Expected samples at the given evaluation time.
+exp_samples:
+  [ - <sample> ]
+```
+
+#### `<sample>`
+
+```
+# Labels of the sample in usual series notation '<metric name>{<label name>=<label value>, ...}'
+# Examples:
+#      series_name{label1="value1", label2="value2"}
+#      go_goroutines{job="prometheus", instance="localhost:9090"}
+labels: <string>
+
+# The expected value of the Metricsql expression.
+value: <number>
+```
+
+### Example
+
+This is an example input file for unit testing which will pass.
+`test.yaml` is the test file which follows the syntax above and `alerts.yaml` contains the alerting rules.
+
+With `rules.yaml` in the same directory, run `./vmalert -unittestFile=./unittest/testdata/test.yaml`.
+
+#### `test.yaml`
+
+```
+rule_files:
+  - rules.yaml
+
+evaluation_interval: 1m
+
+tests:
+  - interval: 1m
+    input_series:
+      - series: 'up{job="prometheus", instance="localhost:9090"}'
+        values: "0+0x1440"
+
+    metricsql_expr_test:
+      - expr: suquery_interval_test
+        eval_time: 4m
+        exp_samples:
+          - labels: '{__name__="suquery_interval_test", datacenter="dc-123", instance="localhost:9090", job="prometheus"}'
+            value: 1
+
+    alert_rule_test:
+      - eval_time: 2h
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts:
+          - exp_labels:
+              job: prometheus
+              severity: page
+              instance: localhost:9090
+              datacenter: dc-123
+            exp_annotations:
+              summary: "Instance localhost:9090 down"
+              description: "localhost:9090 of job prometheus has been down for more than 5 minutes."
+
+      - eval_time: 0
+        groupname: group1
+        alertname: AlwaysFiring
+        exp_alerts:
+          - exp_labels:
+              datacenter: dc-123
+
+      - eval_time: 0
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts: []
+
+    external_labels:
+      datacenter: dc-123
+```
+
+#### `alerts.yaml`
+
+```
+# This is the rules file.
+
+groups:
+  - name: group1
+    rules:
+      - alert: InstanceDown
+        expr: up == 0
+        for: 5m
+        labels:
+          severity: page
+        annotations:
+          summary: "Instance {{ $labels.instance }} down"
+          description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 5 minutes."
+      - alert: AlwaysFiring
+        expr: 1
+
+  - name: group2
+    rules:
+      - record: job:test:count_over_time1m
+        expr: sum without(instance) (count_over_time(test[1m]))
+      - record: suquery_interval_test
+        expr: count_over_time(up[5m:])
+```
+
 ## Monitoring
 
 `vmalert` exports various metrics in Prometheus exposition format at `http://vmalert-host:8880/metrics` page.
@@ -1293,6 +1535,11 @@ The shortlist of configuration flags is the following:
      Path to file with TLS key if -tls is set. The provided key file is automatically re-read every second, so it can be dynamically updated
   -tlsMinVersion string
      Optional minimum TLS version to use for incoming requests over HTTPS if -tls is set. Supported values: TLS10, TLS11, TLS12, TLS13
+  -unittestFile array
+     Path to the unit test files. When set, vmalert starts in unit test mode and performs only tests on configured files.
+     Examples:
+      -unittestFile="./unittest/testdata/test1.yaml,./unittest/testdata/test2.yaml".
+     See more information here https://docs.victoriametrics.com/vmalert.html#unit-testing-for-rules.
   -version
      Show VictoriaMetrics version
 ```
diff --git a/docs/vmctl.md b/docs/vmctl.md
index 75893d311..0dae0a3ee 100644
--- a/docs/vmctl.md
+++ b/docs/vmctl.md
@@ -736,6 +736,9 @@ requires an Authentication header like `X-Scope-OrgID`. You can define it via th
 
 ## Migrating data from VictoriaMetrics
 
+The simplest way to migrate data between VM instances is to copy data.
+See more details [here](https://docs.victoriametrics.com/Single-server-VictoriaMetrics.html#data-migration).
+
 vmctl uses [native binary protocol](https://docs.victoriametrics.com/#how-to-export-data-in-native-format)
 (available since [1.42.0 release](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.42.0))
 o migrate data between VM instances: single to single, cluster to cluster, single to cluster and vice versa.
diff --git a/go.mod b/go.mod
index 89d35a675..45bc9ed2e 100644
--- a/go.mod
+++ b/go.mod
@@ -39,7 +39,10 @@ require (
 	gopkg.in/yaml.v2 v2.4.0
 )
 
-require github.com/bmatcuk/doublestar/v4 v4.6.0
+require (
+	github.com/bmatcuk/doublestar/v4 v4.6.0
+	golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1
+)
 
 require (
 	cloud.google.com/go v0.110.6 // indirect
@@ -115,7 +118,6 @@ require (
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/goleak v1.2.1 // indirect
 	golang.org/x/crypto v0.11.0 // indirect
-	golang.org/x/exp v0.0.0-20230713183714-613f0c0eb8a1 // indirect
 	golang.org/x/sync v0.3.0 // indirect
 	golang.org/x/text v0.11.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
diff --git a/lib/logstorage/log_rows.go b/lib/logstorage/log_rows.go
index 6efbc8a3d..ce759b85a 100644
--- a/lib/logstorage/log_rows.go
+++ b/lib/logstorage/log_rows.go
@@ -230,6 +230,7 @@ func (lr *LogRows) GetRowString(idx int) string {
 // GetLogRows returns LogRows from the pool for the given streamFields.
 //
 // streamFields is a set of field names, which must be associated with the stream.
+// ignoreFields is a set of field names, which must be ignored during data ingestion.
 //
 // Return back it to the pool with PutLogRows() when it is no longer needed.
 func GetLogRows(streamFields, ignoreFields []string) *LogRows {
diff --git a/lib/logstorage/tenant_id.go b/lib/logstorage/tenant_id.go
index 49873dcee..1d406b875 100644
--- a/lib/logstorage/tenant_id.go
+++ b/lib/logstorage/tenant_id.go
@@ -87,7 +87,7 @@ func GetTenantIDFromString(s string) (TenantID, error) {
 	if colon < 0 {
 		account, err := getUint32FromString(s)
 		if err != nil {
-			return tenantID, fmt.Errorf("cannot parse %q as TenantID: %w", s, err)
+			return tenantID, fmt.Errorf("cannot parse accountID from %q: %w", s, err)
 		}
 		tenantID.AccountID = account
 
@@ -96,13 +96,13 @@ func GetTenantIDFromString(s string) (TenantID, error) {
 
 	account, err := getUint32FromString(s[:colon])
 	if err != nil {
-		return tenantID, fmt.Errorf("cannot parse %q as TenantID: %w", s, err)
+		return tenantID, fmt.Errorf("cannot parse accountID part from %q: %w", s, err)
 	}
 	tenantID.AccountID = account
 
 	project, err := getUint32FromString(s[colon+1:])
 	if err != nil {
-		return tenantID, fmt.Errorf("cannot parse %q as TenantID: %w", s, err)
+		return tenantID, fmt.Errorf("cannot parse projectID part from %q: %w", s, err)
 	}
 	tenantID.ProjectID = project
 
diff --git a/lib/promutils/duration.go b/lib/promutils/duration.go
index 2fc57c52e..8c5abe807 100644
--- a/lib/promutils/duration.go
+++ b/lib/promutils/duration.go
@@ -45,6 +45,14 @@ func (pd *Duration) Duration() time.Duration {
 	return pd.D
 }
 
+// ParseTime returns time for pd.
+func (pd *Duration) ParseTime() time.Time {
+	if pd == nil {
+		return time.Time{}
+	}
+	return time.UnixMilli(pd.Duration().Milliseconds())
+}
+
 // ParseDuration parses duration string in Prometheus format
 func ParseDuration(s string) (time.Duration, error) {
 	ms, err := metricsql.DurationValue(s, 0)
diff --git a/lib/slicesutil/resize.go b/lib/slicesutil/resize.go
deleted file mode 100644
index b650b6ca9..000000000
--- a/lib/slicesutil/resize.go
+++ /dev/null
@@ -1,20 +0,0 @@
-package slicesutil
-
-import "math/bits"
-
-// ResizeNoCopyMayOverallocate resizes dst to minimum n bytes and returns the resized buffer (which may be newly allocated).
-//
-// If newly allocated buffer is returned then b contents isn't copied to it.
-func ResizeNoCopyMayOverallocate[T any](dst []T, n int) []T {
-	if n <= cap(dst) {
-		return dst[:n]
-	}
-	nNew := roundToNearestPow2(n)
-	dstNew := make([]T, nNew)
-	return dstNew[:n]
-}
-
-func roundToNearestPow2(n int) int {
-	pow2 := uint8(bits.Len(uint(n - 1)))
-	return 1 << pow2
-}
diff --git a/lib/streamaggr/histogram_bucket.go b/lib/streamaggr/histogram_bucket.go
index 7d844d09e..7361b6238 100644
--- a/lib/streamaggr/histogram_bucket.go
+++ b/lib/streamaggr/histogram_bucket.go
@@ -12,7 +12,7 @@ import (
 type histogramBucketAggrState struct {
 	m sync.Map
 
-	intervalSecs uint64
+	stalenessInterval uint64
 }
 
 type histogramBucketStateValue struct {
@@ -22,16 +22,15 @@ type histogramBucketStateValue struct {
 	deleted        bool
 }
 
-func newHistogramBucketAggrState(interval time.Duration) *histogramBucketAggrState {
-	intervalSecs := uint64(interval.Seconds() + 1)
+func newHistogramBucketAggrState(stalenessInterval time.Duration) *histogramBucketAggrState {
 	return &histogramBucketAggrState{
-		intervalSecs: intervalSecs,
+		stalenessInterval: uint64(stalenessInterval.Seconds()),
 	}
 }
 
 func (as *histogramBucketAggrState) pushSample(inputKey, outputKey string, value float64) {
 	currentTime := fasttime.UnixTimestamp()
-	deleteDeadline := currentTime + 2*as.intervalSecs
+	deleteDeadline := currentTime + as.stalenessInterval
 
 again:
 	v, ok := as.m.Load(outputKey)
diff --git a/lib/streamaggr/increase.go b/lib/streamaggr/increase.go
index e21bfedad..8a43e6443 100644
--- a/lib/streamaggr/increase.go
+++ b/lib/streamaggr/increase.go
@@ -12,7 +12,7 @@ type increaseAggrState struct {
 	m sync.Map
 
 	ignoreInputDeadline uint64
-	intervalSecs        uint64
+	stalenessInterval   uint64
 }
 
 type increaseStateValue struct {
@@ -23,18 +23,18 @@ type increaseStateValue struct {
 	deleted        bool
 }
 
-func newIncreaseAggrState(interval time.Duration) *increaseAggrState {
+func newIncreaseAggrState(interval time.Duration, stalenessInterval time.Duration) *increaseAggrState {
 	currentTime := fasttime.UnixTimestamp()
 	intervalSecs := uint64(interval.Seconds() + 1)
 	return &increaseAggrState{
 		ignoreInputDeadline: currentTime + intervalSecs,
-		intervalSecs:        intervalSecs,
+		stalenessInterval:   uint64(stalenessInterval.Seconds()),
 	}
 }
 
 func (as *increaseAggrState) pushSample(inputKey, outputKey string, value float64) {
 	currentTime := fasttime.UnixTimestamp()
-	deleteDeadline := currentTime + 2*as.intervalSecs
+	deleteDeadline := currentTime + as.stalenessInterval
 
 again:
 	v, ok := as.m.Load(outputKey)
diff --git a/lib/streamaggr/streamaggr.go b/lib/streamaggr/streamaggr.go
index 03a0e0210..03ca4f5ed 100644
--- a/lib/streamaggr/streamaggr.go
+++ b/lib/streamaggr/streamaggr.go
@@ -79,6 +79,10 @@ type Config struct {
 	// Interval is the interval between aggregations.
 	Interval string `yaml:"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, increase and histogram_bucket.
+	StalenessInterval string `yaml:"staleness_interval,omitempty"`
+
 	// Outputs is a list of output aggregate functions to produce.
 	//
 	// The following names are allowed:
@@ -254,6 +258,18 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 		return nil, fmt.Errorf("the minimum supported aggregation interval is 1s; got %s", interval)
 	}
 
+	// check cfg.StalenessInterval
+	stalenessInterval := interval * 2
+	if cfg.StalenessInterval != "" {
+		stalenessInterval, err = time.ParseDuration(cfg.StalenessInterval)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse `staleness_interval: %q`: %w", cfg.StalenessInterval, err)
+		}
+		if stalenessInterval < interval {
+			return nil, fmt.Errorf("staleness_interval cannot be less than interval (%s); got %s", cfg.Interval, cfg.StalenessInterval)
+		}
+	}
+
 	// initialize input_relabel_configs and output_relabel_configs
 	inputRelabeling, err := promrelabel.ParseRelabelConfigs(cfg.InputRelabelConfigs)
 	if err != nil {
@@ -308,9 +324,9 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 		}
 		switch output {
 		case "total":
-			aggrStates[i] = newTotalAggrState(interval)
+			aggrStates[i] = newTotalAggrState(interval, stalenessInterval)
 		case "increase":
-			aggrStates[i] = newIncreaseAggrState(interval)
+			aggrStates[i] = newIncreaseAggrState(interval, stalenessInterval)
 		case "count_series":
 			aggrStates[i] = newCountSeriesAggrState()
 		case "count_samples":
@@ -330,7 +346,7 @@ func newAggregator(cfg *Config, pushFunc PushFunc, dedupInterval time.Duration)
 		case "stdvar":
 			aggrStates[i] = newStdvarAggrState()
 		case "histogram_bucket":
-			aggrStates[i] = newHistogramBucketAggrState(interval)
+			aggrStates[i] = newHistogramBucketAggrState(stalenessInterval)
 		default:
 			return nil, fmt.Errorf("unsupported output=%q; supported values: %s; "+
 				"see https://docs.victoriametrics.com/vmagent.html#stream-aggregation", output, supportedOutputs)
diff --git a/lib/streamaggr/total.go b/lib/streamaggr/total.go
index b2e935acc..92b1cd2bd 100644
--- a/lib/streamaggr/total.go
+++ b/lib/streamaggr/total.go
@@ -13,7 +13,7 @@ type totalAggrState struct {
 	m sync.Map
 
 	ignoreInputDeadline uint64
-	intervalSecs        uint64
+	stalenessInterval   uint64
 }
 
 type totalStateValue struct {
@@ -29,18 +29,18 @@ type lastValueState struct {
 	deleteDeadline uint64
 }
 
-func newTotalAggrState(interval time.Duration) *totalAggrState {
+func newTotalAggrState(interval time.Duration, stalenessInterval time.Duration) *totalAggrState {
 	currentTime := fasttime.UnixTimestamp()
 	intervalSecs := uint64(interval.Seconds() + 1)
 	return &totalAggrState{
 		ignoreInputDeadline: currentTime + intervalSecs,
-		intervalSecs:        intervalSecs,
+		stalenessInterval:   uint64(stalenessInterval.Seconds()),
 	}
 }
 
 func (as *totalAggrState) pushSample(inputKey, outputKey string, value float64) {
 	currentTime := fasttime.UnixTimestamp()
-	deleteDeadline := currentTime + as.intervalSecs + (as.intervalSecs >> 1)
+	deleteDeadline := currentTime + as.stalenessInterval
 
 again:
 	v, ok := as.m.Load(outputKey)


Test patch:
diff --git a/app/vlinsert/loki/loki_json_test.go b/app/vlinsert/loki/loki_json_test.go
index 5588035c6..93cf8652a 100644
--- a/app/vlinsert/loki/loki_json_test.go
+++ b/app/vlinsert/loki/loki_json_test.go
@@ -1,99 +1,130 @@
 package loki
 
 import (
-	"reflect"
+	"fmt"
 	"strings"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
-func TestProcessJSONRequest(t *testing.T) {
-	type item struct {
-		ts     int64
-		fields []logstorage.Field
-	}
-
-	same := func(s string, expected []item) {
+func TestParseJSONRequestFailure(t *testing.T) {
+	f := func(s string) {
 		t.Helper()
-		r := strings.NewReader(s)
-		actual := make([]item, 0)
-		n, err := processJSONRequest(r, func(timestamp int64, fields []logstorage.Field) {
-			actual = append(actual, item{
-				ts:     timestamp,
-				fields: fields,
-			})
+		n, err := parseJSONRequest([]byte(s), func(timestamp int64, fields []logstorage.Field) {
+			t.Fatalf("unexpected call to parseJSONRequest callback!")
 		})
-
-		if err != nil {
-			t.Fatalf("unexpected error: %s", err)
-		}
-
-		if len(actual) != len(expected) || n != len(expected) {
-			t.Fatalf("unexpected len(actual)=%d; expecting %d", len(actual), len(expected))
+		if err == nil {
+			t.Fatalf("expecting non-nil error")
 		}
-
-		for i, actualItem := range actual {
-			expectedItem := expected[i]
-			if actualItem.ts != expectedItem.ts {
-				t.Fatalf("unexpected timestamp for item #%d; got %d; expecting %d", i, actualItem.ts, expectedItem.ts)
-			}
-			if !reflect.DeepEqual(actualItem.fields, expectedItem.fields) {
-				t.Fatalf("unexpected fields for item #%d; got %v; expecting %v", i, actualItem.fields, expectedItem.fields)
-			}
+		if n != 0 {
+			t.Fatalf("unexpected number of parsed lines: %d; want 0", n)
 		}
 	}
+	f(``)
 
-	fail := func(s string) {
-		t.Helper()
-		r := strings.NewReader(s)
-		actual := make([]item, 0)
-		_, err := processJSONRequest(r, func(timestamp int64, fields []logstorage.Field) {
-			actual = append(actual, item{
-				ts:     timestamp,
-				fields: fields,
-			})
-		})
+	// Invalid json
+	f(`{}`)
+	f(`[]`)
+	f(`"foo"`)
+	f(`123`)
 
-		if err == nil {
-			t.Fatalf("expected to fail with body: %q", s)
-		}
+	// invalid type for `streams` item
+	f(`{"streams":123}`)
 
-	}
+	// Missing `values` item
+	f(`{"streams":[{}]}`)
 
-	same(`{"streams":[{"stream":{"foo":"bar"},"values":[["1577836800000000000","baz"]]}]}`, []item{
-		{
-			ts: 1577836800000000000,
-			fields: []logstorage.Field{
-				{
-					Name:  "foo",
-					Value: "bar",
-				},
-				{
-					Name:  "_msg",
-					Value: "baz",
-				},
-			},
-		},
-	})
+	// Invalid type for `values` item
+	f(`{"streams":[{"values":"foobar"}]}`)
+
+	// Invalid type for `stream` item
+	f(`{"streams":[{"stream":[],"values":[]}]}`)
+
+	// Invalid type for `values` individual item
+	f(`{"streams":[{"values":[123]}]}`)
+
+	// Invalid length of `values` individual item
+	f(`{"streams":[{"values":[[]]}]}`)
+	f(`{"streams":[{"values":[["123"]]}]}`)
+	f(`{"streams":[{"values":[["123","456","789"]]}]}`)
+
+	// Invalid type for timestamp inside `values` individual item
+	f(`{"streams":[{"values":[[123,"456"]}]}`)
 
-	fail(``)
-	fail(`{"streams":[{"stream":{"foo" = "bar"},"values":[["1577836800000000000","baz"]]}]}`)
-	fail(`{"streams":[{"stream":{"foo": "bar"}`)
+	// Invalid type for log message
+	f(`{"streams":[{"values":[["123",1234]]}]}`)
 }
 
-func Test_parseLokiTimestamp(t *testing.T) {
-	f := func(s string, expected int64) {
+func TestParseJSONRequestSuccess(t *testing.T) {
+	f := func(s string, resultExpected string) {
 		t.Helper()
-		actual, err := parseLokiTimestamp(s)
+		var lines []string
+		n, err := parseJSONRequest([]byte(s), func(timestamp int64, fields []logstorage.Field) {
+			var a []string
+			for _, f := range fields {
+				a = append(a, f.String())
+			}
+			line := fmt.Sprintf("_time:%d %s", timestamp, strings.Join(a, " "))
+			lines = append(lines, line)
+		})
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if actual != expected {
-			t.Fatalf("unexpected timestamp; got %d; expecting %d", actual, expected)
+		if n != len(lines) {
+			t.Fatalf("unexpected number of lines parsed; got %d; want %d", n, len(lines))
+		}
+		result := strings.Join(lines, "\n")
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
 		}
 	}
 
-	f("1687510468000000000", 1687510468000000000)
-	f("1577836800000000000", 1577836800000000000)
+	// Empty streams
+	f(`{"streams":[]}`, ``)
+	f(`{"streams":[{"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{},"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{"foo":"bar"},"values":[]}]}`, ``)
+
+	// Empty stream labels
+	f(`{"streams":[{"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+	f(`{"streams":[{"stream":{},"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+
+	// Non-empty stream labels
+	f(`{"streams":[{"stream":{
+	"label1": "value1",
+	"label2": "value2"
+},"values":[
+	["1577836800000000001", "foo bar"],
+	["1477836900005000002", "abc"],
+	["147.78369e9", "foobar"]
+]}]}`, `_time:1577836800000000001 "label1":"value1" "label2":"value2" "_msg":"foo bar"
+_time:1477836900005000002 "label1":"value1" "label2":"value2" "_msg":"abc"
+_time:147783690000 "label1":"value1" "label2":"value2" "_msg":"foobar"`)
+
+	// Multiple streams
+	f(`{
+	"streams": [
+		{
+			"stream": {
+				"foo": "bar",
+				"a": "b"
+			},
+			"values": [
+				["1577836800000000001", "foo bar"],
+				["1577836900005000002", "abc"]
+			]
+		},
+		{
+			"stream": {
+				"x": "y"
+			},
+			"values": [
+				["1877836900005000002", "yx"]
+			]
+		}
+	]
+}`, `_time:1577836800000000001 "foo":"bar" "a":"b" "_msg":"foo bar"
+_time:1577836900005000002 "foo":"bar" "a":"b" "_msg":"abc"
+_time:1877836900005000002 "x":"y" "_msg":"yx"`)
 }
diff --git a/app/vlinsert/loki/loki_json_timing_test.go b/app/vlinsert/loki/loki_json_timing_test.go
index 7da50b2fe..9c51f593a 100644
--- a/app/vlinsert/loki/loki_json_timing_test.go
+++ b/app/vlinsert/loki/loki_json_timing_test.go
@@ -3,71 +3,76 @@ package loki
 import (
 	"fmt"
 	"strconv"
-	"strings"
 	"testing"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
-func BenchmarkProcessJSONRequest(b *testing.B) {
+func BenchmarkParseJSONRequest(b *testing.B) {
 	for _, streams := range []int{5, 10} {
 		for _, rows := range []int{100, 1000} {
 			for _, labels := range []int{10, 50} {
 				b.Run(fmt.Sprintf("streams_%d/rows_%d/labels_%d", streams, rows, labels), func(b *testing.B) {
-					benchmarkProcessJSONRequest(b, streams, rows, labels)
+					benchmarkParseJSONRequest(b, streams, rows, labels)
 				})
 			}
 		}
 	}
 }
 
-func benchmarkProcessJSONRequest(b *testing.B, streams, rows, labels int) {
-	s := getJSONBody(streams, rows, labels)
+func benchmarkParseJSONRequest(b *testing.B, streams, rows, labels int) {
 	b.ReportAllocs()
-	b.SetBytes(int64(len(s)))
+	b.SetBytes(int64(streams * rows))
 	b.RunParallel(func(pb *testing.PB) {
+		data := getJSONBody(streams, rows, labels)
 		for pb.Next() {
-			_, err := processJSONRequest(strings.NewReader(s), func(timestamp int64, fields []logstorage.Field) {})
+			_, err := parseJSONRequest(data, func(timestamp int64, fields []logstorage.Field) {})
 			if err != nil {
-				b.Fatalf("unexpected error: %s", err)
+				panic(fmt.Errorf("unexpected error: %s", err))
 			}
 		}
 	})
 }
 
-func getJSONBody(streams, rows, labels int) string {
-	body := `{"streams":[`
+func getJSONBody(streams, rows, labels int) []byte {
+	body := append([]byte{}, `{"streams":[`...)
 	now := time.Now().UnixNano()
 	valuePrefix := fmt.Sprintf(`["%d","value_`, now)
 
 	for i := 0; i < streams; i++ {
-		body += `{"stream":{`
+		body = append(body, `{"stream":{`...)
 
 		for j := 0; j < labels; j++ {
-			body += `"label_` + strconv.Itoa(j) + `":"value_` + strconv.Itoa(j) + `"`
+			body = append(body, `"label_`...)
+			body = strconv.AppendInt(body, int64(j), 10)
+			body = append(body, `":"value_`...)
+			body = strconv.AppendInt(body, int64(j), 10)
+			body = append(body, '"')
 			if j < labels-1 {
-				body += `,`
+				body = append(body, ',')
 			}
 
 		}
-		body += `}, "values":[`
+		body = append(body, `}, "values":[`...)
 
 		for j := 0; j < rows; j++ {
-			body += valuePrefix + strconv.Itoa(j) + `"]`
+			body = append(body, valuePrefix...)
+			body = strconv.AppendInt(body, int64(j), 10)
+			body = append(body, `"]`...)
 			if j < rows-1 {
-				body += `,`
+				body = append(body, ',')
 			}
 		}
 
-		body += `]}`
+		body = append(body, `]}`...)
 		if i < streams-1 {
-			body += `,`
+			body = append(body, ',')
 		}
 
 	}
 
-	body += `]}`
+	body = append(body, `]}`...)
 
 	return body
 }
diff --git a/app/vlinsert/loki/loki_protobuf_test.go b/app/vlinsert/loki/loki_protobuf_test.go
index e590668ee..f6eb5f0ec 100644
--- a/app/vlinsert/loki/loki_protobuf_test.go
+++ b/app/vlinsert/loki/loki_protobuf_test.go
@@ -1,50 +1,171 @@
 package loki
 
 import (
-	"bytes"
-	"strconv"
+	"fmt"
+	"strings"
 	"testing"
 	"time"
 
-	"github.com/golang/snappy"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/golang/snappy"
 )
 
-func TestProcessProtobufRequest(t *testing.T) {
-	body := getProtobufBody(5, 5, 5)
+func TestParseProtobufRequestSuccess(t *testing.T) {
+	f := func(s string, resultExpected string) {
+		t.Helper()
+		var pr PushRequest
+		n, err := parseJSONRequest([]byte(s), func(timestamp int64, fields []logstorage.Field) {
+			msg := ""
+			for _, f := range fields {
+				if f.Name == "_msg" {
+					msg = f.Value
+				}
+			}
+			var a []string
+			for _, f := range fields {
+				if f.Name == "_msg" {
+					continue
+				}
+				item := fmt.Sprintf("%s=%q", f.Name, f.Value)
+				a = append(a, item)
+			}
+			labels := "{" + strings.Join(a, ", ") + "}"
+			pr.Streams = append(pr.Streams, Stream{
+				Labels: labels,
+				Entries: []Entry{
+					{
+						Timestamp: time.Unix(0, timestamp),
+						Line:      msg,
+					},
+				},
+			})
+		})
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if n != len(pr.Streams) {
+			t.Fatalf("unexpected number of streams; got %d; want %d", len(pr.Streams), n)
+		}
+
+		data, err := pr.Marshal()
+		if err != nil {
+			t.Fatalf("unexpected error when marshaling PushRequest: %s", err)
+		}
+		encodedData := snappy.Encode(nil, data)
 
-	reader := bytes.NewReader(body)
-	_, err := processProtobufRequest(reader, func(timestamp int64, fields []logstorage.Field) {})
-	if err != nil {
-		t.Fatalf("unexpected error: %s", err)
+		var lines []string
+		n, err = parseProtobufRequest(encodedData, func(timestamp int64, fields []logstorage.Field) {
+			var a []string
+			for _, f := range fields {
+				a = append(a, f.String())
+			}
+			line := fmt.Sprintf("_time:%d %s", timestamp, strings.Join(a, " "))
+			lines = append(lines, line)
+		})
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if n != len(lines) {
+			t.Fatalf("unexpected number of lines parsed; got %d; want %d", n, len(lines))
+		}
+		result := strings.Join(lines, "\n")
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
+		}
 	}
-}
 
-func getProtobufBody(streams, rows, labels int) []byte {
-	var pr PushRequest
+	// Empty streams
+	f(`{"streams":[]}`, ``)
+	f(`{"streams":[{"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{},"values":[]}]}`, ``)
+	f(`{"streams":[{"stream":{"foo":"bar"},"values":[]}]}`, ``)
 
-	for i := 0; i < streams; i++ {
-		var st Stream
+	// Empty stream labels
+	f(`{"streams":[{"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
+	f(`{"streams":[{"stream":{},"values":[["1577836800000000001", "foo bar"]]}]}`, `_time:1577836800000000001 "_msg":"foo bar"`)
 
-		st.Labels = `{`
-		for j := 0; j < labels; j++ {
-			st.Labels += `label_` + strconv.Itoa(j) + `="value_` + strconv.Itoa(j) + `"`
-			if j < labels-1 {
-				st.Labels += `,`
-			}
+	// Non-empty stream labels
+	f(`{"streams":[{"stream":{
+	"label1": "value1",
+	"label2": "value2"
+},"values":[
+	["1577836800000000001", "foo bar"],
+	["1477836900005000002", "abc"],
+	["147.78369e9", "foobar"]
+]}]}`, `_time:1577836800000000001 "label1":"value1" "label2":"value2" "_msg":"foo bar"
+_time:1477836900005000002 "label1":"value1" "label2":"value2" "_msg":"abc"
+_time:147783690000 "label1":"value1" "label2":"value2" "_msg":"foobar"`)
+
+	// Multiple streams
+	f(`{
+	"streams": [
+		{
+			"stream": {
+				"foo": "bar",
+				"a": "b"
+			},
+			"values": [
+				["1577836800000000001", "foo bar"],
+				["1577836900005000002", "abc"]
+			]
+		},
+		{
+			"stream": {
+				"x": "y"
+			},
+			"values": [
+				["1877836900005000002", "yx"]
+			]
 		}
-		st.Labels += `}`
+	]
+}`, `_time:1577836800000000001 "foo":"bar" "a":"b" "_msg":"foo bar"
+_time:1577836900005000002 "foo":"bar" "a":"b" "_msg":"abc"
+_time:1877836900005000002 "x":"y" "_msg":"yx"`)
+}
 
-		for j := 0; j < rows; j++ {
-			st.Entries = append(st.Entries, Entry{Timestamp: time.Now(), Line: "value_" + strconv.Itoa(j)})
+func TestParsePromLabelsSuccess(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+		fields, err := parsePromLabels(nil, s)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
 		}
 
-		pr.Streams = append(pr.Streams, st)
+		var a []string
+		for _, f := range fields {
+			a = append(a, fmt.Sprintf("%s=%q", f.Name, f.Value))
+		}
+		result := "{" + strings.Join(a, ", ") + "}"
+		if result != s {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, s)
+		}
 	}
 
-	body, _ := pr.Marshal()
-	encodedBody := snappy.Encode(nil, body)
+	f("{}")
+	f(`{foo="bar"}`)
+	f(`{foo="bar", baz="x", y="z"}`)
+	f(`{foo="ba\"r\\z\n", a="", b="\"\\"}`)
+}
+
+func TestParsePromLabelsFailure(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+		fields, err := parsePromLabels(nil, s)
+		if err == nil {
+			t.Fatalf("expecting non-nil error")
+		}
+		if len(fields) > 0 {
+			t.Fatalf("unexpected non-empty fields: %s", fields)
+		}
+	}
 
-	return encodedBody
+	f("")
+	f("{")
+	f(`{foo}`)
+	f(`{foo=bar}`)
+	f(`{foo="bar}`)
+	f(`{foo="ba\",r}`)
+	f(`{foo="bar" baz="aa"}`)
+	f(`foobar`)
+	f(`foo{bar="baz"}`)
 }
diff --git a/app/vlinsert/loki/loki_protobuf_timing_test.go b/app/vlinsert/loki/loki_protobuf_timing_test.go
index 0da5fe741..18f5b89ef 100644
--- a/app/vlinsert/loki/loki_protobuf_timing_test.go
+++ b/app/vlinsert/loki/loki_protobuf_timing_test.go
@@ -1,35 +1,65 @@
 package loki
 
 import (
-	"bytes"
 	"fmt"
+	"strconv"
 	"testing"
+	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
+	"github.com/golang/snappy"
 )
 
-func BenchmarkProcessProtobufRequest(b *testing.B) {
+func BenchmarkParseProtobufRequest(b *testing.B) {
 	for _, streams := range []int{5, 10} {
 		for _, rows := range []int{100, 1000} {
 			for _, labels := range []int{10, 50} {
 				b.Run(fmt.Sprintf("streams_%d/rows_%d/labels_%d", streams, rows, labels), func(b *testing.B) {
-					benchmarkProcessProtobufRequest(b, streams, rows, labels)
+					benchmarkParseProtobufRequest(b, streams, rows, labels)
 				})
 			}
 		}
 	}
 }
 
-func benchmarkProcessProtobufRequest(b *testing.B, streams, rows, labels int) {
-	body := getProtobufBody(streams, rows, labels)
+func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
 	b.ReportAllocs()
-	b.SetBytes(int64(len(body)))
+	b.SetBytes(int64(streams * rows))
 	b.RunParallel(func(pb *testing.PB) {
+		body := getProtobufBody(streams, rows, labels)
 		for pb.Next() {
-			_, err := processProtobufRequest(bytes.NewBuffer(body), func(timestamp int64, fields []logstorage.Field) {})
+			_, err := parseProtobufRequest(body, func(timestamp int64, fields []logstorage.Field) {})
 			if err != nil {
-				b.Fatalf("unexpected error: %s", err)
+				panic(fmt.Errorf("unexpected error: %s", err))
 			}
 		}
 	})
 }
+
+func getProtobufBody(streams, rows, labels int) []byte {
+	var pr PushRequest
+
+	for i := 0; i < streams; i++ {
+		var st Stream
+
+		st.Labels = `{`
+		for j := 0; j < labels; j++ {
+			st.Labels += `label_` + strconv.Itoa(j) + `="value_` + strconv.Itoa(j) + `"`
+			if j < labels-1 {
+				st.Labels += `,`
+			}
+		}
+		st.Labels += `}`
+
+		for j := 0; j < rows; j++ {
+			st.Entries = append(st.Entries, Entry{Timestamp: time.Now(), Line: "value_" + strconv.Itoa(j)})
+		}
+
+		pr.Streams = append(pr.Streams, st)
+	}
+
+	body, _ := pr.Marshal()
+	encodedBody := snappy.Encode(nil, body)
+
+	return encodedBody
+}
diff --git a/app/vmalert/manager_test.go b/app/vmalert/manager_test.go
index 230808e3c..8ef07baf7 100644
--- a/app/vmalert/manager_test.go
+++ b/app/vmalert/manager_test.go
@@ -257,7 +257,7 @@ func TestManagerUpdate(t *testing.T) {
 func TestManagerUpdateNegative(t *testing.T) {
 	testCases := []struct {
 		notifiers []notifier.Notifier
-		rw        *remotewrite.Client
+		rw        remotewrite.RWClient
 		cfg       config.Group
 		expErr    string
 	}{
diff --git a/app/vmalert/remotewrite/debug_client_test.go b/app/vmalert/remotewrite/debug_client_test.go
new file mode 100644
index 000000000..40c498cbe
--- /dev/null
+++ b/app/vmalert/remotewrite/debug_client_test.go
@@ -0,0 +1,50 @@
+package remotewrite
+
+import (
+	"testing"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+)
+
+func TestDebugClient_Push(t *testing.T) {
+	testSrv := newRWServer()
+	oldAddr := *addr
+	*addr = testSrv.URL
+	defer func() {
+		*addr = oldAddr
+	}()
+
+	client, err := NewDebugClient()
+	if err != nil {
+		t.Fatalf("failed to create debug client: %s", err)
+	}
+
+	const rowsN = 100
+	var sent int
+	for i := 0; i < rowsN; i++ {
+		s := prompbmarshal.TimeSeries{
+			Samples: []prompbmarshal.Sample{{
+				Value:     float64(i),
+				Timestamp: time.Now().Unix(),
+			}},
+		}
+		err := client.Push(s)
+		if err != nil {
+			t.Fatalf("unexpected err: %s", err)
+		}
+		if err == nil {
+			sent++
+		}
+	}
+	if sent == 0 {
+		t.Fatalf("0 series sent")
+	}
+	if err := client.Close(); err != nil {
+		t.Fatalf("failed to close client: %s", err)
+	}
+	got := testSrv.accepted()
+	if got != sent {
+		t.Fatalf("expected to have %d series; got %d", sent, got)
+	}
+}
diff --git a/app/vmalert/unitest_test.go b/app/vmalert/unitest_test.go
new file mode 100644
index 000000000..70f3b46db
--- /dev/null
+++ b/app/vmalert/unitest_test.go
@@ -0,0 +1,40 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestUnitRule(t *testing.T) {
+	testCases := []struct {
+		name              string
+		disableGroupLabel bool
+		files             []string
+		failed            bool
+	}{
+		{
+			name:   "run multi files",
+			files:  []string{"./unittest/testdata/test1.yaml", "./unittest/testdata/test2.yaml"},
+			failed: false,
+		},
+		{
+			name:              "disable group label",
+			disableGroupLabel: true,
+			files:             []string{"./unittest/testdata/disable-group-label.yaml"},
+			failed:            false,
+		},
+		{
+			name:   "failing test",
+			files:  []string{"./unittest/testdata/failed-test.yaml"},
+			failed: true,
+		},
+	}
+	for _, tc := range testCases {
+		oldFlag := *disableAlertGroupLabel
+		*disableAlertGroupLabel = tc.disableGroupLabel
+		fail := unitRule(tc.files...)
+		if fail != tc.failed {
+			t.Fatalf("failed to test %s, expect %t, got %t", tc.name, tc.failed, fail)
+		}
+		*disableAlertGroupLabel = oldFlag
+	}
+}
diff --git a/app/vmalert/unittest.go b/app/vmalert/unittest.go
new file mode 100644
index 000000000..b51bdb5af
--- /dev/null
+++ b/app/vmalert/unittest.go
@@ -0,0 +1,436 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sort"
+	"time"
+
+	"gopkg.in/yaml.v2"
+
+	vmalertconfig "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/unittest"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/promremotewrite"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/prometheus"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var (
+	storagePath string
+	// insert series from 1970-01-01T00:00:00
+	testStartTime = time.Unix(0, 0).UTC()
+
+	testPromWriteHTTPPath = "http://127.0.0.1" + *httpListenAddr + "/api/v1/write"
+	testDataSourcePath    = "http://127.0.0.1" + *httpListenAddr + "/prometheus"
+	testRemoteWritePath   = "http://127.0.0.1" + *httpListenAddr
+	testHealthHTTPPath    = "http://127.0.0.1" + *httpListenAddr + "/health"
+)
+
+const (
+	testStoragePath = "vmalert-unittest"
+	testLogLevel    = "ERROR"
+)
+
+func unitRule(files ...string) bool {
+	storagePath = filepath.Join(os.TempDir(), testStoragePath)
+	processFlags()
+	vminsert.Init()
+	vmselect.Init()
+	// storagePath will be created again when closing vmselect, so remove it again.
+	defer fs.MustRemoveAll(storagePath)
+	defer vminsert.Stop()
+	defer vmselect.Stop()
+	return rulesUnitTest(files...)
+}
+
+func rulesUnitTest(files ...string) bool {
+	var failed bool
+	for _, f := range files {
+		if err := ruleUnitTest(f); err != nil {
+			fmt.Println("  FAILED")
+			fmt.Printf("\nfailed to run unit test for file %q: \n%s", f, err)
+			failed = true
+		} else {
+			fmt.Println("  SUCCESS")
+		}
+	}
+	return failed
+}
+
+func ruleUnitTest(filename string) []error {
+	fmt.Println("\nUnit Testing: ", filename)
+	b, err := os.ReadFile(filename)
+	if err != nil {
+		return []error{fmt.Errorf("failed to read file: %w", err)}
+	}
+
+	var unitTestInp unitTestFile
+	if err := yaml.UnmarshalStrict(b, &unitTestInp); err != nil {
+		return []error{fmt.Errorf("failed to unmarshal file: %w", err)}
+	}
+	if err := resolveAndGlobFilepaths(filepath.Dir(filename), &unitTestInp); err != nil {
+		return []error{fmt.Errorf("failed to resolve path for `rule_files`: %w", err)}
+	}
+
+	if unitTestInp.EvaluationInterval.Duration() == 0 {
+		fmt.Println("evaluation_interval set to 1m by default")
+		unitTestInp.EvaluationInterval = &promutils.Duration{D: 1 * time.Minute}
+	}
+
+	groupOrderMap := make(map[string]int)
+	for i, gn := range unitTestInp.GroupEvalOrder {
+		if _, ok := groupOrderMap[gn]; ok {
+			return []error{fmt.Errorf("group name repeated in `group_eval_order`: %s", gn)}
+		}
+		groupOrderMap[gn] = i
+	}
+
+	testGroups, err := vmalertconfig.Parse(unitTestInp.RuleFiles, nil, true)
+	if err != nil {
+		return []error{fmt.Errorf("failed to parse `rule_files`: %w", err)}
+	}
+
+	var errs []error
+	for _, t := range unitTestInp.Tests {
+		if err := verifyTestGroup(t); err != nil {
+			errs = append(errs, err)
+			continue
+		}
+		testErrs := t.test(unitTestInp.EvaluationInterval.Duration(), groupOrderMap, testGroups)
+		errs = append(errs, testErrs...)
+	}
+
+	if len(errs) > 0 {
+		return errs
+	}
+	return nil
+}
+
+func verifyTestGroup(group testGroup) error {
+	var testGroupName string
+	if group.TestGroupName != "" {
+		testGroupName = fmt.Sprintf("testGroupName: %s\n", group.TestGroupName)
+	}
+	for _, at := range group.AlertRuleTests {
+		if at.Alertname == "" {
+			return fmt.Errorf("\n%s    missing required filed \"alertname\"", testGroupName)
+		}
+		if !*disableAlertGroupLabel && at.GroupName == "" {
+			return fmt.Errorf("\n%s    missing required filed \"groupname\" when flag \"disableAlertGroupLabel\" is false", testGroupName)
+		}
+		if at.EvalTime == nil {
+			return fmt.Errorf("\n%s    missing required filed \"eval_time\"", testGroupName)
+		}
+	}
+	for _, et := range group.MetricsqlExprTests {
+		if et.Expr == "" {
+			return fmt.Errorf("\n%s    missing required filed \"expr\"", testGroupName)
+		}
+		if et.EvalTime == nil {
+			return fmt.Errorf("\n%s    missing required filed \"eval_time\"", testGroupName)
+		}
+	}
+	return nil
+}
+
+func processFlags() {
+	flag.Parse()
+	for _, fv := range []struct {
+		flag  string
+		value string
+	}{
+		{flag: "storageDataPath", value: storagePath},
+		{flag: "loggerLevel", value: testLogLevel},
+		{flag: "search.disableCache", value: "true"},
+		// set storage retention time to 100 years, allow to store series from 1970-01-01T00:00:00.
+		{flag: "retentionPeriod", value: "100y"},
+		{flag: "datasource.url", value: testDataSourcePath},
+		{flag: "remoteWrite.url", value: testRemoteWritePath},
+	} {
+		// panics if flag doesn't exist
+		if err := flag.Lookup(fv.flag).Value.Set(fv.value); err != nil {
+			logger.Fatalf("unable to set %q with value %q, err: %v", fv.flag, fv.value, err)
+		}
+	}
+}
+
+func setUp() {
+	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
+	go httpserver.Serve(*httpListenAddr, false, func(w http.ResponseWriter, r *http.Request) bool {
+		switch r.URL.Path {
+		case "/prometheus/api/v1/query":
+			if err := prometheus.QueryHandler(nil, time.Now(), w, r); err != nil {
+				httpserver.Errorf(w, r, "%s", err)
+			}
+			return true
+		case "/prometheus/api/v1/write", "/api/v1/write":
+			if err := promremotewrite.InsertHandler(r); err != nil {
+				httpserver.Errorf(w, r, "%s", err)
+			}
+			return true
+		default:
+		}
+		return false
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	readyCheckFunc := func() bool {
+		resp, err := http.Get(testHealthHTTPPath)
+		if err != nil {
+			return false
+		}
+		_ = resp.Body.Close()
+		return resp.StatusCode == 200
+	}
+checkCheck:
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Fatalf("http server can't be ready in 30s")
+		default:
+			if readyCheckFunc() {
+				break checkCheck
+			}
+			time.Sleep(3 * time.Second)
+		}
+	}
+}
+
+func tearDown() {
+	if err := httpserver.Stop(*httpListenAddr); err != nil {
+		logger.Errorf("cannot stop the webservice: %s", err)
+	}
+	vmstorage.Stop()
+	metrics.UnregisterAllMetrics()
+	fs.MustRemoveAll(storagePath)
+}
+
+// resolveAndGlobFilepaths joins all relative paths in a configuration
+// with a given base directory and replaces all globs with matching files.
+func resolveAndGlobFilepaths(baseDir string, utf *unitTestFile) error {
+	for i, rf := range utf.RuleFiles {
+		if rf != "" && !filepath.IsAbs(rf) {
+			utf.RuleFiles[i] = filepath.Join(baseDir, rf)
+		}
+	}
+
+	var globbedFiles []string
+	for _, rf := range utf.RuleFiles {
+		m, err := filepath.Glob(rf)
+		if err != nil {
+			return err
+		}
+		if len(m) == 0 {
+			fmt.Fprintln(os.Stderr, "  WARNING: no file match pattern", rf)
+		}
+		globbedFiles = append(globbedFiles, m...)
+	}
+	utf.RuleFiles = globbedFiles
+	return nil
+}
+
+func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]int, testGroups []vmalertconfig.Group) (checkErrs []error) {
+	// set up vmstorage and http server for ingest and read queries
+	setUp()
+	// tear down vmstorage and clean the data dir
+	defer tearDown()
+
+	err := unittest.WriteInputSeries(tg.InputSeries, tg.Interval, testStartTime, testPromWriteHTTPPath)
+	if err != nil {
+		return []error{err}
+	}
+
+	q, err := datasource.Init(nil)
+	if err != nil {
+		return []error{fmt.Errorf("failed to init datasource: %v", err)}
+	}
+	rw, err := remotewrite.NewDebugClient()
+	if err != nil {
+		return []error{fmt.Errorf("failed to init wr: %v", err)}
+	}
+
+	alertEvalTimesMap := map[time.Duration]struct{}{}
+	alertExpResultMap := map[time.Duration]map[string]map[string][]unittest.ExpAlert{}
+	for _, at := range tg.AlertRuleTests {
+		et := at.EvalTime.Duration()
+		alertEvalTimesMap[et] = struct{}{}
+		if _, ok := alertExpResultMap[et]; !ok {
+			alertExpResultMap[et] = make(map[string]map[string][]unittest.ExpAlert)
+		}
+		if _, ok := alertExpResultMap[et][at.GroupName]; !ok {
+			alertExpResultMap[et][at.GroupName] = make(map[string][]unittest.ExpAlert)
+		}
+		alertExpResultMap[et][at.GroupName][at.Alertname] = at.ExpAlerts
+	}
+	alertEvalTimes := make([]time.Duration, 0, len(alertEvalTimesMap))
+	for k := range alertEvalTimesMap {
+		alertEvalTimes = append(alertEvalTimes, k)
+	}
+	sort.Slice(alertEvalTimes, func(i, j int) bool {
+		return alertEvalTimes[i] < alertEvalTimes[j]
+	})
+
+	// sort group eval order according to the given "group_eval_order".
+	sort.Slice(testGroups, func(i, j int) bool {
+		return groupOrderMap[testGroups[i].Name] < groupOrderMap[testGroups[j].Name]
+	})
+
+	// create groups with given rule
+	var groups []*Group
+	for _, group := range testGroups {
+		ng := newGroup(group, q, *evaluationInterval, tg.ExternalLabels)
+		groups = append(groups, ng)
+	}
+
+	e := &executor{
+		rw:                       rw,
+		notifiers:                func() []notifier.Notifier { return nil },
+		previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
+	}
+
+	evalIndex := 0
+	maxEvalTime := testStartTime.Add(tg.maxEvalTime())
+	for ts := testStartTime; ts.Before(maxEvalTime) || ts.Equal(maxEvalTime); ts = ts.Add(evalInterval) {
+		for _, g := range groups {
+			resolveDuration := getResolveDuration(g.Interval, *resendDelay, *maxResolveDuration)
+			errs := e.execConcurrently(context.Background(), g.Rules, ts, g.Concurrency, resolveDuration, g.Limit)
+			for err := range errs {
+				if err != nil {
+					checkErrs = append(checkErrs, fmt.Errorf("\nfailed to exec group: %q, time: %s, err: %w", g.Name,
+						ts, err))
+				}
+			}
+			// flush series after each group evaluation
+			vmstorage.Storage.DebugFlush()
+		}
+
+		// check alert_rule_test case at every eval time
+		for evalIndex < len(alertEvalTimes) {
+			if ts.Sub(testStartTime) > alertEvalTimes[evalIndex] ||
+				alertEvalTimes[evalIndex] >= ts.Add(evalInterval).Sub(testStartTime) {
+				break
+			}
+			gotAlertsMap := map[string]map[string]unittest.LabelsAndAnnotations{}
+			for _, g := range groups {
+				if *disableAlertGroupLabel {
+					g.Name = ""
+				}
+				if _, ok := alertExpResultMap[time.Duration(ts.UnixNano())][g.Name]; !ok {
+					continue
+				}
+				if _, ok := gotAlertsMap[g.Name]; !ok {
+					gotAlertsMap[g.Name] = make(map[string]unittest.LabelsAndAnnotations)
+				}
+				for _, rule := range g.Rules {
+					ar, isAlertRule := rule.(*AlertingRule)
+					if !isAlertRule {
+						continue
+					}
+					if _, ok := alertExpResultMap[time.Duration(ts.UnixNano())][g.Name][ar.Name]; ok {
+						for _, got := range ar.alerts {
+							if got.State != notifier.StateFiring {
+								continue
+							}
+							laa := unittest.LabelAndAnnotation{
+								Labels:      datasource.ConvertToLabels(got.Labels),
+								Annotations: datasource.ConvertToLabels(got.Annotations),
+							}
+							gotAlertsMap[g.Name][ar.Name] = append(gotAlertsMap[g.Name][ar.Name], laa)
+						}
+					}
+
+				}
+			}
+			for groupname, gres := range alertExpResultMap[alertEvalTimes[evalIndex]] {
+				for alertname, res := range gres {
+					var expAlerts unittest.LabelsAndAnnotations
+					for _, expAlert := range res {
+						if expAlert.ExpLabels == nil {
+							expAlert.ExpLabels = make(map[string]string)
+						}
+						// alertGroupNameLabel is added as additional labels when `disableAlertGroupLabel` is false
+						if !*disableAlertGroupLabel {
+							expAlert.ExpLabels[alertGroupNameLabel] = groupname
+						}
+						// alertNameLabel is added as additional labels in vmalert.
+						expAlert.ExpLabels[alertNameLabel] = alertname
+						expAlerts = append(expAlerts, unittest.LabelAndAnnotation{
+							Labels:      datasource.ConvertToLabels(expAlert.ExpLabels),
+							Annotations: datasource.ConvertToLabels(expAlert.ExpAnnotations),
+						})
+					}
+					sort.Sort(expAlerts)
+
+					gotAlerts := gotAlertsMap[groupname][alertname]
+					sort.Sort(gotAlerts)
+					if !reflect.DeepEqual(expAlerts, gotAlerts) {
+						var testGroupName string
+						if tg.TestGroupName != "" {
+							testGroupName = fmt.Sprintf("testGroupName: %s,\n", tg.TestGroupName)
+						}
+						expString := unittest.IndentLines(expAlerts.String(), "            ")
+						gotString := unittest.IndentLines(gotAlerts.String(), "            ")
+						checkErrs = append(checkErrs, fmt.Errorf("\n%s    groupname: %s, alertname: %s, time: %s, \n        exp:%v, \n        got:%v ",
+							testGroupName, groupname, alertname, alertEvalTimes[evalIndex].String(), expString, gotString))
+					}
+				}
+			}
+			evalIndex++
+		}
+
+	}
+
+	checkErrs = append(checkErrs, unittest.CheckMetricsqlCase(tg.MetricsqlExprTests, q)...)
+	return checkErrs
+}
+
+// unitTestFile holds the contents of a single unit test file
+type unitTestFile struct {
+	RuleFiles          []string            `yaml:"rule_files"`
+	EvaluationInterval *promutils.Duration `yaml:"evaluation_interval"`
+	GroupEvalOrder     []string            `yaml:"group_eval_order"`
+	Tests              []testGroup         `yaml:"tests"`
+}
+
+// testGroup is a group of input series and test cases associated with it
+type testGroup struct {
+	Interval           *promutils.Duration          `yaml:"interval"`
+	InputSeries        []unittest.Series            `yaml:"input_series"`
+	AlertRuleTests     []unittest.AlertTestCase     `yaml:"alert_rule_test"`
+	MetricsqlExprTests []unittest.MetricsqlTestCase `yaml:"metricsql_expr_test"`
+	ExternalLabels     map[string]string            `yaml:"external_labels"`
+	TestGroupName      string                       `yaml:"name"`
+}
+
+// maxEvalTime returns the max eval time among all alert_rule_test and metricsql_expr_test
+func (tg *testGroup) maxEvalTime() time.Duration {
+	var maxd time.Duration
+	for _, alert := range tg.AlertRuleTests {
+		if alert.EvalTime.Duration() > maxd {
+			maxd = alert.EvalTime.Duration()
+		}
+	}
+	for _, met := range tg.MetricsqlExprTests {
+		if met.EvalTime.Duration() > maxd {
+			maxd = met.EvalTime.Duration()
+		}
+	}
+	return maxd
+}
diff --git a/app/vmalert/unittest/alerting.go b/app/vmalert/unittest/alerting.go
new file mode 100644
index 000000000..23f0845db
--- /dev/null
+++ b/app/vmalert/unittest/alerting.go
@@ -0,0 +1,19 @@
+package unittest
+
+import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+)
+
+// AlertTestCase holds alert_rule_test cases defined in test file
+type AlertTestCase struct {
+	EvalTime  *promutils.Duration `yaml:"eval_time"`
+	GroupName string              `yaml:"groupname"`
+	Alertname string              `yaml:"alertname"`
+	ExpAlerts []ExpAlert          `yaml:"exp_alerts"`
+}
+
+// ExpAlert holds exp_alerts defined in test file
+type ExpAlert struct {
+	ExpLabels      map[string]string `yaml:"exp_labels"`
+	ExpAnnotations map[string]string `yaml:"exp_annotations"`
+}
diff --git a/app/vmalert/unittest/input.go b/app/vmalert/unittest/input.go
new file mode 100644
index 000000000..6ba19bab3
--- /dev/null
+++ b/app/vmalert/unittest/input.go
@@ -0,0 +1,177 @@
+package unittest
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	testutil "github.com/VictoriaMetrics/VictoriaMetrics/app/victoria-metrics/test"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/metricsql"
+)
+
+// Series holds input_series defined in the test file
+type Series struct {
+	Series string `yaml:"series"`
+	Values string `yaml:"values"`
+}
+
+// sequenceValue is an omittable value in a sequence of time series values.
+type sequenceValue struct {
+	Value   float64
+	Omitted bool
+}
+
+func httpWrite(address string, r io.Reader) {
+	resp, err := http.Post(address, "", r)
+	if err != nil {
+		logger.Fatalf("failed to send to storage: %v", err)
+	}
+	resp.Body.Close()
+}
+
+// WriteInputSeries send input series to vmstorage and flush them
+func WriteInputSeries(input []Series, interval *promutils.Duration, startStamp time.Time, dst string) error {
+	r := testutil.WriteRequest{}
+	for _, data := range input {
+		expr, err := metricsql.Parse(data.Series)
+		if err != nil {
+			return fmt.Errorf("failed to parse series %s: %v", data.Series, err)
+		}
+		promvals, err := parseInputValue(data.Values, true)
+		if err != nil {
+			return fmt.Errorf("failed to parse input series value %s: %v", data.Values, err)
+		}
+		metricExpr, ok := expr.(*metricsql.MetricExpr)
+		if !ok {
+			return fmt.Errorf("failed to parse series %s to metric expr: %v", data.Series, err)
+		}
+		samples := make([]testutil.Sample, 0, len(promvals))
+		ts := startStamp
+		for _, v := range promvals {
+			if !v.Omitted {
+				samples = append(samples, testutil.Sample{
+					Timestamp: ts.UnixMilli(),
+					Value:     v.Value,
+				})
+			}
+			ts = ts.Add(interval.Duration())
+		}
+		var ls []testutil.Label
+		for _, filter := range metricExpr.LabelFilterss[0] {
+			ls = append(ls, testutil.Label{Name: filter.Label, Value: filter.Value})
+		}
+		r.Timeseries = append(r.Timeseries, testutil.TimeSeries{Labels: ls, Samples: samples})
+	}
+
+	data, err := testutil.Compress(r)
+	if err != nil {
+		return fmt.Errorf("failed to compress data: %v", err)
+	}
+	// write input series to vm
+	httpWrite(dst, bytes.NewBuffer(data))
+	vmstorage.Storage.DebugFlush()
+	return nil
+}
+
+// parseInputValue support input like "1", "1+1x1 _ -4 3+20x1", see more examples in test.
+func parseInputValue(input string, origin bool) ([]sequenceValue, error) {
+	var res []sequenceValue
+	items := strings.Split(input, " ")
+	reg2 := regexp.MustCompile(`\D?\d*\D?`)
+	for _, item := range items {
+		vals := reg2.FindAllString(item, -1)
+		switch len(vals) {
+		case 1:
+			if vals[0] == "_" {
+				res = append(res, sequenceValue{Omitted: true})
+				continue
+			}
+			v, err := strconv.ParseFloat(vals[0], 64)
+			if err != nil {
+				return nil, err
+			}
+			res = append(res, sequenceValue{Value: v})
+			continue
+		case 2:
+			p1 := vals[0][:len(vals[0])-1]
+			v2, err := strconv.ParseInt(vals[1], 10, 64)
+			if err != nil {
+				return nil, err
+			}
+			option := vals[0][len(vals[0])-1]
+			switch option {
+			case '+':
+				v1, err := strconv.ParseFloat(p1, 64)
+				if err != nil {
+					return nil, err
+				}
+				res = append(res, sequenceValue{Value: v1 + float64(v2)})
+			case 'x':
+				for i := int64(0); i <= v2; i++ {
+					if p1 == "_" {
+						if i == 0 {
+							i = 1
+						}
+						res = append(res, sequenceValue{Omitted: true})
+						continue
+					}
+					v1, err := strconv.ParseFloat(p1, 64)
+					if err != nil {
+						return nil, err
+					}
+					if !origin || v1 == 0 {
+						res = append(res, sequenceValue{Value: v1 * float64(i)})
+						continue
+					}
+					newVal := fmt.Sprintf("%s+0x%s", p1, vals[1])
+					newRes, err := parseInputValue(newVal, false)
+					if err != nil {
+						return nil, err
+					}
+					res = append(res, newRes...)
+					break
+				}
+
+			default:
+				return nil, fmt.Errorf("got invalid operation %b", option)
+			}
+		case 3:
+			r1, err := parseInputValue(fmt.Sprintf("%s%s", vals[1], vals[2]), false)
+			if err != nil {
+				return nil, err
+			}
+			p1 := vals[0][:len(vals[0])-1]
+			v1, err := strconv.ParseFloat(p1, 64)
+			if err != nil {
+				return nil, err
+			}
+			option := vals[0][len(vals[0])-1]
+			var isAdd bool
+			if option == '+' {
+				isAdd = true
+			}
+			for _, r := range r1 {
+				if isAdd {
+					res = append(res, sequenceValue{
+						Value: r.Value + v1,
+					})
+				} else {
+					res = append(res, sequenceValue{
+						Value: v1 - r.Value,
+					})
+				}
+			}
+		default:
+			return nil, fmt.Errorf("unsupported input %s", input)
+		}
+	}
+	return res, nil
+}
diff --git a/app/vmalert/unittest/input_test.go b/app/vmalert/unittest/input_test.go
new file mode 100644
index 000000000..cf8bb7942
--- /dev/null
+++ b/app/vmalert/unittest/input_test.go
@@ -0,0 +1,70 @@
+package unittest
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestParseInputValue(t *testing.T) {
+	testCases := []struct {
+		input  string
+		exp    []sequenceValue
+		failed bool
+	}{
+		{
+			"",
+			nil,
+			true,
+		},
+		{
+			"testfailed",
+			nil,
+			true,
+		},
+		{
+			"-4",
+			[]sequenceValue{{Value: -4}},
+			false,
+		},
+		{
+			"_",
+			[]sequenceValue{{Omitted: true}},
+			false,
+		},
+		{
+			"-4x1",
+			[]sequenceValue{{Value: -4}, {Value: -4}},
+			false,
+		},
+		{
+			"_x1",
+			[]sequenceValue{{Omitted: true}},
+			false,
+		},
+		{
+			"1+1x4",
+			[]sequenceValue{{Value: 1}, {Value: 2}, {Value: 3}, {Value: 4}, {Value: 5}},
+			false,
+		},
+		{
+			"2-1x4",
+			[]sequenceValue{{Value: 2}, {Value: 1}, {Value: 0}, {Value: -1}, {Value: -2}},
+			false,
+		},
+		{
+			"1+1x1 _ -4 3+20x1",
+			[]sequenceValue{{Value: 1}, {Value: 2}, {Omitted: true}, {Value: -4}, {Value: 3}, {Value: 23}},
+			false,
+		},
+	}
+
+	for _, tc := range testCases {
+		output, err := parseInputValue(tc.input, true)
+		if err != nil != tc.failed {
+			t.Fatalf("failed to parse %s, expect %t, got %t", tc.input, tc.failed, err != nil)
+		}
+		if !reflect.DeepEqual(tc.exp, output) {
+			t.Fatalf("expect %v, got %v", tc.exp, output)
+		}
+	}
+}
diff --git a/app/vmalert/unittest/recording.go b/app/vmalert/unittest/recording.go
new file mode 100644
index 000000000..f4789ec5b
--- /dev/null
+++ b/app/vmalert/unittest/recording.go
@@ -0,0 +1,92 @@
+package unittest
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"reflect"
+	"sort"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/metricsql"
+)
+
+// MetricsqlTestCase holds metricsql_expr_test cases defined in test file
+type MetricsqlTestCase struct {
+	Expr       string              `yaml:"expr"`
+	EvalTime   *promutils.Duration `yaml:"eval_time"`
+	ExpSamples []expSample         `yaml:"exp_samples"`
+}
+
+type expSample struct {
+	Labels string  `yaml:"labels"`
+	Value  float64 `yaml:"value"`
+}
+
+// CheckMetricsqlCase will check metricsql_expr_test cases
+func CheckMetricsqlCase(cases []MetricsqlTestCase, q datasource.QuerierBuilder) (checkErrs []error) {
+	queries := q.BuildWithParams(datasource.QuerierParams{QueryParams: url.Values{"nocache": {"1"}, "latency_offset": {"1ms"}}, DataSourceType: "prometheus"})
+Outer:
+	for _, mt := range cases {
+		result, _, err := queries.Query(context.Background(), mt.Expr, mt.EvalTime.ParseTime())
+		if err != nil {
+			checkErrs = append(checkErrs, fmt.Errorf("    expr: %q, time: %s, err: %w", mt.Expr,
+				mt.EvalTime.Duration().String(), err))
+			continue
+		}
+		var gotSamples []parsedSample
+		for _, s := range result.Data {
+			sort.Slice(s.Labels, func(i, j int) bool {
+				return s.Labels[i].Name < s.Labels[j].Name
+			})
+			gotSamples = append(gotSamples, parsedSample{
+				Labels: s.Labels,
+				Value:  s.Values[0],
+			})
+		}
+		var expSamples []parsedSample
+		for _, s := range mt.ExpSamples {
+			expLb := datasource.Labels{}
+			if s.Labels != "" {
+				metricsqlExpr, err := metricsql.Parse(s.Labels)
+				if err != nil {
+					checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s, err: %v", mt.Expr,
+						mt.EvalTime.Duration().String(), fmt.Errorf("failed to parse labels %q: %w", s.Labels, err)))
+					continue Outer
+				}
+				metricsqlMetricExpr, ok := metricsqlExpr.(*metricsql.MetricExpr)
+				if !ok {
+					checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s, err: %v", mt.Expr,
+						mt.EvalTime.Duration().String(), fmt.Errorf("got unsupported metricsql type")))
+					continue Outer
+				}
+				for _, l := range metricsqlMetricExpr.LabelFilterss[0] {
+					expLb = append(expLb, datasource.Label{
+						Name:  l.Label,
+						Value: l.Value,
+					})
+				}
+			}
+			sort.Slice(expLb, func(i, j int) bool {
+				return expLb[i].Name < expLb[j].Name
+			})
+			expSamples = append(expSamples, parsedSample{
+				Labels: expLb,
+				Value:  s.Value,
+			})
+		}
+		sort.Slice(expSamples, func(i, j int) bool {
+			return datasource.LabelCompare(expSamples[i].Labels, expSamples[j].Labels) <= 0
+		})
+		sort.Slice(gotSamples, func(i, j int) bool {
+			return datasource.LabelCompare(gotSamples[i].Labels, gotSamples[j].Labels) <= 0
+		})
+		if !reflect.DeepEqual(expSamples, gotSamples) {
+			checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s,\n        exp: %v\n        got: %v", mt.Expr,
+				mt.EvalTime.Duration().String(), parsedSamplesString(expSamples), parsedSamplesString(gotSamples)))
+		}
+
+	}
+	return
+}
diff --git a/app/vmalert/unittest/testdata/disable-group-label.yaml b/app/vmalert/unittest/testdata/disable-group-label.yaml
new file mode 100644
index 000000000..6060710b6
--- /dev/null
+++ b/app/vmalert/unittest/testdata/disable-group-label.yaml
@@ -0,0 +1,43 @@
+rule_files:
+  - rules.yaml
+
+evaluation_interval: 1m
+
+tests:
+  - interval: 1m
+    input_series:
+      - series: 'up{job="vmagent2", instance="localhost:9090"}'
+        values: "0+0x1440"
+
+    metricsql_expr_test:
+      - expr: suquery_interval_test
+        eval_time: 4m
+        exp_samples:
+          - labels: '{__name__="suquery_interval_test",datacenter="dc-123", instance="localhost:9090", job="vmagent2"}'
+            value: 1
+
+    alert_rule_test:
+      - eval_time: 2h
+        alertname: InstanceDown
+        exp_alerts:
+          - exp_labels:
+              job: vmagent2
+              severity: page
+              instance: localhost:9090
+              datacenter: dc-123
+            exp_annotations:
+              summary: "Instance localhost:9090 down"
+              description: "localhost:9090 of job vmagent2 has been down for more than 5 minutes."
+
+      - eval_time: 0
+        alertname: AlwaysFiring
+        exp_alerts:
+          - exp_labels:
+              datacenter: dc-123
+
+      - eval_time: 0
+        alertname: InstanceDown
+        exp_alerts: []
+
+    external_labels:
+      datacenter: dc-123
diff --git a/app/vmalert/unittest/testdata/failed-test.yaml b/app/vmalert/unittest/testdata/failed-test.yaml
new file mode 100644
index 000000000..1e55296f2
--- /dev/null
+++ b/app/vmalert/unittest/testdata/failed-test.yaml
@@ -0,0 +1,41 @@
+rule_files:
+  - rules.yaml
+
+tests:
+  - interval: 1m
+    name: "Failing test"
+    input_series:
+      - series: test
+        values: "0"
+
+    metricsql_expr_test:
+      - expr: test
+        eval_time: 0m
+        exp_samples:
+          - value: 0
+            labels: test
+
+    # will failed cause there is no "Test" group and rule defined
+    alert_rule_test:
+      - eval_time: 0m
+        groupname: Test
+        alertname: Test
+        exp_alerts:
+          - exp_labels: {}
+
+  - interval: 1m
+    name: Failing alert test
+    input_series:
+      - series: 'up{job="test"}'
+        values: 0x10
+
+    alert_rule_test:
+      # will failed cause rule is firing
+      - eval_time: 5m
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts: []
+      # will failed cause missing groupname
+      - eval_time: 5m
+        alertname: AlwaysFiring
+        exp_alerts: []
diff --git a/app/vmalert/unittest/testdata/long-period.yaml b/app/vmalert/unittest/testdata/long-period.yaml
new file mode 100644
index 000000000..052470b69
--- /dev/null
+++ b/app/vmalert/unittest/testdata/long-period.yaml
@@ -0,0 +1,30 @@
+# can be executed successfully but will take more than 1 minute
+# not included in unit test now
+evaluation_interval: 100d
+
+rule_files:
+  - rules.yaml
+
+tests:
+  - interval: 1d
+    input_series:
+      - series: test
+        # Max time in time.Duration is 106751d from 1970 (2^63/10^9), i.e. 2262.
+        # But VictoriaMetrics supports maxTimestamp value +2 days from now. see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/827.
+        # We input series to 2024-01-01T00:00:00 here.
+        values: "0+1x19723"
+
+    metricsql_expr_test:
+      - expr: timestamp(test)
+        eval_time: 0m
+        exp_samples:
+          - value: 0
+      - expr: test
+        eval_time: 100d
+        exp_samples:
+          - labels: test
+            value: 100
+      - expr: timestamp(test)
+        eval_time: 19000d
+        exp_samples:
+          - value: 1641600000 # 19000d -> seconds.
diff --git a/app/vmalert/unittest/testdata/rules.yaml b/app/vmalert/unittest/testdata/rules.yaml
new file mode 100644
index 000000000..f7455c6d8
--- /dev/null
+++ b/app/vmalert/unittest/testdata/rules.yaml
@@ -0,0 +1,39 @@
+groups:
+  - name: group1
+    rules:
+      - alert: InstanceDown
+        expr: up == 0
+        for: 5m
+        labels:
+          severity: page
+        annotations:
+          summary: "Instance {{ $labels.instance }} down"
+          description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 5 minutes."
+      - alert: AlwaysFiring
+        expr: 1
+      - alert: SameAlertNameWithDifferentGroup
+        expr: absent(test)
+        for: 1m
+
+  - name: group2
+    rules:
+      - record: t1
+        expr: test
+      - record: job:test:count_over_time1m
+        expr: sum without(instance) (count_over_time(test[1m]))
+      - record: suquery_interval_test
+        expr: count_over_time(up[5m:])
+
+      - alert: SameAlertNameWithDifferentGroup
+        expr: absent(test)
+        for: 5m
+
+  - name: group3
+    rules:
+      - record: t2
+        expr: t1
+
+  - name: group4
+    rules:
+      - record: t3
+        expr: t1
diff --git a/app/vmalert/unittest/testdata/test1.yaml b/app/vmalert/unittest/testdata/test1.yaml
new file mode 100644
index 000000000..96f3a30ab
--- /dev/null
+++ b/app/vmalert/unittest/testdata/test1.yaml
@@ -0,0 +1,89 @@
+rule_files:
+  - rules.yaml
+
+evaluation_interval: 1m
+group_eval_order: ["group4", "group2", "group3"]
+
+tests:
+  - interval: 1m
+    name: "basic test"
+    input_series:
+      - series: "test"
+        values: "_x5 1x5"
+
+    alert_rule_test:
+      - eval_time: 1m
+        groupname: group1
+        alertname: SameAlertNameWithDifferentGroup
+        exp_alerts:
+          - {}
+      - eval_time: 1m
+        groupname: group2
+        alertname: SameAlertNameWithDifferentGroup
+        exp_alerts: []
+      - eval_time: 6m
+        groupname: group1
+        alertname: SameAlertNameWithDifferentGroup
+        exp_alerts: []
+
+  - interval: 1m
+    name: "basic test2"
+    input_series:
+      - series: 'up{job="vmagent1", instance="localhost:9090"}'
+        values: "0+0x1440"
+      - series: "test"
+        values: "0+1x1440"
+
+    metricsql_expr_test:
+      - expr: count(ALERTS) by (alertgroup, alertname, alertstate)
+        eval_time: 4m
+        exp_samples:
+          - labels: '{alertgroup="group1", alertname="AlwaysFiring", alertstate="firing"}'
+            value: 1
+          - labels: '{alertgroup="group1", alertname="InstanceDown", alertstate="pending"}'
+            value: 1
+      - expr: t1
+        eval_time: 4m
+        exp_samples:
+          - value: 4
+            labels: '{__name__="t1", datacenter="dc-123"}'
+      - expr: t2
+        eval_time: 4m
+        exp_samples:
+          - value: 4
+            labels: '{__name__="t2", datacenter="dc-123"}'
+      - expr: t3
+        eval_time: 4m
+        exp_samples:
+          # t3 is 3 instead of 4 cause it's rules3 is evaluated before rules1
+          - value: 3
+            labels: '{__name__="t3", datacenter="dc-123"}'
+
+    alert_rule_test:
+      - eval_time: 10m
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts:
+          - exp_labels:
+              job: vmagent1
+              severity: page
+              instance: localhost:9090
+              datacenter: dc-123
+            exp_annotations:
+              summary: "Instance localhost:9090 down"
+              description: "localhost:9090 of job vmagent1 has been down for more than 5 minutes."
+
+      - eval_time: 0
+        groupname: group1
+        alertname: AlwaysFiring
+        exp_alerts:
+          - exp_labels:
+              datacenter: dc-123
+
+      - eval_time: 0
+        groupname: alerts
+        alertname: InstanceDown
+        exp_alerts: []
+
+    external_labels:
+      datacenter: dc-123
diff --git a/app/vmalert/unittest/testdata/test2.yaml b/app/vmalert/unittest/testdata/test2.yaml
new file mode 100644
index 000000000..2d7825742
--- /dev/null
+++ b/app/vmalert/unittest/testdata/test2.yaml
@@ -0,0 +1,46 @@
+rule_files:
+  - rules.yaml
+
+evaluation_interval: 1m
+
+tests:
+  - interval: 1m
+    input_series:
+      - series: 'up{job="vmagent2", instance="localhost:9090"}'
+        values: "0+0x1440"
+
+    metricsql_expr_test:
+      - expr: suquery_interval_test
+        eval_time: 4m
+        exp_samples:
+          - labels: '{__name__="suquery_interval_test",datacenter="dc-123", instance="localhost:9090", job="vmagent2"}'
+            value: 1
+
+    alert_rule_test:
+      - eval_time: 2h
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts:
+          - exp_labels:
+              job: vmagent2
+              severity: page
+              instance: localhost:9090
+              datacenter: dc-123
+            exp_annotations:
+              summary: "Instance localhost:9090 down"
+              description: "localhost:9090 of job vmagent2 has been down for more than 5 minutes."
+
+      - eval_time: 0
+        groupname: group1
+        alertname: AlwaysFiring
+        exp_alerts:
+          - exp_labels:
+              datacenter: dc-123
+
+      - eval_time: 0
+        groupname: group1
+        alertname: InstanceDown
+        exp_alerts: []
+
+    external_labels:
+      datacenter: dc-123
diff --git a/app/vmalert/unittest/type.go b/app/vmalert/unittest/type.go
new file mode 100644
index 000000000..22e881a2c
--- /dev/null
+++ b/app/vmalert/unittest/type.go
@@ -0,0 +1,83 @@
+package unittest
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
+)
+
+// parsedSample is a sample with parsed Labels
+type parsedSample struct {
+	Labels datasource.Labels
+	Value  float64
+}
+
+func (ps *parsedSample) String() string {
+	return ps.Labels.String() + " " + strconv.FormatFloat(ps.Value, 'E', -1, 64)
+}
+
+func parsedSamplesString(pss []parsedSample) string {
+	if len(pss) == 0 {
+		return "nil"
+	}
+	s := pss[0].String()
+	for _, ps := range pss[1:] {
+		s += ", " + ps.String()
+	}
+	return s
+}
+
+// LabelAndAnnotation holds labels and annotations
+type LabelAndAnnotation struct {
+	Labels      datasource.Labels
+	Annotations datasource.Labels
+}
+
+func (la *LabelAndAnnotation) String() string {
+	return "Labels:" + la.Labels.String() + "\nAnnotations:" + la.Annotations.String()
+}
+
+// LabelsAndAnnotations is collection of LabelAndAnnotation
+type LabelsAndAnnotations []LabelAndAnnotation
+
+func (la LabelsAndAnnotations) Len() int { return len(la) }
+
+func (la LabelsAndAnnotations) Swap(i, j int) { la[i], la[j] = la[j], la[i] }
+func (la LabelsAndAnnotations) Less(i, j int) bool {
+	diff := datasource.LabelCompare(la[i].Labels, la[j].Labels)
+	if diff != 0 {
+		return diff < 0
+	}
+	return datasource.LabelCompare(la[i].Annotations, la[j].Annotations) < 0
+}
+
+func (la LabelsAndAnnotations) String() string {
+	if len(la) == 0 {
+		return "[]"
+	}
+	s := "[\n0:" + IndentLines("\n"+la[0].String(), "  ")
+	for i, l := range la[1:] {
+		s += ",\n" + fmt.Sprintf("%d", i+1) + ":" + IndentLines("\n"+l.String(), "  ")
+	}
+	s += "\n]"
+
+	return s
+}
+
+// IndentLines prefixes each line in the supplied string with the given "indent" string.
+func IndentLines(lines, indent string) string {
+	sb := strings.Builder{}
+	n := strings.Split(lines, "\n")
+	for i, l := range n {
+		if i > 0 {
+			sb.WriteString(indent)
+		}
+		sb.WriteString(l)
+		if i != len(n)-1 {
+			sb.WriteRune('\n')
+		}
+	}
+	return sb.String()
+}