added compressor cleanup

.github/workflows/main.yml

@@ -88,35 +88,6 @@ jobs:
         run: make ${{ matrix.scenario}}
 
       - name: Publish coverage
-        uses: codecov/codecov-action@v5
+        uses: codecov/codecov-action@v4
         with:
           file: ./coverage.txt
-
-  integration-test:
-    name: integration-test
-    needs: [lint, test]
-    runs-on: ubuntu-latest
-
-    steps:
-      - name: Code checkout
-        uses: actions/checkout@v4
-
-      - name: Setup Go
-        id: go
-        uses: actions/setup-go@v5
-        with:
-          cache: false
-          go-version: stable
-
-      - name: Cache Go artifacts
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cache/go-build
-            ~/go/bin
-            ~/go/pkg/mod
-          key: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-
-
-      - name: Run integration tests
-        run: make integration-test

app/victoria-metrics/main.go

@@ -14,7 +14,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
@@ -40,20 +39,9 @@ var (
 		"The saved data survives unclean shutdowns such as OOM crash, hardware reset, SIGKILL, etc. "+
 		"Bigger intervals may help increase the lifetime of flash storage with limited write cycles (e.g. Raspberry PI). "+
 		"Smaller intervals increase disk IO load. Minimum supported value is 1s")
-	maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
-		"By default there are no limits on samples ingestion rate.")
 )
 
 func main() {
-	// VictoriaMetrics is optimized for reduced memory allocations,
-	// so it can run with the reduced GOGC in order to reduce the used memory,
-	// while keeping CPU usage spent in GC at low levels.
-	//
-	// Some workloads may need increased GOGC values. Then such values can be set via GOGC environment variable.
-	// It is recommended increasing GOGC if go_memstats_gc_cpu_fraction metric exposed at /metrics page
-	// exceeds 0.05 for extended periods of time.
-	cgroup.SetGOGC(30)
-
 	// Write flags and help message to stdout, since it is easier to grep or pipe.
 	flag.CommandLine.SetOutput(os.Stdout)
 	flag.Usage = usage
@@ -88,7 +76,6 @@ func main() {
 	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
 	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
 	vmselect.Init()
-	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)
 	vminsert.Init()
 
 	startSelfScraper()
@@ -110,7 +97,6 @@ func main() {
 	}
 	logger.Infof("successfully shut down the webservice in %.3f seconds", time.Since(startTime).Seconds())
 	vminsert.Stop()
-	vminsertcommon.StopIngestionRateLimiter()
 
 	vmstorage.Stop()
 	vmselect.Stop()

app/victoria-metrics/self_scraper.go

@@ -10,10 +10,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 )
 
 var (
@@ -49,7 +48,7 @@ func selfScraper(scrapeInterval time.Duration) {
 	var bb bytesutil.ByteBuffer
 	var rows prometheus.Rows
 	var mrs []storage.MetricRow
-	var labels []prompbmarshal.Label
+	var labels []prompb.Label
 	t := time.NewTicker(scrapeInterval)
 	f := func(currentTime time.Time, sendStaleMarkers bool) {
 		currentTimestamp := currentTime.UnixNano() / 1e6
@@ -69,10 +68,6 @@ func selfScraper(scrapeInterval time.Duration) {
 				t := &r.Tags[j]
 				labels = addLabel(labels, t.Key, t.Value)
 			}
-			if timeserieslimits.IsExceeding(labels) {
-				// Skip metric with exceeding labels.
-				continue
-			}
 			if len(mrs) < cap(mrs) {
 				mrs = mrs[:len(mrs)+1]
 			} else {
@@ -104,11 +99,11 @@ func selfScraper(scrapeInterval time.Duration) {
 	}
 }
 
-func addLabel(dst []prompbmarshal.Label, key, value string) []prompbmarshal.Label {
+func addLabel(dst []prompb.Label, key, value string) []prompb.Label {
 	if len(dst) < cap(dst) {
 		dst = dst[:len(dst)+1]
 	} else {
-		dst = append(dst, prompbmarshal.Label{})
+		dst = append(dst, prompb.Label{})
 	}
 	lb := &dst[len(dst)-1]
 	lb.Name = key

app/vlinsert/datadog/datadog.go

@@ -14,7 +14,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
@@ -22,11 +21,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 )
 
-var (
-	datadogStreamFields = flagutil.NewArrayString("datadog.streamFields", "Datadog tags to be used as stream fields.")
-	datadogIgnoreFields = flagutil.NewArrayString("datadog.ignoreFields", "Datadog tags to ignore.")
-)
-
 var parserPool fastjson.ParserPool
 
 // RequestHandler processes Datadog insert requests
@@ -85,21 +79,17 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 
-	if len(cp.StreamFields) == 0 {
-		cp.StreamFields = *datadogStreamFields
-	}
-	if len(cp.IgnoreFields) == 0 {
-		cp.IgnoreFields = *datadogIgnoreFields
-	}
-
 	if err := vlstorage.CanWriteData(); err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return true
 	}
 
-	lmp := cp.NewLogMessageProcessor("datadog")
-	err = readLogsRequest(ts, data, lmp)
+	lmp := cp.NewLogMessageProcessor()
+	n, err := readLogsRequest(ts, data, lmp.AddRow)
 	lmp.MustClose()
+	if n > 0 {
+		rowsIngestedTotal.Add(n)
+	}
 	if err != nil {
 		logger.Warnf("cannot decode log message in /api/v2/logs request: %s, stream fields: %s", err, cp.StreamFields)
 		return true
@@ -115,118 +105,47 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 
 var (
 	v2LogsRequestsTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/datadog/api/v2/logs"}`)
+	rowsIngestedTotal     = metrics.NewCounter(`vl_rows_ingested_total{type="datadog"}`)
 	v2LogsRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/datadog/api/v2/logs"}`)
 )
 
-// datadog message field has two formats:
-//   - regular log message with string text
-//   - nested json format for serverless plugins
-//     which has folowing format:
-//     {"message": {"message": "text","lamdba": {"arn": "string","requestID": "string"}, "timestamp": int64} }
-//
-// See https://github.com/DataDog/datadog-lambda-extension/blob/28b90c7e4e985b72d60b5f5a5147c69c7ac693c4/bottlecap/src/logs/lambda/mod.rs#L24
-func appendMsgFields(fields []logstorage.Field, v *fastjson.Value) ([]logstorage.Field, error) {
-	switch v.Type() {
-	case fastjson.TypeString:
-		val := v.GetStringBytes()
-		fields = append(fields, logstorage.Field{
-			Name:  "_msg",
-			Value: bytesutil.ToUnsafeString(val),
-		})
-	case fastjson.TypeObject:
-		var firstErr error
-		v.GetObject().Visit(func(k []byte, v *fastjson.Value) {
-			if firstErr != nil {
-				return
-			}
-			switch bytesutil.ToUnsafeString(k) {
-			case "message":
-				val := v.GetStringBytes()
-				fields = append(fields, logstorage.Field{
-					Name:  "_msg",
-					Value: bytesutil.ToUnsafeString(val),
-				})
-			case "status":
-				val := v.GetStringBytes()
-				fields = append(fields, logstorage.Field{
-					Name:  "status",
-					Value: bytesutil.ToUnsafeString(val),
-				})
-			case "lamdba":
-				obj, err := v.Object()
-				if err != nil {
-					firstErr = err
-					firstErr = fmt.Errorf("unexpected lambda value type for %q:%q; want object", k, v)
-					return
-				}
-				obj.Visit(func(k []byte, v *fastjson.Value) {
-					if firstErr != nil {
-						return
-					}
-					val, err := v.StringBytes()
-					if err != nil {
-						firstErr = fmt.Errorf("unexpected lambda label value type for %q:%q; want string", k, v)
-						return
-					}
-					fields = append(fields, logstorage.Field{
-						Name:  bytesutil.ToUnsafeString(k),
-						Value: bytesutil.ToUnsafeString(val),
-					})
-				})
-
-			}
-		})
-	default:
-		return fields, fmt.Errorf("unsupported message type %q", v.Type().String())
-	}
-	return fields, nil
-}
-
 // readLogsRequest parses data according to DataDog logs format
 // https://docs.datadoghq.com/api/latest/logs/#send-logs
-func readLogsRequest(ts int64, data []byte, lmp insertutils.LogMessageProcessor) error {
+func readLogsRequest(ts int64, data []byte, processLogMessage func(int64, []logstorage.Field)) (int, error) {
 	p := parserPool.Get()
 	defer parserPool.Put(p)
 	v, err := p.ParseBytes(data)
 	if err != nil {
-		return fmt.Errorf("cannot parse JSON request body: %w", err)
+		return 0, fmt.Errorf("cannot parse JSON request body: %w", err)
 	}
 	records, err := v.Array()
 	if err != nil {
-		return fmt.Errorf("cannot extract array from parsed JSON: %w", err)
+		return 0, fmt.Errorf("cannot extract array from parsed JSON: %w", err)
 	}
 
 	var fields []logstorage.Field
-	for _, r := range records {
+	for m, r := range records {
 		o, err := r.Object()
 		if err != nil {
-			return fmt.Errorf("could not extract log record: %w", err)
+			return m + 1, fmt.Errorf("could not extract log record: %w", err)
 		}
 		o.Visit(func(k []byte, v *fastjson.Value) {
 			if err != nil {
 				return
 			}
-			switch bytesutil.ToUnsafeString(k) {
+			val, e := v.StringBytes()
+			if e != nil {
+				err = fmt.Errorf("unexpected label value type for %q:%q; want string", k, v)
+				return
+			}
+			switch string(k) {
 			case "message":
-				fields, err = appendMsgFields(fields, v)
-				if err != nil {
-					return
-				}
-			case "timestamp":
-				val, e := v.Int64()
-				if e != nil {
-					err = fmt.Errorf("failed to parse timestamp for %q:%q", k, v)
-				}
-				if val > 0 {
-					ts = val * 1e6
-				}
+				fields = append(fields, logstorage.Field{
+					Name:  "_msg",
+					Value: bytesutil.ToUnsafeString(val),
+				})
 			case "ddtags":
 				// https://docs.datadoghq.com/getting_started/tagging/
-				val, e := v.StringBytes()
-				if e != nil {
-					err = fmt.Errorf("unexpected label value type for %q:%q; want string", k, v)
-					return
-				}
 				var pair []byte
 				idx := 0
 				for idx >= 0 {
@@ -253,22 +172,14 @@ func readLogsRequest(ts int64, data []byte, lmp insertutils.LogMessageProcessor)
 					}
 				}
 			default:
-				val, e := v.StringBytes()
-				if e != nil {
-					err = fmt.Errorf("unexpected label value type for %q:%q; want string", k, v)
-					return
-				}
 				fields = append(fields, logstorage.Field{
 					Name:  bytesutil.ToUnsafeString(k),
 					Value: bytesutil.ToUnsafeString(val),
 				})
 			}
 		})
-		if err != nil {
-			return err
-		}
-		lmp.AddRow(ts, fields, nil)
+		processLogMessage(ts, fields)
 		fields = fields[:0]
 	}
-	return nil
+	return len(records), nil
 }

app/vlinsert/datadog/datadog_test.go

@@ -1,10 +1,12 @@
 package datadog
 
 import (
+	"fmt"
+	"strings"
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
 func TestReadLogsRequestFailure(t *testing.T) {
@@ -13,12 +15,16 @@ func TestReadLogsRequestFailure(t *testing.T) {
 
 		ts := time.Now().UnixNano()
 
-		lmp := &insertutils.TestLogMessageProcessor{}
-		if err := readLogsRequest(ts, []byte(data), lmp); err == nil {
+		processLogMessage := func(timestamp int64, fields []logstorage.Field) {
+			t.Fatalf("unexpected call to processLogMessage with timestamp=%d, fields=%s", timestamp, fields)
+		}
+
+		rows, err := readLogsRequest(ts, []byte(data), processLogMessage)
+		if err == nil {
 			t.Fatalf("expecting non-empty error")
 		}
-		if err := lmp.Verify(nil, ""); err != nil {
-			t.Fatalf("unexpected error: %s", err)
+		if rows != 0 {
+			t.Fatalf("unexpected non-zero rows=%d", rows)
 		}
 	}
 	f("foobar")
@@ -33,16 +39,30 @@ func TestReadLogsRequestSuccess(t *testing.T) {
 		t.Helper()
 
 		ts := time.Now().UnixNano()
-		var timestampsExpected []int64
-		for i := 0; i < rowsExpected; i++ {
-			timestampsExpected = append(timestampsExpected, ts)
+		var result string
+		processLogMessage := func(_ int64, fields []logstorage.Field) {
+			a := make([]string, len(fields))
+			for i, f := range fields {
+				a[i] = fmt.Sprintf("%q:%q", f.Name, f.Value)
+			}
+			if len(result) > 0 {
+				result = result + "\n"
+			}
+			s := "{" + strings.Join(a, ",") + "}"
+			result += s
 		}
-		lmp := &insertutils.TestLogMessageProcessor{}
-		if err := readLogsRequest(ts, []byte(data), lmp); err != nil {
+
+		// Read the request without compression
+		rows, err := readLogsRequest(ts, []byte(data), processLogMessage)
+		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if err := lmp.Verify(timestampsExpected, resultExpected); err != nil {
-			t.Fatalf("unexpected error: %s", err)
+		if rows != rowsExpected {
+			t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
+		}
+
+		if result != resultExpected {
+			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
 		}
 	}
 
@@ -54,12 +74,6 @@ func TestReadLogsRequestSuccess(t *testing.T) {
 			"hostname":"127.0.0.1",
 			"message":"bar",
 			"service":"test"
-		}, {
-			"ddsource":"nginx",
-                        "ddtags":"tag1:value1,tag2:value2",
-                        "hostname":"127.0.0.1",
-                        "message":{"message": "nested"},
-                        "service":"test"
 		}, {
 			"ddsource":"nginx",
 			"ddtags":"tag1:value1,tag2:value2",
@@ -92,9 +106,8 @@ func TestReadLogsRequestSuccess(t *testing.T) {
                         "service":"test"
 		}
 	]`
-	rowsExpected := 7
+	rowsExpected := 6
 	resultExpected := `{"ddsource":"nginx","tag1":"value1","tag2":"value2","hostname":"127.0.0.1","_msg":"bar","service":"test"}
-{"ddsource":"nginx","tag1":"value1","tag2":"value2","hostname":"127.0.0.1","_msg":"nested","service":"test"}
 {"ddsource":"nginx","tag1":"value1","tag2":"value2","hostname":"127.0.0.1","_msg":"foobar","service":"test"}
 {"ddsource":"nginx","tag1":"value1","tag2":"value2","hostname":"127.0.0.1","_msg":"baz","service":"test"}
 {"ddsource":"nginx","tag1":"value1","tag2":"value2","hostname":"127.0.0.1","_msg":"xyz","service":"test"}

app/vlinsert/elasticsearch/elasticsearch.go

@@ -1,6 +1,8 @@
 package elasticsearch
 
 import (
+	"bufio"
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -99,10 +101,9 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 			httpserver.Errorf(w, r, "%s", err)
 			return true
 		}
-		lmp := cp.NewLogMessageProcessor("elasticsearch_bulk")
+		lmp := cp.NewLogMessageProcessor()
 		isGzip := r.Header.Get("Content-Encoding") == "gzip"
-		streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
-		n, err := readBulkRequest(streamName, r.Body, isGzip, cp.TimeField, cp.MsgFields, lmp)
+		n, err := readBulkRequest(r.Body, isGzip, cp.TimeField, cp.MsgFields, lmp)
 		lmp.MustClose()
 		if err != nil {
 			logger.Warnf("cannot decode log message #%d in /_bulk request: %s, stream fields: %s", n, err, cp.StreamFields)
@@ -128,10 +129,11 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 
 var (
 	bulkRequestsTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/elasticsearch/_bulk"}`)
+	rowsIngestedTotal   = metrics.NewCounter(`vl_rows_ingested_total{type="elasticsearch_bulk"}`)
 	bulkRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/elasticsearch/_bulk"}`)
 )
 
-func readBulkRequest(streamName string, r io.Reader, isGzip bool, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (int, error) {
+func readBulkRequest(r io.Reader, isGzip bool, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (int, error) {
 	// See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
 
 	if isGzip {
@@ -146,29 +148,48 @@ func readBulkRequest(streamName string, r io.Reader, isGzip bool, timeField stri
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
-	lr := insertutils.NewLineReader(streamName, wcr)
+	lb := lineBufferPool.Get()
+	defer lineBufferPool.Put(lb)
+
+	lb.B = bytesutil.ResizeNoCopyNoOverallocate(lb.B, insertutils.MaxLineSizeBytes.IntN())
+	sc := bufio.NewScanner(wcr)
+	sc.Buffer(lb.B, len(lb.B))
 
 	n := 0
+	nCheckpoint := 0
 	for {
-		ok, err := readBulkLine(lr, timeField, msgFields, lmp)
+		ok, err := readBulkLine(sc, timeField, msgFields, lmp)
 		wcr.DecConcurrency()
 		if err != nil || !ok {
+			rowsIngestedTotal.Add(n - nCheckpoint)
 			return n, err
 		}
 		n++
+		if batchSize := n - nCheckpoint; n >= 1000 {
+			rowsIngestedTotal.Add(batchSize)
+			nCheckpoint = n
+		}
 	}
 }
 
-func readBulkLine(lr *insertutils.LineReader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (bool, error) {
+var lineBufferPool bytesutil.ByteBufferPool
+
+func readBulkLine(sc *bufio.Scanner, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (bool, error) {
 	var line []byte
 
 	// Read the command, must be "create" or "index"
 	for len(line) == 0 {
-		if !lr.NextLine() {
-			err := lr.Err()
-			return false, err
+		if !sc.Scan() {
+			if err := sc.Err(); err != nil {
+				if errors.Is(err, bufio.ErrTooLong) {
+					return false, fmt.Errorf(`cannot read "create" or "index" command, since its size exceeds -insert.maxLineSizeBytes=%d`,
+						insertutils.MaxLineSizeBytes.IntN())
+				}
+				return false, err
+			}
+			return false, nil
 		}
-		line = lr.Line
+		line = sc.Bytes()
 	}
 	lineStr := bytesutil.ToUnsafeString(line)
 	if !strings.Contains(lineStr, `"create"`) && !strings.Contains(lineStr, `"index"`) {
@@ -176,18 +197,16 @@ func readBulkLine(lr *insertutils.LineReader, timeField string, msgFields []stri
 	}
 
 	// Decode log message
-	if !lr.NextLine() {
-		if err := lr.Err(); err != nil {
+	if !sc.Scan() {
+		if err := sc.Err(); err != nil {
+			if errors.Is(err, bufio.ErrTooLong) {
+				return false, fmt.Errorf("cannot read log message, since its size exceeds -insert.maxLineSizeBytes=%d", insertutils.MaxLineSizeBytes.IntN())
+			}
 			return false, err
 		}
 		return false, fmt.Errorf(`missing log message after the "create" or "index" command`)
 	}
-	line = lr.Line
-	if len(line) == 0 {
-		// Special case - the line could be too long, so it was skipped.
-		// Continue parsing next lines.
-		return true, nil
-	}
+	line = sc.Bytes()
 	p := logstorage.GetJSONParser()
 	if err := p.ParseLogMessage(line); err != nil {
 		return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
@@ -201,7 +220,7 @@ func readBulkLine(lr *insertutils.LineReader, timeField string, msgFields []stri
 		ts = time.Now().UnixNano()
 	}
 	logstorage.RenameField(p.Fields, msgFields, "_msg")
-	lmp.AddRow(ts, p.Fields, nil)
+	lmp.AddRow(ts, p.Fields)
 	logstorage.PutJSONParser(p)
 
 	return true, nil

app/vlinsert/elasticsearch/elasticsearch_test.go

@@ -15,7 +15,7 @@ func TestReadBulkRequest_Failure(t *testing.T) {
 
 		tlp := &insertutils.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
-		rows, err := readBulkRequest("test", r, false, "_time", []string{"_msg"}, tlp)
+		rows, err := readBulkRequest(r, false, "_time", []string{"_msg"}, tlp)
 		if err == nil {
 			t.Fatalf("expecting non-empty error")
 		}
@@ -33,7 +33,7 @@ foobar`)
 }
 
 func TestReadBulkRequest_Success(t *testing.T) {
-	f := func(data, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
+	f := func(data, timeField, msgField string, rowsExpected int, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
 		msgFields := []string{"non_existing_foo", msgField, "non_exiting_bar"}
@@ -41,14 +41,14 @@ func TestReadBulkRequest_Success(t *testing.T) {
 
 		// Read the request without compression
 		r := bytes.NewBufferString(data)
-		rows, err := readBulkRequest("test", r, false, timeField, msgFields, tlp)
+		rows, err := readBulkRequest(r, false, timeField, msgFields, tlp)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if rows != len(timestampsExpected) {
-			t.Fatalf("unexpected rows read; got %d; want %d", rows, len(timestampsExpected))
+		if rows != rowsExpected {
+			t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
 		}
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(rowsExpected, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 
@@ -56,22 +56,22 @@ func TestReadBulkRequest_Success(t *testing.T) {
 		tlp = &insertutils.TestLogMessageProcessor{}
 		compressedData := compressData(data)
 		r = bytes.NewBufferString(compressedData)
-		rows, err = readBulkRequest("test", r, true, timeField, msgFields, tlp)
+		rows, err = readBulkRequest(r, true, timeField, msgFields, tlp)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if rows != len(timestampsExpected) {
-			t.Fatalf("unexpected rows read; got %d; want %d", rows, len(timestampsExpected))
+		if rows != rowsExpected {
+			t.Fatalf("unexpected rows read; got %d; want %d", rows, rowsExpected)
 		}
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(rowsExpected, timestampsExpected, resultExpected); err != nil {
 			t.Fatalf("verification failure after compression: %s", err)
 		}
 	}
 
 	// Verify an empty data
-	f("", "_time", "_msg", nil, "")
-	f("\n", "_time", "_msg", nil, "")
-	f("\n\n", "_time", "_msg", nil, "")
+	f("", "_time", "_msg", 0, nil, "")
+	f("\n", "_time", "_msg", 0, nil, "")
+	f("\n\n", "_time", "_msg", 0, nil, "")
 
 	// Verify non-empty data
 	data := `{"create":{"_index":"filebeat-8.8.0"}}
@@ -85,12 +85,13 @@ func TestReadBulkRequest_Success(t *testing.T) {
 `
 	timeField := "@timestamp"
 	msgField := "message"
+	rowsExpected := 4
 	timestampsExpected := []int64{1686026891735000000, 1686023292735000000, 1686026893735000000, 1686026893000000000}
 	resultExpected := `{"log.offset":"71770","log.file.path":"/var/log/auth.log","_msg":"foobar"}
 {"_msg":"baz"}
 {"_msg":"xyz","x":"y"}
 {"_msg":"qwe rty"}`
-	f(data, timeField, msgField, timestampsExpected, resultExpected)
+	f(data, timeField, msgField, rowsExpected, timestampsExpected, resultExpected)
 }
 
 func compressData(s string) string {

app/vlinsert/elasticsearch/elasticsearch_timing_test.go

@@ -41,7 +41,7 @@ func benchmarkReadBulkRequest(b *testing.B, isGzip bool) {
 		r := &bytes.Reader{}
 		for pb.Next() {
 			r.Reset(dataBytes)
-			_, err := readBulkRequest("test", r, isGzip, timeField, msgFields, blp)
+			_, err := readBulkRequest(r, isGzip, timeField, msgFields, blp)
 			if err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}

app/vlinsert/insertutils/common_params.go

@@ -137,12 +137,10 @@ func GetCommonParamsForSyslog(tenantID logstorage.TenantID, streamFields, ignore
 
 // LogMessageProcessor is an interface for log message processors.
 type LogMessageProcessor interface {
-	// AddRow must add row to the LogMessageProcessor with the given timestamp and fields.
-	//
-	// If streamFields is non-nil, then the given streamFields must be used as log stream fields instead of pre-configured fields.
+	// AddRow must add row to the LogMessageProcessor with the given timestamp and the given fields.
 	//
 	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can re-use them.
-	AddRow(timestamp int64, fields, streamFields []logstorage.Field)
+	AddRow(timestamp int64, fields []logstorage.Field)
 
 	// MustClose() must flush all the remaining fields and free up resources occupied by LogMessageProcessor.
 	MustClose()
@@ -156,9 +154,6 @@ type logMessageProcessor struct {
 
 	cp *CommonParams
 	lr *logstorage.LogRows
-
-	rowsIngestedTotal  *metrics.Counter
-	bytesIngestedTotal *metrics.Counter
 }
 
 func (lmp *logMessageProcessor) initPeriodicFlush() {
@@ -188,16 +183,10 @@ func (lmp *logMessageProcessor) initPeriodicFlush() {
 }
 
 // AddRow adds new log message to lmp with the given timestamp and fields.
-//
-// If streamFields is non-nil, then it is used as log stream fields instead of the pre-configured stream fields.
-func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []logstorage.Field) {
+func (lmp *logMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field) {
 	lmp.mu.Lock()
 	defer lmp.mu.Unlock()
 
-	lmp.rowsIngestedTotal.Inc()
-	n := logstorage.EstimatedJSONRowLen(fields)
-	lmp.bytesIngestedTotal.Add(n)
-
 	if len(fields) > *MaxFieldsPerLine {
 		rf := logstorage.RowFormatter(fields)
 		logger.Warnf("dropping log line with %d fields; it exceeds -insert.maxFieldsPerLine=%d; %s", len(fields), *MaxFieldsPerLine, rf)
@@ -205,7 +194,7 @@ func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []l
 		return
 	}
 
-	lmp.lr.MustAdd(lmp.cp.TenantID, timestamp, fields, streamFields)
+	lmp.lr.MustAdd(lmp.cp.TenantID, timestamp, fields)
 	if lmp.cp.Debug {
 		s := lmp.lr.GetRowString(0)
 		lmp.lr.ResetKeepSettings()
@@ -238,17 +227,12 @@ func (lmp *logMessageProcessor) MustClose() {
 // NewLogMessageProcessor returns new LogMessageProcessor for the given cp.
 //
 // MustClose() must be called on the returned LogMessageProcessor when it is no longer needed.
-func (cp *CommonParams) NewLogMessageProcessor(protocolName string) LogMessageProcessor {
+func (cp *CommonParams) NewLogMessageProcessor() LogMessageProcessor {
 	lr := logstorage.GetLogRows(cp.StreamFields, cp.IgnoreFields, cp.ExtraFields, *defaultMsgValue)
-	rowsIngestedTotal := metrics.GetOrCreateCounter(fmt.Sprintf("vl_rows_ingested_total{type=%q}", protocolName))
-	bytesIngestedTotal := metrics.GetOrCreateCounter(fmt.Sprintf("vl_bytes_ingested_total{type=%q}", protocolName))
 	lmp := &logMessageProcessor{
 		cp: cp,
 		lr: lr,
 
-		rowsIngestedTotal:  rowsIngestedTotal,
-		bytesIngestedTotal: bytesIngestedTotal,
-
 		stopCh: make(chan struct{}),
 	}
 	lmp.initPeriodicFlush()

app/vlinsert/insertutils/line_reader.go

@@ -1,135 +0,0 @@
-package insertutils
-
-import (
-	"bytes"
-	"errors"
-	"fmt"
-	"io"
-
-	"github.com/VictoriaMetrics/metrics"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
-)
-
-// LineReader reads newline-delimited lines from the underlying reader
-type LineReader struct {
-	// Line contains the next line read after the call to NextLine
-	Line []byte
-
-	// name is the LineReader name
-	name string
-
-	// r is the underlying reader to read data from
-	r io.Reader
-
-	// buf is a buffer for reading the next line
-	buf []byte
-
-	// err is the last error when reading data from r
-	err error
-
-	// eofReached is set to true when all the data is read from r
-	eofReached bool
-}
-
-// NewLineReader returns LineReader for r.
-func NewLineReader(name string, r io.Reader) *LineReader {
-	return &LineReader{
-		name: name,
-		r:    r,
-	}
-}
-
-// NextLine reads the next line from the underlying reader.
-//
-// It returns true if the next line is successfully read into Line.
-// If the line length exceeds MaxLineSizeBytes, then this line is skipped
-// and an empty line is returned instead.
-//
-// If false is returned, then no more lines left to read from r.
-// Check for Err in this case.
-func (lr *LineReader) NextLine() bool {
-	for {
-		if len(lr.buf) == 0 {
-			if lr.err != nil || lr.eofReached {
-				return false
-			}
-			if !lr.readMoreData() {
-				return false
-			}
-			if len(lr.buf) == 0 && lr.eofReached {
-				return false
-			}
-		}
-
-		if n := bytes.IndexByte(lr.buf, '\n'); n >= 0 {
-			lr.Line = append(lr.Line[:0], lr.buf[:n]...)
-			lr.buf = append(lr.buf[:0], lr.buf[n+1:]...)
-			return true
-		}
-		if lr.eofReached {
-			lr.Line = append(lr.Line[:0], lr.buf...)
-			lr.buf = lr.buf[:0]
-			return true
-		}
-		if !lr.readMoreData() {
-			return false
-		}
-	}
-}
-
-// Err returns the last error after NextLine call.
-func (lr *LineReader) Err() error {
-	if lr.err == nil {
-		return nil
-	}
-	return fmt.Errorf("%s: %s", lr.name, lr.err)
-}
-
-func (lr *LineReader) readMoreData() bool {
-	bufLen := len(lr.buf)
-	if bufLen >= MaxLineSizeBytes.IntN() {
-		logger.Warnf("%s: the line length exceeds -insert.maxLineSizeBytes=%d; skipping it; line contents=%q", lr.name, MaxLineSizeBytes.IntN(), lr.buf)
-		tooLongLinesSkipped.Inc()
-		return lr.skipUntilNextLine()
-	}
-
-	lr.buf = slicesutil.SetLength(lr.buf, MaxLineSizeBytes.IntN())
-	n, err := lr.r.Read(lr.buf[bufLen:])
-	lr.buf = lr.buf[:bufLen+n]
-	if err != nil {
-		if errors.Is(err, io.EOF) {
-			lr.eofReached = true
-			return true
-		}
-		lr.err = fmt.Errorf("cannot read the next line: %s", err)
-	}
-	return n > 0
-}
-
-var tooLongLinesSkipped = metrics.NewCounter("vl_too_long_lines_skipped_total")
-
-func (lr *LineReader) skipUntilNextLine() bool {
-	for {
-		lr.buf = slicesutil.SetLength(lr.buf, MaxLineSizeBytes.IntN())
-		n, err := lr.r.Read(lr.buf)
-		lr.buf = lr.buf[:n]
-		if err != nil {
-			if errors.Is(err, io.EOF) {
-				lr.eofReached = true
-				lr.buf = lr.buf[:0]
-				return true
-			}
-			lr.err = fmt.Errorf("cannot skip the current line: %s", err)
-			return false
-		}
-		if n := bytes.IndexByte(lr.buf, '\n'); n >= 0 {
-			// Include \n in the buf, so too long line is replaced with an empty line.
-			// This is needed for maintaining synchorinzation consistency between lines
-			// in protocols such as Elasticsearch bulk import.
-			lr.buf = append(lr.buf[:0], lr.buf[n:]...)
-			return true
-		}
-	}
-}

app/vlinsert/insertutils/line_reader_test.go

@@ -1,161 +0,0 @@
-package insertutils
-
-import (
-	"bytes"
-	"fmt"
-	"io"
-	"reflect"
-	"testing"
-)
-
-func TestLineReader_Success(t *testing.T) {
-	f := func(data string, linesExpected []string) {
-		t.Helper()
-
-		r := bytes.NewBufferString(data)
-		lr := NewLineReader("foo", r)
-		var lines []string
-		for lr.NextLine() {
-			lines = append(lines, string(lr.Line))
-		}
-		if err := lr.Err(); err != nil {
-			t.Fatalf("unexpected error: %s", err)
-		}
-		if lr.NextLine() {
-			t.Fatalf("expecting error on the second call to NextLine()")
-		}
-		if !reflect.DeepEqual(lines, linesExpected) {
-			t.Fatalf("unexpected lines\ngot\n%q\nwant\n%q", lines, linesExpected)
-		}
-	}
-
-	f("", nil)
-	f("\n", []string{""})
-	f("\n\n", []string{"", ""})
-	f("foo", []string{"foo"})
-	f("foo\n", []string{"foo"})
-	f("\nfoo", []string{"", "foo"})
-	f("foo\n\n", []string{"foo", ""})
-	f("foo\nbar", []string{"foo", "bar"})
-	f("foo\nbar\n", []string{"foo", "bar"})
-	f("\nfoo\n\nbar\n\n", []string{"", "foo", "", "bar", ""})
-}
-
-func TestLineReader_SkipUntilNextLine(t *testing.T) {
-	f := func(data string, linesExpected []string) {
-		t.Helper()
-
-		r := bytes.NewBufferString(data)
-		lr := NewLineReader("foo", r)
-		var lines []string
-		for lr.NextLine() {
-			lines = append(lines, string(lr.Line))
-		}
-		if err := lr.Err(); err != nil {
-			t.Fatalf("unexpected error for data=%q: %s", data, err)
-		}
-		if lr.NextLine() {
-			t.Fatalf("expecting error on the second call to NextLine()")
-		}
-		if !reflect.DeepEqual(lines, linesExpected) {
-			t.Fatalf("unexpected lines for data=%q\ngot\n%q\nwant\n%q", data, lines, linesExpected)
-		}
-	}
-
-	for _, overflow := range []int{0, 100, MaxLineSizeBytes.IntN(), MaxLineSizeBytes.IntN() + 1, 2 * MaxLineSizeBytes.IntN()} {
-		longLineLen := MaxLineSizeBytes.IntN() + overflow
-		longLine := string(make([]byte, longLineLen))
-
-		// Single long line
-		data := longLine
-		f(data, nil)
-
-		// Multiple long lines
-		data = longLine + "\n" + longLine
-		f(data, []string{""})
-
-		data = longLine + "\n" + longLine + "\n"
-		f(data, []string{"", ""})
-
-		// Long line in the middle
-		data = "foo\n" + longLine + "\nbar"
-		f(data, []string{"foo", "", "bar"})
-
-		// Multiple long lines in the middle
-		data = "foo\n" + longLine + "\n" + longLine + "\nbar"
-		f(data, []string{"foo", "", "", "bar"})
-
-		// Long line in the end
-		data = "foo\n" + longLine
-		f(data, []string{"foo"})
-
-		// Long line in the end
-		data = "foo\n" + longLine + "\n"
-		f(data, []string{"foo", ""})
-	}
-}
-
-func TestLineReader_Failure(t *testing.T) {
-	f := func(data string, linesExpected []string) {
-		t.Helper()
-
-		fr := &failureReader{
-			r: bytes.NewBufferString(data),
-		}
-		lr := NewLineReader("foo", fr)
-		var lines []string
-		for lr.NextLine() {
-			lines = append(lines, string(lr.Line))
-		}
-		if err := lr.Err(); err == nil {
-			t.Fatalf("expecting non-nil error")
-		}
-		if lr.NextLine() {
-			t.Fatalf("expecting error on the second call to NextLine()")
-		}
-		if err := lr.Err(); err == nil {
-			t.Fatalf("expecting non-nil error on the second call")
-		}
-		if !reflect.DeepEqual(lines, linesExpected) {
-			t.Fatalf("unexpected lines\ngot\n%q\nwant\n%q", lines, linesExpected)
-		}
-	}
-
-	f("", nil)
-	f("foo", nil)
-	f("foo\n", []string{"foo"})
-	f("\n", []string{""})
-	f("foo\nbar", []string{"foo"})
-	f("foo\nbar\n", []string{"foo", "bar"})
-	f("\nfoo\nbar\n\n", []string{"", "foo", "bar", ""})
-
-	// long line
-	longLineLen := MaxLineSizeBytes.IntN()
-	for _, overflow := range []int{0, 100, MaxLineSizeBytes.IntN(), MaxLineSizeBytes.IntN() + 1, 2 * MaxLineSizeBytes.IntN()} {
-		longLine := string(make([]byte, longLineLen+overflow))
-
-		data := longLine
-		f(data, nil)
-
-		data = "foo\n" + longLine
-		f(data, []string{"foo"})
-
-		data = longLine + "\nfoo"
-		f(data, []string{""})
-
-		data = longLine + "\nfoo\n"
-		f(data, []string{"", "foo"})
-	}
-}
-
-type failureReader struct {
-	r io.Reader
-}
-
-func (r *failureReader) Read(p []byte) (int, error) {
-	n, _ := r.r.Read(p)
-	if n > 0 {
-		return n, nil
-	}
-	return 0, fmt.Errorf("some error")
-}

app/vlinsert/insertutils/testutils.go

@@ -15,10 +15,7 @@ type TestLogMessageProcessor struct {
 }
 
 // AddRow adds row with the given timestamp and fields to tlp
-func (tlp *TestLogMessageProcessor) AddRow(timestamp int64, fields, streamFields []logstorage.Field) {
-	if streamFields != nil {
-		panic(fmt.Errorf("BUG: streamFields must be nil; got %v", streamFields))
-	}
+func (tlp *TestLogMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field) {
 	tlp.timestamps = append(tlp.timestamps, timestamp)
 	tlp.rows = append(tlp.rows, string(logstorage.MarshalFieldsToJSON(nil, fields)))
 }
@@ -28,10 +25,10 @@ func (tlp *TestLogMessageProcessor) MustClose() {
 }
 
 // Verify verifies the number of rows, timestamps and results after AddRow calls.
-func (tlp *TestLogMessageProcessor) Verify(timestampsExpected []int64, resultExpected string) error {
+func (tlp *TestLogMessageProcessor) Verify(rowsExpected int, timestampsExpected []int64, resultExpected string) error {
 	result := strings.Join(tlp.rows, "\n")
-	if len(tlp.rows) != len(timestampsExpected) {
-		return fmt.Errorf("unexpected rows read; got %d; want %d;\nrows read:\n%s\nrows wanted\n%s", len(tlp.rows), len(timestampsExpected), result, resultExpected)
+	if len(tlp.rows) != rowsExpected {
+		return fmt.Errorf("unexpected rows read; got %d; want %d;\nrows read:\n%s\nrows wanted\n%s", len(tlp.rows), rowsExpected, result, resultExpected)
 	}
 
 	if !reflect.DeepEqual(tlp.timestamps, timestampsExpected) {
@@ -48,7 +45,7 @@ func (tlp *TestLogMessageProcessor) Verify(timestampsExpected []int64, resultExp
 type BenchmarkLogMessageProcessor struct{}
 
 // AddRow implements LogMessageProcessor interface.
-func (blp *BenchmarkLogMessageProcessor) AddRow(_ int64, _, _ []logstorage.Field) {
+func (blp *BenchmarkLogMessageProcessor) AddRow(_ int64, _ []logstorage.Field) {
 }
 
 // MustClose implements LogMessageProcessor interface.

app/vlinsert/journald/journald.go

@@ -120,8 +120,8 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
 		return
 	}
 
-	lmp := cp.NewLogMessageProcessor("journald")
-	err = parseJournaldRequest(data, lmp, cp)
+	lmp := cp.NewLogMessageProcessor()
+	n, err := parseJournaldRequest(data, lmp, cp)
 	lmp.MustClose()
 	if err != nil {
 		errorsTotal.Inc()
@@ -129,6 +129,8 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
 		return
 	}
 
+	rowsIngestedJournaldTotal.Add(n)
+
 	// update requestJournaldDuration only for successfully parsed requests
 	// There is no need in updating requestJournaldDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
@@ -136,14 +138,16 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
 }
 
 var (
-	requestsJournaldTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/journald/upload"}`)
-	errorsTotal           = metrics.NewCounter(`vl_http_errors_total{path="/insert/journald/upload"}`)
+	rowsIngestedJournaldTotal = metrics.NewCounter(`vl_rows_ingested_total{type="journald", format="journald"}`)
 
-	requestJournaldDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/journald/upload"}`)
+	requestsJournaldTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/journald/upload",format="journald"}`)
+	errorsTotal           = metrics.NewCounter(`vl_http_errors_total{path="/insert/journald/upload",format="journald"}`)
+
+	requestJournaldDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/journald/upload",format="journald"}`)
 )
 
 // See https://systemd.io/JOURNAL_EXPORT_FORMATS/#journal-export-format
-func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *insertutils.CommonParams) error {
+func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *insertutils.CommonParams) (rowsIngested int, err error) {
 	var fields []logstorage.Field
 	var ts int64
 	var size uint64
@@ -166,14 +170,15 @@ func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *
 				if ts == 0 {
 					ts = currentTimestamp
 				}
-				lmp.AddRow(ts, fields, nil)
+				lmp.AddRow(ts, fields)
+				rowsIngested++
 				fields = fields[:0]
 			}
 			// skip newline separator
 			data = data[1:]
 			continue
 		case idx < 0:
-			return fmt.Errorf("missing new line separator, unread data left=%d", len(data))
+			return rowsIngested, fmt.Errorf("missing new line separator, unread data left=%d", len(data))
 		}
 
 		idx = bytes.IndexByte(line, '=')
@@ -186,46 +191,46 @@ func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *
 		} else {
 			name = bytesutil.ToUnsafeString(line)
 			if len(data) == 0 {
-				return fmt.Errorf("unexpected zero data for binary field value of key=%s", name)
+				return rowsIngested, fmt.Errorf("unexpected zero data for binary field value of key=%s", name)
 			}
 			// size of binary data encoded as le i64 at the begging
 			idx, err := binary.Decode(data, binary.LittleEndian, &size)
 			if err != nil {
-				return fmt.Errorf("failed to extract binary field %q value size: %w", name, err)
+				return rowsIngested, fmt.Errorf("failed to extract binary field %q value size: %w", name, err)
 			}
 			// skip binary data sise
 			data = data[idx:]
 			if size == 0 {
-				return fmt.Errorf("unexpected zero binary data size decoded %d", size)
+				return rowsIngested, fmt.Errorf("unexpected zero binary data size decoded %d", size)
 			}
 			if int(size) > len(data) {
-				return fmt.Errorf("binary data size=%d cannot exceed size of the data at buffer=%d", size, len(data))
+				return rowsIngested, fmt.Errorf("binary data size=%d cannot exceed size of the data at buffer=%d", size, len(data))
 			}
 			value = bytesutil.ToUnsafeString(data[:size])
 			data = data[int(size):]
 			// binary data must has new line separator for the new line or next field
 			if len(data) == 0 {
-				return fmt.Errorf("unexpected empty buffer after binary field=%s read", name)
+				return rowsIngested, fmt.Errorf("unexpected empty buffer after binary field=%s read", name)
 			}
 			lastB := data[0]
 			if lastB != '\n' {
-				return fmt.Errorf("expected new line separator after binary field=%s, got=%s", name, string(lastB))
+				return rowsIngested, fmt.Errorf("expected new line separator after binary field=%s, got=%s", name, string(lastB))
 			}
 			data = data[1:]
 		}
 		// https://github.com/systemd/systemd/blob/main/src/libsystemd/sd-journal/journal-file.c#L1703
 		if len(name) > journaldEntryMaxNameLen {
-			return fmt.Errorf("journald entry name should not exceed %d symbols, got: %q", journaldEntryMaxNameLen, name)
+			return rowsIngested, fmt.Errorf("journald entry name should not exceed %d symbols, got: %q", journaldEntryMaxNameLen, name)
 		}
 		if !allowedJournaldEntryNameChars.MatchString(name) {
-			return fmt.Errorf("journald entry name should consist of `A-Z0-9_` characters and must start from non-digit symbol")
+			return rowsIngested, fmt.Errorf("journald entry name should consist of `A-Z0-9_` characters and must start from non-digit symbol")
 		}
 		if name == cp.TimeField {
-			n, err := strconv.ParseInt(value, 10, 64)
+			ts, err = strconv.ParseInt(value, 10, 64)
 			if err != nil {
-				return fmt.Errorf("failed to parse Journald timestamp, %w", err)
+				return 0, fmt.Errorf("failed to parse Journald timestamp, %w", err)
 			}
-			ts = n * 1e3
+			ts *= 1e3
 			continue
 		}
 
@@ -244,7 +249,8 @@ func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *
 		if ts == 0 {
 			ts = currentTimestamp
 		}
-		lmp.AddRow(ts, fields, nil)
+		lmp.AddRow(ts, fields)
+		rowsIngested++
 	}
-	return nil
+	return rowsIngested, nil
 }

app/vlinsert/journald/journald_test.go

@@ -14,11 +14,12 @@ func TestPushJournaldOk(t *testing.T) {
 			TimeField: "__REALTIME_TIMESTAMP",
 			MsgFields: []string{"MESSAGE"},
 		}
-		if err := parseJournaldRequest([]byte(src), tlp, cp); err != nil {
+		n, err := parseJournaldRequest([]byte(src), tlp, cp)
+		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(n, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 	}
@@ -49,7 +50,8 @@ func TestPushJournald_Failure(t *testing.T) {
 			TimeField: "__REALTIME_TIMESTAMP",
 			MsgFields: []string{"MESSAGE"},
 		}
-		if err := parseJournaldRequest([]byte(data), tlp, cp); err == nil {
+		_, err := parseJournaldRequest([]byte(data), tlp, cp)
+		if err == nil {
 			t.Fatalf("expected non nil error")
 		}
 	}

app/vlinsert/jsonline/jsonline.go

@@ -1,6 +1,8 @@
 package jsonline
 
 import (
+	"bufio"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -8,6 +10,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
 	"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"
@@ -49,9 +52,8 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 		reader = zr
 	}
 
-	lmp := cp.NewLogMessageProcessor("jsonline")
-	streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
-	err = processStreamInternal(streamName, reader, cp.TimeField, cp.MsgFields, lmp)
+	lmp := cp.NewLogMessageProcessor()
+	err = processStreamInternal(reader, cp.TimeField, cp.MsgFields, lmp)
 	lmp.MustClose()
 
 	if err != nil {
@@ -64,15 +66,20 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func processStreamInternal(streamName string, r io.Reader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) error {
+func processStreamInternal(r io.Reader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) error {
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
-	lr := insertutils.NewLineReader(streamName, wcr)
+	lb := lineBufferPool.Get()
+	defer lineBufferPool.Put(lb)
+
+	lb.B = bytesutil.ResizeNoCopyNoOverallocate(lb.B, insertutils.MaxLineSizeBytes.IntN())
+	sc := bufio.NewScanner(wcr)
+	sc.Buffer(lb.B, len(lb.B))
 
 	n := 0
 	for {
-		ok, err := readLine(lr, timeField, msgFields, lmp)
+		ok, err := readLine(sc, timeField, msgFields, lmp)
 		wcr.DecConcurrency()
 		if err != nil {
 			errorsTotal.Inc()
@@ -82,17 +89,23 @@ func processStreamInternal(streamName string, r io.Reader, timeField string, msg
 			return nil
 		}
 		n++
+		rowsIngestedTotal.Inc()
 	}
 }
 
-func readLine(lr *insertutils.LineReader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (bool, error) {
+func readLine(sc *bufio.Scanner, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (bool, error) {
 	var line []byte
 	for len(line) == 0 {
-		if !lr.NextLine() {
-			err := lr.Err()
-			return false, err
+		if !sc.Scan() {
+			if err := sc.Err(); err != nil {
+				if errors.Is(err, bufio.ErrTooLong) {
+					return false, fmt.Errorf(`cannot read json line, since its size exceeds -insert.maxLineSizeBytes=%d`, insertutils.MaxLineSizeBytes.IntN())
+				}
+				return false, err
+			}
+			return false, nil
 		}
-		line = lr.Line
+		line = sc.Bytes()
 	}
 
 	p := logstorage.GetJSONParser()
@@ -104,13 +117,17 @@ func readLine(lr *insertutils.LineReader, timeField string, msgFields []string,
 		return false, fmt.Errorf("cannot get timestamp: %w", err)
 	}
 	logstorage.RenameField(p.Fields, msgFields, "_msg")
-	lmp.AddRow(ts, p.Fields, nil)
+	lmp.AddRow(ts, p.Fields)
 	logstorage.PutJSONParser(p)
 
 	return true, nil
 }
 
+var lineBufferPool bytesutil.ByteBufferPool
+
 var (
+	rowsIngestedTotal = metrics.NewCounter(`vl_rows_ingested_total{type="jsonline"}`)
+
 	requestsTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/jsonline"}`)
 	errorsTotal   = metrics.NewCounter(`vl_http_errors_total{path="/insert/jsonline"}`)
 

app/vlinsert/jsonline/jsonline_test.go

@@ -8,17 +8,17 @@ import (
 )
 
 func TestProcessStreamInternal_Success(t *testing.T) {
-	f := func(data, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
+	f := func(data, timeField, msgField string, rowsExpected int, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
 		msgFields := []string{msgField}
 		tlp := &insertutils.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
-		if err := processStreamInternal("test", r, timeField, msgFields, tlp); err != nil {
+		if err := processStreamInternal(r, timeField, msgFields, tlp); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(rowsExpected, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 	}
@@ -29,11 +29,12 @@ func TestProcessStreamInternal_Success(t *testing.T) {
 `
 	timeField := "@timestamp"
 	msgField := "message"
+	rowsExpected := 3
 	timestampsExpected := []int64{1686026891735000000, 1686023292735000000, 1686026893735000000}
 	resultExpected := `{"log.offset":"71770","log.file.path":"/var/log/auth.log","_msg":"foobar"}
 {"_msg":"baz"}
 {"_msg":"xyz","x":"y"}`
-	f(data, timeField, msgField, timestampsExpected, resultExpected)
+	f(data, timeField, msgField, rowsExpected, timestampsExpected, resultExpected)
 
 	// Non-existing msgField
 	data = `{"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
@@ -41,10 +42,11 @@ func TestProcessStreamInternal_Success(t *testing.T) {
 `
 	timeField = "@timestamp"
 	msgField = "foobar"
+	rowsExpected = 2
 	timestampsExpected = []int64{1686026891735000000, 1686023292735000000}
 	resultExpected = `{"log.offset":"71770","log.file.path":"/var/log/auth.log","message":"foobar"}
 {"message":"baz"}`
-	f(data, timeField, msgField, timestampsExpected, resultExpected)
+	f(data, timeField, msgField, rowsExpected, timestampsExpected, resultExpected)
 }
 
 func TestProcessStreamInternal_Failure(t *testing.T) {
@@ -53,7 +55,7 @@ func TestProcessStreamInternal_Failure(t *testing.T) {
 
 		tlp := &insertutils.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
-		if err := processStreamInternal("test", r, "time", nil, tlp); err == nil {
+		if err := processStreamInternal(r, "time", nil, tlp); err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
 	}

app/vlinsert/loki/loki_json.go

@@ -53,15 +53,16 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
-	lmp := cp.NewLogMessageProcessor("loki_json")
-	useDefaultStreamFields := len(cp.StreamFields) == 0
-	err = parseJSONRequest(data, lmp, useDefaultStreamFields)
+	lmp := cp.NewLogMessageProcessor()
+	n, err := parseJSONRequest(data, lmp)
 	lmp.MustClose()
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse Loki json request: %s; data=%s", err, data)
 		return
 	}
 
+	rowsIngestedJSONTotal.Add(n)
+
 	// update requestJSONDuration only for successfully parsed requests
 	// There is no need in updating requestJSONDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
@@ -69,29 +70,31 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 }
 
 var (
-	requestsJSONTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="json"}`)
-	requestJSONDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
+	requestsJSONTotal     = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="json"}`)
+	rowsIngestedJSONTotal = metrics.NewCounter(`vl_rows_ingested_total{type="loki",format="json"}`)
+	requestJSONDuration   = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
 )
 
-func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefaultStreamFields bool) error {
+func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor) (int, error) {
 	p := parserPool.Get()
 	defer parserPool.Put(p)
 	v, err := p.ParseBytes(data)
 	if err != nil {
-		return fmt.Errorf("cannot parse JSON request body: %w", err)
+		return 0, fmt.Errorf("cannot parse JSON request body: %w", err)
 	}
 
 	streamsV := v.Get("streams")
 	if streamsV == nil {
-		return fmt.Errorf("missing `streams` item in the parsed JSON")
+		return 0, fmt.Errorf("missing `streams` item in the parsed JSON")
 	}
 	streams, err := streamsV.Array()
 	if err != nil {
-		return fmt.Errorf("`streams` item in the parsed JSON must contain an array; got %q", streamsV)
+		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 _, stream := range streams {
 		// populate common labels from `stream` dict
 		commonFields = commonFields[:0]
@@ -100,7 +103,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 		if labelsV != nil {
 			o, err := labelsV.Object()
 			if err != nil {
-				return fmt.Errorf("`stream` item in the parsed JSON must contain an object; got %q", labelsV)
+				return rowsIngested, fmt.Errorf("`stream` item in the parsed JSON must contain an object; got %q", labelsV)
 			}
 			labels = o
 		}
@@ -116,37 +119,37 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 			})
 		})
 		if err != nil {
-			return fmt.Errorf("error when parsing `stream` object: %w", err)
+			return rowsIngested, fmt.Errorf("error when parsing `stream` object: %w", err)
 		}
 
 		// populate messages from `values` array
 		linesV := stream.Get("values")
 		if linesV == nil {
-			return fmt.Errorf("missing `values` item in the parsed `stream` object %q", stream)
+			return rowsIngested, fmt.Errorf("missing `values` item in the parsed `stream` object %q", stream)
 		}
 		lines, err := linesV.Array()
 		if err != nil {
-			return fmt.Errorf("`values` item in the parsed JSON must contain an array; got %q", linesV)
+			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 fmt.Errorf("unexpected contents of `values` item; want array; got %q", line)
+				return rowsIngested, fmt.Errorf("unexpected contents of `values` item; want array; got %q", line)
 			}
 			if len(lineA) < 2 || len(lineA) > 3 {
-				return fmt.Errorf("unexpected number of values in `values` item array %q; got %d want 2 or 3", line, len(lineA))
+				return rowsIngested, fmt.Errorf("unexpected number of values in `values` item array %q; got %d want 2 or 3", line, len(lineA))
 			}
 
 			// parse timestamp
 			timestamp, err := lineA[0].StringBytes()
 			if err != nil {
-				return fmt.Errorf("unexpected log timestamp type for %q; want string", lineA[0])
+				return rowsIngested, fmt.Errorf("unexpected log timestamp type for %q; want string", lineA[0])
 			}
 			ts, err := parseLokiTimestamp(bytesutil.ToUnsafeString(timestamp))
 			if err != nil {
-				return fmt.Errorf("cannot parse log timestamp %q: %w", timestamp, err)
+				return rowsIngested, fmt.Errorf("cannot parse log timestamp %q: %w", timestamp, err)
 			}
 			if ts == 0 {
 				ts = currentTimestamp
@@ -155,7 +158,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 			// parse log message
 			msg, err := lineA[1].StringBytes()
 			if err != nil {
-				return fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
+				return rowsIngested, fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
 			}
 
 			fields = append(fields[:len(commonFields)], logstorage.Field{
@@ -167,7 +170,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 			if len(lineA) > 2 {
 				structuredMetadata, err := lineA[2].Object()
 				if err != nil {
-					return fmt.Errorf("unexpected structured metadata type for %q; want JSON object", lineA[2])
+					return rowsIngested, fmt.Errorf("unexpected structured metadata type for %q; want JSON object", lineA[2])
 				}
 
 				structuredMetadata.Visit(func(k []byte, v *fastjson.Value) {
@@ -183,18 +186,15 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 					})
 				})
 				if err != nil {
-					return fmt.Errorf("error when parsing `structuredMetadata` object: %w", err)
+					return rowsIngested, fmt.Errorf("error when parsing `structuredMetadata` object: %w", err)
 				}
 			}
-			var streamFields []logstorage.Field
-			if useDefaultStreamFields {
-				streamFields = commonFields
-			}
-			lmp.AddRow(ts, fields, streamFields)
+			lmp.AddRow(ts, fields)
 		}
+		rowsIngested += len(lines)
 	}
 
-	return nil
+	return rowsIngested, nil
 }
 
 func parseLokiTimestamp(s string) (int64, error) {

app/vlinsert/loki/loki_json_test.go

@@ -11,11 +11,12 @@ func TestParseJSONRequest_Failure(t *testing.T) {
 		t.Helper()
 
 		tlp := &insertutils.TestLogMessageProcessor{}
-		if err := parseJSONRequest([]byte(s), tlp, false); err == nil {
+		n, err := parseJSONRequest([]byte(s), tlp)
+		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
-		if err := tlp.Verify(nil, ""); err != nil {
-			t.Fatalf("unexpected error: %s", err)
+		if n != 0 {
+			t.Fatalf("unexpected number of parsed lines: %d; want 0", n)
 		}
 	}
 	f(``)
@@ -65,10 +66,11 @@ func TestParseJSONRequest_Success(t *testing.T) {
 
 		tlp := &insertutils.TestLogMessageProcessor{}
 
-		if err := parseJSONRequest([]byte(s), tlp, false); err != nil {
+		n, err := parseJSONRequest([]byte(s), tlp)
+		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(n, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 	}

app/vlinsert/loki/loki_json_timing_test.go

@@ -28,7 +28,8 @@ func benchmarkParseJSONRequest(b *testing.B, streams, rows, labels int) {
 	b.RunParallel(func(pb *testing.PB) {
 		data := getJSONBody(streams, rows, labels)
 		for pb.Next() {
-			if err := parseJSONRequest(data, blp, false); err != nil {
+			_, err := parseJSONRequest(data, blp)
+			if err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}
 		}

app/vlinsert/loki/loki_protobuf.go

@@ -44,15 +44,16 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
-	lmp := cp.NewLogMessageProcessor("loki_protobuf")
-	useDefaultStreamFields := len(cp.StreamFields) == 0
-	err = parseProtobufRequest(data, lmp, useDefaultStreamFields)
+	lmp := cp.NewLogMessageProcessor()
+	n, err := parseProtobufRequest(data, lmp)
 	lmp.MustClose()
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse Loki protobuf request: %s", err)
 		return
 	}
 
+	rowsIngestedProtobufTotal.Add(n)
+
 	// update requestProtobufDuration only for successfully parsed requests
 	// There is no need in updating requestProtobufDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
@@ -60,17 +61,18 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 }
 
 var (
-	requestsProtobufTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
-	requestProtobufDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
+	requestsProtobufTotal     = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
+	rowsIngestedProtobufTotal = metrics.NewCounter(`vl_rows_ingested_total{type="loki",format="protobuf"}`)
+	requestProtobufDuration   = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
 )
 
-func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefaultStreamFields bool) error {
+func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor) (int, error) {
 	bb := bytesBufPool.Get()
 	defer bytesBufPool.Put(bb)
 
 	buf, err := snappy.Decode(bb.B[:cap(bb.B)], data)
 	if err != nil {
-		return fmt.Errorf("cannot decode snappy-encoded request body: %w", err)
+		return 0, fmt.Errorf("cannot decode snappy-encoded request body: %w", err)
 	}
 	bb.B = buf
 
@@ -79,12 +81,13 @@ func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useD
 
 	err = req.UnmarshalProtobuf(bb.B)
 	if err != nil {
-		return fmt.Errorf("cannot parse request body: %w", err)
+		return 0, fmt.Errorf("cannot parse request body: %w", err)
 	}
 
 	fields := getFields()
 	defer putFields(fields)
 
+	rowsIngested := 0
 	streams := req.Streams
 	currentTimestamp := time.Now().UnixNano()
 	for i := range streams {
@@ -93,7 +96,7 @@ func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useD
 		// Labels are same for all entries in the stream.
 		fields.fields, err = parsePromLabels(fields.fields[:0], stream.Labels)
 		if err != nil {
-			return fmt.Errorf("cannot parse stream labels %q: %w", stream.Labels, err)
+			return rowsIngested, fmt.Errorf("cannot parse stream labels %q: %w", stream.Labels, err)
 		}
 		commonFieldsLen := len(fields.fields)
 
@@ -119,14 +122,11 @@ func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useD
 				ts = currentTimestamp
 			}
 
-			var streamFields []logstorage.Field
-			if useDefaultStreamFields {
-				streamFields = fields.fields[:commonFieldsLen]
-			}
-			lmp.AddRow(ts, fields.fields, streamFields)
+			lmp.AddRow(ts, fields.fields)
 		}
+		rowsIngested += len(stream.Entries)
 	}
-	return nil
+	return rowsIngested, nil
 }
 
 func getFields() *fields {

app/vlinsert/loki/loki_protobuf_test.go

@@ -15,10 +15,7 @@ type testLogMessageProcessor struct {
 	pr PushRequest
 }
 
-func (tlp *testLogMessageProcessor) AddRow(timestamp int64, fields, streamFields []logstorage.Field) {
-	if streamFields != nil {
-		panic(fmt.Errorf("unexpected non-nil streamFields: %v", streamFields))
-	}
+func (tlp *testLogMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field) {
 	msg := ""
 	for _, f := range fields {
 		if f.Name == "_msg" {
@@ -53,21 +50,23 @@ func TestParseProtobufRequest_Success(t *testing.T) {
 		t.Helper()
 
 		tlp := &testLogMessageProcessor{}
-		if err := parseJSONRequest([]byte(s), tlp, false); err != nil {
+		n, err := parseJSONRequest([]byte(s), tlp)
+		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if len(tlp.pr.Streams) != len(timestampsExpected) {
-			t.Fatalf("unexpected number of streams; got %d; want %d", len(tlp.pr.Streams), len(timestampsExpected))
+		if n != len(tlp.pr.Streams) {
+			t.Fatalf("unexpected number of streams; got %d; want %d", len(tlp.pr.Streams), n)
 		}
 
 		data := tlp.pr.MarshalProtobuf(nil)
 		encodedData := snappy.Encode(nil, data)
 
 		tlp2 := &insertutils.TestLogMessageProcessor{}
-		if err := parseProtobufRequest(encodedData, tlp2, false); err != nil {
+		n, err = parseProtobufRequest(encodedData, tlp2)
+		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if err := tlp2.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp2.Verify(n, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 	}

app/vlinsert/loki/loki_protobuf_timing_test.go

@@ -31,7 +31,8 @@ func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
 	b.RunParallel(func(pb *testing.PB) {
 		body := getProtobufBody(streams, rows, labels)
 		for pb.Next() {
-			if err := parseProtobufRequest(body, blp, false); err != nil {
+			_, err := parseProtobufRequest(body, blp)
+			if err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}
 		}

app/vlinsert/main.go

@@ -1,7 +1,6 @@
 package vlinsert
 
 import (
-	"fmt"
 	"net/http"
 	"strings"
 
@@ -35,15 +34,9 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	path = strings.TrimPrefix(path, "/insert")
 	path = strings.ReplaceAll(path, "//", "/")
 
-	switch path {
-	case "/jsonline":
+	if path == "/jsonline" {
 		jsonline.RequestHandler(w, r)
 		return true
-	case "/ready":
-		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(200)
-		fmt.Fprintf(w, `{"status":"ok"}`)
-		return true
 	}
 	switch {
 	case strings.HasPrefix(path, "/elasticsearch/"):

app/vlinsert/opentelemetry/opentelemetry.go

@@ -66,15 +66,16 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 		return
 	}
 
-	lmp := cp.NewLogMessageProcessor("opentelelemtry_protobuf")
-	useDefaultStreamFields := len(cp.StreamFields) == 0
-	err = pushProtobufRequest(data, lmp, useDefaultStreamFields)
+	lmp := cp.NewLogMessageProcessor()
+	n, err := pushProtobufRequest(data, lmp)
 	lmp.MustClose()
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse OpenTelemetry protobuf request: %s", err)
 		return
 	}
 
+	rowsIngestedProtobufTotal.Add(n)
+
 	// update requestProtobufDuration only for successfully parsed requests
 	// There is no need in updating requestProtobufDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
@@ -82,19 +83,22 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 }
 
 var (
+	rowsIngestedProtobufTotal = metrics.NewCounter(`vl_rows_ingested_total{type="opentelemetry",format="protobuf"}`)
+
 	requestsProtobufTotal = metrics.NewCounter(`vl_http_requests_total{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
 	errorsTotal           = metrics.NewCounter(`vl_http_errors_total{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
 
 	requestProtobufDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/opentelemetry/v1/logs",format="protobuf"}`)
 )
 
-func pushProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefaultStreamFields bool) error {
+func pushProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor) (int, error) {
 	var req pb.ExportLogsServiceRequest
 	if err := req.UnmarshalProtobuf(data); err != nil {
 		errorsTotal.Inc()
-		return fmt.Errorf("cannot unmarshal request from %d bytes: %w", len(data), err)
+		return 0, fmt.Errorf("cannot unmarshal request from %d bytes: %w", len(data), err)
 	}
 
+	var rowsIngested int
 	var commonFields []logstorage.Field
 	for _, rl := range req.ResourceLogs {
 		attributes := rl.Resource.Attributes
@@ -105,14 +109,16 @@ func pushProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useDe
 		}
 		commonFieldsLen := len(commonFields)
 		for _, sc := range rl.ScopeLogs {
-			commonFields = pushFieldsFromScopeLogs(&sc, commonFields[:commonFieldsLen], lmp, useDefaultStreamFields)
+			var scopeIngested int
+			commonFields, scopeIngested = pushFieldsFromScopeLogs(&sc, commonFields[:commonFieldsLen], lmp)
+			rowsIngested += scopeIngested
 		}
 	}
 
-	return nil
+	return rowsIngested, nil
 }
 
-func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field, lmp insertutils.LogMessageProcessor, useDefaultStreamFields bool) []logstorage.Field {
+func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field, lmp insertutils.LogMessageProcessor) ([]logstorage.Field, int) {
 	fields := commonFields
 	for _, lr := range sc.LogRecords {
 		fields = fields[:len(commonFields)]
@@ -131,11 +137,7 @@ func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field,
 			Value: lr.FormatSeverity(),
 		})
 
-		var streamFields []logstorage.Field
-		if useDefaultStreamFields {
-			streamFields = commonFields
-		}
-		lmp.AddRow(lr.ExtractTimestampNano(), fields, streamFields)
+		lmp.AddRow(lr.ExtractTimestampNano(), fields)
 	}
-	return fields
+	return fields, len(sc.LogRecords)
 }

app/vlinsert/opentelemetry/opentelemetry_test.go

@@ -16,11 +16,12 @@ func TestPushProtoOk(t *testing.T) {
 
 		pData := lr.MarshalProtobuf(nil)
 		tlp := &insertutils.TestLogMessageProcessor{}
-		if err := pushProtobufRequest(pData, tlp, false); err != nil {
+		n, err := pushProtobufRequest(pData, tlp)
+		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(n, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 	}

app/vlinsert/opentelemetry/opentemetry_timing_test.go

@@ -27,7 +27,8 @@ func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
 	b.RunParallel(func(pb *testing.PB) {
 		body := getProtobufBody(streams, rows, labels)
 		for pb.Next() {
-			if err := pushProtobufRequest(body, blp, false); err != nil {
+			_, err := pushProtobufRequest(body, blp)
+			if err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}
 		}

app/vlinsert/syslog/syslog.go

@@ -314,7 +314,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod st
 				}
 				bb.B = bb.B[:n]
 				udpRequestsTotal.Inc()
-				if err := processStream("udp", bb.NewReader(), compressMethod, useLocalTimestamp, cp); err != nil {
+				if err := processStream(bb.NewReader(), compressMethod, useLocalTimestamp, cp); err != nil {
 					logger.Errorf("syslog: cannot process UDP data from %s at %s: %s", remoteAddr, localAddr, err)
 				}
 			}
@@ -354,7 +354,7 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod stri
 		wg.Add(1)
 		go func() {
 			cp := insertutils.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
-			if err := processStream("tcp", c, compressMethod, useLocalTimestamp, cp); err != nil {
+			if err := processStream(c, compressMethod, useLocalTimestamp, cp); err != nil {
 				logger.Errorf("syslog: cannot process TCP data at %q: %s", addr, err)
 			}
 
@@ -369,12 +369,12 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod stri
 }
 
 // processStream parses a stream of syslog messages from r and ingests them into vlstorage.
-func processStream(protocol string, r io.Reader, compressMethod string, useLocalTimestamp bool, cp *insertutils.CommonParams) error {
+func processStream(r io.Reader, compressMethod string, useLocalTimestamp bool, cp *insertutils.CommonParams) error {
 	if err := vlstorage.CanWriteData(); err != nil {
 		return err
 	}
 
-	lmp := cp.NewLogMessageProcessor("syslog_" + protocol)
+	lmp := cp.NewLogMessageProcessor()
 	err := processStreamInternal(r, compressMethod, useLocalTimestamp, lmp)
 	lmp.MustClose()
 
@@ -436,6 +436,7 @@ func processUncompressedStream(r io.Reader, useLocalTimestamp bool, lmp insertut
 			return fmt.Errorf("cannot read line #%d: %s", n, err)
 		}
 		n++
+		rowsIngestedTotal.Inc()
 	}
 	return slr.Error()
 }
@@ -567,7 +568,7 @@ func processLine(line []byte, currentYear int, timezone *time.Location, useLocal
 		ts = nsecs
 	}
 	logstorage.RenameField(p.Fields, msgFields, "_msg")
-	lmp.AddRow(ts, p.Fields, nil)
+	lmp.AddRow(ts, p.Fields)
 	logstorage.PutSyslogParser(p)
 
 	return nil
@@ -576,6 +577,8 @@ func processLine(line []byte, currentYear int, timezone *time.Location, useLocal
 var msgFields = []string{"message"}
 
 var (
+	rowsIngestedTotal = metrics.NewCounter(`vl_rows_ingested_total{type="syslog"}`)
+
 	errorsTotal = metrics.NewCounter(`vl_errors_total{type="syslog"}`)
 
 	udpRequestsTotal = metrics.NewCounter(`vl_udp_reqests_total{type="syslog"}`)

app/vlinsert/syslog/syslog_test.go

@@ -75,7 +75,7 @@ func TestSyslogLineReader_Failure(t *testing.T) {
 }
 
 func TestProcessStreamInternal_Success(t *testing.T) {
-	f := func(data string, currentYear int, timestampsExpected []int64, resultExpected string) {
+	f := func(data string, currentYear, rowsExpected int, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
 		MustInit()
@@ -89,7 +89,7 @@ func TestProcessStreamInternal_Success(t *testing.T) {
 		if err := processStreamInternal(r, "", false, tlp); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+		if err := tlp.Verify(rowsExpected, timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
 		}
 	}
@@ -99,11 +99,12 @@ func TestProcessStreamInternal_Success(t *testing.T) {
 48 <165>Jun  4 12:08:33 abcd systemd[345]: abc defg<123>1 2023-06-03T17:42:12.345Z mymachine.example.com appname 12345 ID47 [exampleSDID@32473 iut="3" eventSource="Application 123 = ] 56" eventID="11211"] This is a test message with structured data.
 `
 	currentYear := 2023
+	rowsExpected := 3
 	timestampsExpected := []int64{1685794113000000000, 1685880513000000000, 1685814132345000000}
 	resultExpected := `{"format":"rfc3164","hostname":"abcd","app_name":"systemd","_msg":"Starting Update the local ESM caches..."}
 {"priority":"165","facility":"20","severity":"5","format":"rfc3164","hostname":"abcd","app_name":"systemd","proc_id":"345","_msg":"abc defg"}
 {"priority":"123","facility":"15","severity":"3","format":"rfc5424","hostname":"mymachine.example.com","app_name":"appname","proc_id":"12345","msg_id":"ID47","exampleSDID@32473.iut":"3","exampleSDID@32473.eventSource":"Application 123 = ] 56","exampleSDID@32473.eventID":"11211","_msg":"This is a test message with structured data."}`
-	f(data, currentYear, timestampsExpected, resultExpected)
+	f(data, currentYear, rowsExpected, timestampsExpected, resultExpected)
 }
 
 func TestProcessStreamInternal_Failure(t *testing.T) {

app/vlogscli/json_prettifier.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"io"
 	"sort"
-	"strings"
 	"sync"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
@@ -235,11 +234,5 @@ func getJSONString(s string) string {
 	if err != nil {
 		panic(fmt.Errorf("unexpected error when marshaling string to JSON: %w", err))
 	}
-	return jsonHTMLReplacer.Replace(string(data))
+	return string(data)
 }
-
-var jsonHTMLReplacer = strings.NewReplacer(
-	`\u003c`, "\u003c",
-	`\u003e`, "\u003e",
-	`\u0026`, "\u0026",
-)

app/vlselect/logsql/facets_response.qtpl

@@ -1,49 +0,0 @@
-{% import (
-	"slices"
-) %}
-
-{% stripspace %}
-
-{% func FacetsResponse(m map[string][]facetEntry) %}
-{
-	{% code
-		sortedKeys := make([]string, 0, len(m))
-		for k := range m {
-			sortedKeys = append(sortedKeys, k)
-		}
-		slices.Sort(sortedKeys)
-	%}
-	"facets":[
-		{% if len(sortedKeys) > 0 %}
-			{%= facetsLine(m, sortedKeys[0]) %}
-			{% for _, k := range sortedKeys[1:] %}
-				,{%= facetsLine(m, k) %}
-			{% endfor %}
-		{% endif %}
-	]
-}
-{% endfunc %}
-
-{% func facetsLine(m map[string][]facetEntry, k string) %}
-{
-	"field_name":{%q= k %},
-	"values":[
-		{% code	fes := m[k] %}
-		{% if len(fes) > 0 %}
-			{%= facetLine(fes[0]) %}
-			{% for _, fe := range fes[1:] %}
-				,{%= facetLine(fe) %}
-			{% endfor %}
-		{% endif %}
-	]
-}
-{% endfunc %}
-
-{% func facetLine(fe facetEntry) %}
-{
-	"field_value":{%q= fe.value %},
-	"hits":{%s= fe.hits %}
-}
-{% endfunc %}
-
-{% endstripspace %}

app/vlselect/logsql/facets_response.qtpl.go

@@ -1,178 +0,0 @@
-// Code generated by qtc from "facets_response.qtpl". DO NOT EDIT.
-// See https://github.com/valyala/quicktemplate for details.
-
-//line app/vlselect/logsql/facets_response.qtpl:1
-package logsql
-
-//line app/vlselect/logsql/facets_response.qtpl:1
-import (
-	"slices"
-)
-
-//line app/vlselect/logsql/facets_response.qtpl:7
-import (
-	qtio422016 "io"
-
-	qt422016 "github.com/valyala/quicktemplate"
-)
-
-//line app/vlselect/logsql/facets_response.qtpl:7
-var (
-	_ = qtio422016.Copy
-	_ = qt422016.AcquireByteBuffer
-)
-
-//line app/vlselect/logsql/facets_response.qtpl:7
-func StreamFacetsResponse(qw422016 *qt422016.Writer, m map[string][]facetEntry) {
-//line app/vlselect/logsql/facets_response.qtpl:7
-	qw422016.N().S(`{`)
-//line app/vlselect/logsql/facets_response.qtpl:10
-	sortedKeys := make([]string, 0, len(m))
-	for k := range m {
-		sortedKeys = append(sortedKeys, k)
-	}
-	slices.Sort(sortedKeys)
-
-//line app/vlselect/logsql/facets_response.qtpl:15
-	qw422016.N().S(`"facets":[`)
-//line app/vlselect/logsql/facets_response.qtpl:17
-	if len(sortedKeys) > 0 {
-//line app/vlselect/logsql/facets_response.qtpl:18
-		streamfacetsLine(qw422016, m, sortedKeys[0])
-//line app/vlselect/logsql/facets_response.qtpl:19
-		for _, k := range sortedKeys[1:] {
-//line app/vlselect/logsql/facets_response.qtpl:19
-			qw422016.N().S(`,`)
-//line app/vlselect/logsql/facets_response.qtpl:20
-			streamfacetsLine(qw422016, m, k)
-//line app/vlselect/logsql/facets_response.qtpl:21
-		}
-//line app/vlselect/logsql/facets_response.qtpl:22
-	}
-//line app/vlselect/logsql/facets_response.qtpl:22
-	qw422016.N().S(`]}`)
-//line app/vlselect/logsql/facets_response.qtpl:25
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:25
-func WriteFacetsResponse(qq422016 qtio422016.Writer, m map[string][]facetEntry) {
-//line app/vlselect/logsql/facets_response.qtpl:25
-	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vlselect/logsql/facets_response.qtpl:25
-	StreamFacetsResponse(qw422016, m)
-//line app/vlselect/logsql/facets_response.qtpl:25
-	qt422016.ReleaseWriter(qw422016)
-//line app/vlselect/logsql/facets_response.qtpl:25
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:25
-func FacetsResponse(m map[string][]facetEntry) string {
-//line app/vlselect/logsql/facets_response.qtpl:25
-	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vlselect/logsql/facets_response.qtpl:25
-	WriteFacetsResponse(qb422016, m)
-//line app/vlselect/logsql/facets_response.qtpl:25
-	qs422016 := string(qb422016.B)
-//line app/vlselect/logsql/facets_response.qtpl:25
-	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vlselect/logsql/facets_response.qtpl:25
-	return qs422016
-//line app/vlselect/logsql/facets_response.qtpl:25
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:27
-func streamfacetsLine(qw422016 *qt422016.Writer, m map[string][]facetEntry, k string) {
-//line app/vlselect/logsql/facets_response.qtpl:27
-	qw422016.N().S(`{"field_name":`)
-//line app/vlselect/logsql/facets_response.qtpl:29
-	qw422016.N().Q(k)
-//line app/vlselect/logsql/facets_response.qtpl:29
-	qw422016.N().S(`,"values":[`)
-//line app/vlselect/logsql/facets_response.qtpl:31
-	fes := m[k]
-
-//line app/vlselect/logsql/facets_response.qtpl:32
-	if len(fes) > 0 {
-//line app/vlselect/logsql/facets_response.qtpl:33
-		streamfacetLine(qw422016, fes[0])
-//line app/vlselect/logsql/facets_response.qtpl:34
-		for _, fe := range fes[1:] {
-//line app/vlselect/logsql/facets_response.qtpl:34
-			qw422016.N().S(`,`)
-//line app/vlselect/logsql/facets_response.qtpl:35
-			streamfacetLine(qw422016, fe)
-//line app/vlselect/logsql/facets_response.qtpl:36
-		}
-//line app/vlselect/logsql/facets_response.qtpl:37
-	}
-//line app/vlselect/logsql/facets_response.qtpl:37
-	qw422016.N().S(`]}`)
-//line app/vlselect/logsql/facets_response.qtpl:40
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:40
-func writefacetsLine(qq422016 qtio422016.Writer, m map[string][]facetEntry, k string) {
-//line app/vlselect/logsql/facets_response.qtpl:40
-	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vlselect/logsql/facets_response.qtpl:40
-	streamfacetsLine(qw422016, m, k)
-//line app/vlselect/logsql/facets_response.qtpl:40
-	qt422016.ReleaseWriter(qw422016)
-//line app/vlselect/logsql/facets_response.qtpl:40
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:40
-func facetsLine(m map[string][]facetEntry, k string) string {
-//line app/vlselect/logsql/facets_response.qtpl:40
-	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vlselect/logsql/facets_response.qtpl:40
-	writefacetsLine(qb422016, m, k)
-//line app/vlselect/logsql/facets_response.qtpl:40
-	qs422016 := string(qb422016.B)
-//line app/vlselect/logsql/facets_response.qtpl:40
-	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vlselect/logsql/facets_response.qtpl:40
-	return qs422016
-//line app/vlselect/logsql/facets_response.qtpl:40
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:42
-func streamfacetLine(qw422016 *qt422016.Writer, fe facetEntry) {
-//line app/vlselect/logsql/facets_response.qtpl:42
-	qw422016.N().S(`{"field_value":`)
-//line app/vlselect/logsql/facets_response.qtpl:44
-	qw422016.N().Q(fe.value)
-//line app/vlselect/logsql/facets_response.qtpl:44
-	qw422016.N().S(`,"hits":`)
-//line app/vlselect/logsql/facets_response.qtpl:45
-	qw422016.N().S(fe.hits)
-//line app/vlselect/logsql/facets_response.qtpl:45
-	qw422016.N().S(`}`)
-//line app/vlselect/logsql/facets_response.qtpl:47
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:47
-func writefacetLine(qq422016 qtio422016.Writer, fe facetEntry) {
-//line app/vlselect/logsql/facets_response.qtpl:47
-	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vlselect/logsql/facets_response.qtpl:47
-	streamfacetLine(qw422016, fe)
-//line app/vlselect/logsql/facets_response.qtpl:47
-	qt422016.ReleaseWriter(qw422016)
-//line app/vlselect/logsql/facets_response.qtpl:47
-}
-
-//line app/vlselect/logsql/facets_response.qtpl:47
-func facetLine(fe facetEntry) string {
-//line app/vlselect/logsql/facets_response.qtpl:47
-	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vlselect/logsql/facets_response.qtpl:47
-	writefacetLine(qb422016, fe)
-//line app/vlselect/logsql/facets_response.qtpl:47
-	qs422016 := string(qb422016.B)
-//line app/vlselect/logsql/facets_response.qtpl:47
-	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vlselect/logsql/facets_response.qtpl:47
-	return qs422016
-//line app/vlselect/logsql/facets_response.qtpl:47
-}

app/vlselect/logsql/logsql.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"math"
 	"net/http"
-	"regexp"
 	"slices"
 	"sort"
 	"strconv"
@@ -14,7 +13,6 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/metrics"
-	"github.com/valyala/fastjson"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@@ -25,82 +23,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
-// ProcessFacetsRequest handles /select/logsql/facets request.
-//
-// See https://docs.victoriametrics.com/victorialogs/querying/#querying-facets
-func ProcessFacetsRequest(ctx context.Context, w http.ResponseWriter, r *http.Request) {
-	q, tenantIDs, err := parseCommonArgs(r)
-	if err != nil {
-		httpserver.Errorf(w, r, "%s", err)
-		return
-	}
-
-	limit, err := httputils.GetInt(r, "limit")
-	if err != nil {
-		httpserver.Errorf(w, r, "%s", err)
-		return
-	}
-	maxValuesPerField, err := httputils.GetInt(r, "max_values_per_field")
-	if err != nil {
-		httpserver.Errorf(w, r, "%s", err)
-		return
-	}
-	maxValueLen, err := httputils.GetInt(r, "max_value_len")
-	if err != nil {
-		httpserver.Errorf(w, r, "%s", err)
-		return
-	}
-	keepConstFields := httputils.GetBool(r, "keep_const_fields")
-
-	q.DropAllPipes()
-	q.AddFacetsPipe(limit, maxValuesPerField, maxValueLen, keepConstFields)
-
-	var mLock sync.Mutex
-	m := make(map[string][]facetEntry)
-	writeBlock := func(_ uint, _ []int64, columns []logstorage.BlockColumn) {
-		if len(columns) == 0 || len(columns[0].Values) == 0 {
-			return
-		}
-		if len(columns) != 3 {
-			logger.Panicf("BUG: expecting 3 columns; got %d columns", len(columns))
-		}
-
-		fieldNames := columns[0].Values
-		fieldValues := columns[1].Values
-		hits := columns[2].Values
-
-		bb := blockResultPool.Get()
-		for i := range fieldNames {
-			fieldName := strings.Clone(fieldNames[i])
-			fieldValue := strings.Clone(fieldValues[i])
-			hitsStr := strings.Clone(hits[i])
-
-			mLock.Lock()
-			m[fieldName] = append(m[fieldName], facetEntry{
-				value: fieldValue,
-				hits:  hitsStr,
-			})
-			mLock.Unlock()
-		}
-		blockResultPool.Put(bb)
-	}
-
-	// Execute the query
-	if err := vlstorage.RunQuery(ctx, tenantIDs, q, writeBlock); err != nil {
-		httpserver.Errorf(w, r, "cannot execute query [%s]: %s", q, err)
-		return
-	}
-
-	// Write response
-	w.Header().Set("Content-Type", "application/json")
-	WriteFacetsResponse(w, m)
-}
-
-type facetEntry struct {
-	value string
-	hits  string
-}
-
 // ProcessHitsRequest handles /select/logsql/hits request.
 //
 // See https://docs.victoriametrics.com/victorialogs/querying/#querying-hits-stats
@@ -1095,20 +1017,18 @@ func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID,
 	}
 
 	// Parse optional extra_filters
-	extraFiltersStr := r.FormValue("extra_filters")
-	extraFilters, err := parseExtraFilters(extraFiltersStr)
+	extraFilters, err := getExtraFilters(r, "extra_filters")
 	if err != nil {
 		return nil, nil, err
 	}
 	q.AddExtraFilters(extraFilters)
 
 	// Parse optional extra_stream_filters
-	extraStreamFiltersStr := r.FormValue("extra_stream_filters")
-	extraStreamFilters, err := parseExtraStreamFilters(extraStreamFiltersStr)
+	extraStreamFilters, err := getExtraFilters(r, "extra_stream_filters")
 	if err != nil {
 		return nil, nil, err
 	}
-	q.AddExtraFilters(extraStreamFilters)
+	q.AddExtraStreamFilters(extraStreamFilters)
 
 	return q, tenantIDs, nil
 }
@@ -1126,114 +1046,15 @@ func getTimeNsec(r *http.Request, argName string) (int64, bool, error) {
 	return nsecs, true, nil
 }
 
-func parseExtraFilters(s string) (*logstorage.Filter, error) {
+func getExtraFilters(r *http.Request, argName string) ([]logstorage.Field, error) {
+	s := r.FormValue(argName)
 	if s == "" {
 		return nil, nil
 	}
-	if !strings.HasPrefix(s, `{"`) {
-		return logstorage.ParseFilter(s)
-	}
 
-	// Extra filters in the form {"field":"value",...}.
-	kvs, err := parseExtraFiltersJSON(s)
-	if err != nil {
-		return nil, err
+	var p logstorage.JSONParser
+	if err := p.ParseLogMessage([]byte(s)); err != nil {
+		return nil, fmt.Errorf("cannot parse %s: %w", argName, err)
 	}
-
-	filters := make([]string, len(kvs))
-	for i, kv := range kvs {
-		if len(kv.values) == 1 {
-			filters[i] = fmt.Sprintf("%q:=%q", kv.key, kv.values[0])
-		} else {
-			orValues := make([]string, len(kv.values))
-			for j, v := range kv.values {
-				orValues[j] = fmt.Sprintf("%q", v)
-			}
-			filters[i] = fmt.Sprintf("%q:in(%s)", kv.key, strings.Join(orValues, ","))
-		}
-	}
-	s = strings.Join(filters, " ")
-	return logstorage.ParseFilter(s)
-}
-
-func parseExtraStreamFilters(s string) (*logstorage.Filter, error) {
-	if s == "" {
-		return nil, nil
-	}
-	if !strings.HasPrefix(s, `{"`) {
-		return logstorage.ParseFilter(s)
-	}
-
-	// Extra stream filters in the form {"field":"value",...}.
-	kvs, err := parseExtraFiltersJSON(s)
-	if err != nil {
-		return nil, err
-	}
-
-	filters := make([]string, len(kvs))
-	for i, kv := range kvs {
-		if len(kv.values) == 1 {
-			filters[i] = fmt.Sprintf("%q=%q", kv.key, kv.values[0])
-		} else {
-			orValues := make([]string, len(kv.values))
-			for j, v := range kv.values {
-				orValues[j] = regexp.QuoteMeta(v)
-			}
-			filters[i] = fmt.Sprintf("%q=~%q", kv.key, strings.Join(orValues, "|"))
-		}
-	}
-	s = "{" + strings.Join(filters, ",") + "}"
-	return logstorage.ParseFilter(s)
-}
-
-type extraFilter struct {
-	key    string
-	values []string
-}
-
-func parseExtraFiltersJSON(s string) ([]extraFilter, error) {
-	v, err := fastjson.Parse(s)
-	if err != nil {
-		return nil, err
-	}
-	o := v.GetObject()
-
-	var errOuter error
-	var filters []extraFilter
-	o.Visit(func(k []byte, v *fastjson.Value) {
-		if errOuter != nil {
-			return
-		}
-		switch v.Type() {
-		case fastjson.TypeString:
-			filters = append(filters, extraFilter{
-				key:    string(k),
-				values: []string{string(v.GetStringBytes())},
-			})
-		case fastjson.TypeArray:
-			a := v.GetArray()
-			if len(a) == 0 {
-				return
-			}
-			orValues := make([]string, len(a))
-			for i, av := range a {
-				ov, err := av.StringBytes()
-				if err != nil {
-					errOuter = fmt.Errorf("cannot obtain string item at the array for key %q; item: %s", k, av)
-					return
-				}
-				orValues[i] = string(ov)
-			}
-			filters = append(filters, extraFilter{
-				key:    string(k),
-				values: orValues,
-			})
-		default:
-			errOuter = fmt.Errorf("unexpected type of value for key %q: %s; value: %s", k, v.Type(), v)
-		}
-	})
-	if errOuter != nil {
-		return nil, errOuter
-	}
-	return filters, nil
+	return p.Fields, nil
 }

app/vlselect/logsql/logsql_test.go

@@ -1,103 +0,0 @@
-package logsql
-
-import (
-	"testing"
-)
-
-func TestParseExtraFilters_Success(t *testing.T) {
-	f := func(s, resultExpected string) {
-		t.Helper()
-
-		f, err := parseExtraFilters(s)
-		if err != nil {
-			t.Fatalf("unexpected error in parseExtraFilters: %s", err)
-		}
-		result := f.String()
-		if result != resultExpected {
-			t.Fatalf("unexpected result\ngot\n%s\nwant\n%s", result, resultExpected)
-		}
-	}
-
-	f("", "")
-
-	// JSON string
-	f(`{"foo":"bar"}`, `foo:=bar`)
-	f(`{"foo":["bar","baz"]}`, `foo:in(bar,baz)`)
-	f(`{"z":"=b ","c":["d","e,"],"a":[],"_msg":"x"}`, `z:="=b " c:in(d,"e,") =x`)
-
-	// LogsQL filter
-	f(`foobar`, `foobar`)
-	f(`foo:bar`, `foo:bar`)
-	f(`foo:(bar or baz) error _time:5m {"foo"=bar,baz="z"}`, `(foo:bar or foo:baz) error _time:5m {foo="bar",baz="z"}`)
-}
-
-func TestParseExtraFilters_Failure(t *testing.T) {
-	f := func(s string) {
-		t.Helper()
-
-		_, err := parseExtraFilters(s)
-		if err == nil {
-			t.Fatalf("expecting non-nil error")
-		}
-	}
-
-	// Invalid JSON
-	f(`{"foo"}`)
-	f(`[1,2]`)
-	f(`{"foo":[1]}`)
-
-	// Invliad LogsQL filter
-	f(`foo:(bar`)
-
-	// excess pipe
-	f(`foo | count()`)
-}
-
-func TestParseExtraStreamFilters_Success(t *testing.T) {
-	f := func(s, resultExpected string) {
-		t.Helper()
-
-		f, err := parseExtraStreamFilters(s)
-		if err != nil {
-			t.Fatalf("unexpected error in parseExtraStreamFilters: %s", err)
-		}
-		result := f.String()
-		if result != resultExpected {
-			t.Fatalf("unexpected result;\ngot\n%s\nwant\n%s", result, resultExpected)
-		}
-	}
-
-	f("", "")
-
-	// JSON string
-	f(`{"foo":"bar"}`, `{foo="bar"}`)
-	f(`{"foo":["bar","baz"]}`, `{foo=~"bar|baz"}`)
-	f(`{"z":"b","c":["d","e|\""],"a":[],"_msg":"x"}`, `{z="b",c=~"d|e\\|\"",_msg="x"}`)
-
-	// LogsQL filter
-	f(`foobar`, `foobar`)
-	f(`foo:bar`, `foo:bar`)
-	f(`foo:(bar or baz) error _time:5m {"foo"=bar,baz="z"}`, `(foo:bar or foo:baz) error _time:5m {foo="bar",baz="z"}`)
-}
-
-func TestParseExtraStreamFilters_Failure(t *testing.T) {
-	f := func(s string) {
-		t.Helper()
-
-		_, err := parseExtraStreamFilters(s)
-		if err == nil {
-			t.Fatalf("expecting non-nil error")
-		}
-	}
-
-	// Invalid JSON
-	f(`{"foo"}`)
-	f(`[1,2]`)
-	f(`{"foo":[1]}`)
-
-	// Invliad LogsQL filter
-	f(`foo:(bar`)
-
-	// excess pipe
-	f(`foo | count()`)
-}

app/vlselect/main.go

@@ -177,10 +177,6 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 func processSelectRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, path string) bool {
 	httpserver.EnableCORS(w, r)
 	switch path {
-	case "/select/logsql/facets":
-		logsqlFacetsRequests.Inc()
-		logsql.ProcessFacetsRequest(ctx, w, r)
-		return true
 	case "/select/logsql/field_names":
 		logsqlFieldNamesRequests.Inc()
 		logsql.ProcessFieldNamesRequest(ctx, w, r)
@@ -240,7 +236,6 @@ func getMaxQueryDuration(r *http.Request) time.Duration {
 }
 
 var (
-	logsqlFacetsRequests            = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/facets"}`)
 	logsqlFieldNamesRequests        = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_names"}`)
 	logsqlFieldValuesRequests       = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_values"}`)
 	logsqlHitsRequests              = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/hits"}`)

app/vlselect/vmui/asset-manifest.json

@@ -1,13 +1,13 @@
 {
   "files": {
-    "main.css": "./static/css/main.fa83344e.css",
-    "main.js": "./static/js/main.8ad2bc1f.js",
+    "main.css": "./static/css/main.faf86aa5.css",
+    "main.js": "./static/js/main.b204330a.js",
     "static/js/685.f772060c.chunk.js": "./static/js/685.f772060c.chunk.js",
     "static/media/MetricsQL.md": "./static/media/MetricsQL.a00044c91d9781cf8557.md",
     "index.html": "./index.html"
   },
   "entrypoints": [
-    "static/css/main.fa83344e.css",
-    "static/js/main.8ad2bc1f.js"
+    "static/css/main.faf86aa5.css",
+    "static/js/main.b204330a.js"
   ]
 }

app/vlselect/vmui/index.html

@@ -1 +1 @@
-<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="./favicon.svg"/><link rel="apple-touch-icon" href="./favicon.svg"/><link rel="mask-icon" href="./favicon.svg" color="#000000"><meta name="viewport" content="width=device-width,initial-scale=1,maximum-scale=5"/><meta name="theme-color" content="#000000"/><meta name="description" content="Explore your log data with VictoriaLogs UI"/><link rel="manifest" href="./manifest.json"/><title>UI for VictoriaLogs</title><meta name="twitter:card" content="summary"><meta name="twitter:title" content="UI for VictoriaLogs"><meta name="twitter:site" content="@https://victoriametrics.com/products/victorialogs/"><meta name="twitter:description" content="Explore your log data with VictoriaLogs UI"><meta name="twitter:image" content="./preview.jpg"><meta property="og:type" content="website"><meta property="og:title" content="UI for VictoriaLogs"><meta property="og:url" content="https://victoriametrics.com/products/victorialogs/"><meta property="og:description" content="Explore your log data with VictoriaLogs UI"><script defer="defer" src="./static/js/main.8ad2bc1f.js"></script><link href="./static/css/main.fa83344e.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>
+<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="./favicon.svg"/><link rel="apple-touch-icon" href="./favicon.svg"/><link rel="mask-icon" href="./favicon.svg" color="#000000"><meta name="viewport" content="width=device-width,initial-scale=1,maximum-scale=5"/><meta name="theme-color" content="#000000"/><meta name="description" content="Explore your log data with VictoriaLogs UI"/><link rel="manifest" href="./manifest.json"/><title>UI for VictoriaLogs</title><meta name="twitter:card" content="summary"><meta name="twitter:title" content="UI for VictoriaLogs"><meta name="twitter:site" content="@https://victoriametrics.com/products/victorialogs/"><meta name="twitter:description" content="Explore your log data with VictoriaLogs UI"><meta name="twitter:image" content="./preview.jpg"><meta property="og:type" content="website"><meta property="og:title" content="UI for VictoriaLogs"><meta property="og:url" content="https://victoriametrics.com/products/victorialogs/"><meta property="og:description" content="Explore your log data with VictoriaLogs UI"><script defer="defer" src="./static/js/main.b204330a.js"></script><link href="./static/css/main.faf86aa5.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>

app/vmagent/main.go

@@ -30,7 +30,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -46,7 +45,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 )
 
 var (
@@ -79,9 +77,6 @@ var (
 	dryRun        = flag.Bool("dryRun", false, "Whether to check config files without running vmagent. The following files are checked: "+
 		"-promscrape.config, -remoteWrite.relabelConfig, -remoteWrite.urlRelabelConfig, -remoteWrite.streamAggr.config . "+
 		"Unknown config entries aren't allowed in -promscrape.config by default. This can be changed by passing -promscrape.config.strictParse=false command-line flag")
-	maxLabelsPerTimeseries = flag.Int("maxLabelsPerTimeseries", 0, "The maximum number of labels per time series to be accepted. Series with superfluous labels are ignored. In this case the vm_rows_ignored_total{reason=\"too_many_labels\"} metric at /metrics page is incremented")
-	maxLabelNameLen        = flag.Int("maxLabelNameLen", 0, "The maximum length of label names in the accepted time series. Series with longer label name are ignored. In this case the vm_rows_ignored_total{reason=\"too_long_label_name\"} metric at /metrics page is incremented")
-	maxLabelValueLen       = flag.Int("maxLabelValueLen", 0, "The maximum length of label values in the accepted time series. Series with longer label value are ignored. In this case the vm_rows_ignored_total{reason=\"too_long_label_value\"} metric at /metrics page is incremented")
 )
 
 var (
@@ -98,15 +93,6 @@ var (
 )
 
 func main() {
-	// vmagent is optimized for reduced memory allocations,
-	// so it can run with the reduced GOGC in order to reduce the used memory,
-	// while keeping CPU usage spent in GC at low levels.
-	//
-	// Some workloads may need increased GOGC values. Then such values can be set via GOGC environment variable.
-	// It is recommended increasing GOGC if go_memstats_gc_cpu_fraction metric exposed at /metrics page
-	// exceeds 0.05 for extended periods of time.
-	cgroup.SetGOGC(30)
-
 	// Write flags and help message to stdout, since it is easier to grep or pipe.
 	flag.CommandLine.SetOutput(os.Stdout)
 	flag.Usage = usage
@@ -114,7 +100,6 @@ func main() {
 	remotewrite.InitSecretFlags()
 	buildinfo.Init()
 	logger.Init()
-	timeserieslimits.Init(*maxLabelsPerTimeseries, *maxLabelNameLen, *maxLabelValueLen)
 
 	if promscrape.IsDryRun() {
 		if err := promscrape.CheckConfig(); err != nil {
@@ -513,7 +498,7 @@ func processMultitenantRequest(w http.ResponseWriter, r *http.Request, path stri
 		httpserver.Errorf(w, r, `unsupported multitenant prefix: %q; expected "insert"`, p.Prefix)
 		return true
 	}
-	at, err := auth.NewTokenPossibleMultitenant(p.AuthToken)
+	at, err := auth.NewToken(p.AuthToken)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain auth token: %s", err)
 		return true
@@ -525,13 +510,7 @@ func processMultitenantRequest(w http.ResponseWriter, r *http.Request, path stri
 			httpserver.Errorf(w, r, "%s", err)
 			return true
 		}
-		statusCode := http.StatusNoContent
-		if strings.HasPrefix(p.Suffix, "prometheus/api/v1/import/prometheus/metrics/job/") {
-			// Return 200 status code for pushgateway requests.
-			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3636
-			statusCode = http.StatusOK
-		}
-		w.WriteHeader(statusCode)
+		w.WriteHeader(http.StatusNoContent)
 		return true
 	}
 	if strings.HasPrefix(p.Suffix, "datadog/") {

app/vmagent/remotewrite/remotewrite.go

@@ -7,10 +7,13 @@ import (
 	"net/url"
 	"path/filepath"
 	"slices"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bloomfilter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@@ -18,7 +21,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
@@ -28,7 +30,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/cespare/xxhash/v2"
 )
@@ -98,6 +99,9 @@ var (
 	// rwctxsGlobal contains statically populated entries when -remoteWrite.url is specified.
 	rwctxsGlobal []*remoteWriteCtx
 
+	// Data without tenant id is written to defaultAuthToken if -enableMultitenantHandlers is specified.
+	defaultAuthToken = &auth.Token{}
+
 	// ErrQueueFullHTTPRetry must be returned when TryPush() returns false.
 	ErrQueueFullHTTPRetry = &httpserver.ErrorWithStatusCode{
 		Err: fmt.Errorf("remote storage systems cannot keep up with the data ingestion rate; retry the request later " +
@@ -205,7 +209,7 @@ func Init() {
 
 	initStreamAggrConfigGlobal()
 
-	rwctxsGlobal = newRemoteWriteCtxs(*remoteWriteURLs)
+	rwctxsGlobal = newRemoteWriteCtxs(nil, *remoteWriteURLs)
 
 	disableOnDiskQueues := []bool(*disableOnDiskQueue)
 	disableOnDiskQueueAny = slices.Contains(disableOnDiskQueues, true)
@@ -290,7 +294,7 @@ var (
 	relabelConfigTimestamp    = metrics.NewCounter(`vmagent_relabel_config_last_reload_success_timestamp_seconds`)
 )
 
-func newRemoteWriteCtxs(urls []string) []*remoteWriteCtx {
+func newRemoteWriteCtxs(at *auth.Token, urls []string) []*remoteWriteCtx {
 	if len(urls) == 0 {
 		logger.Panicf("BUG: urls must be non-empty")
 	}
@@ -312,6 +316,11 @@ func newRemoteWriteCtxs(urls []string) []*remoteWriteCtx {
 			logger.Fatalf("invalid -remoteWrite.url=%q: %s", remoteWriteURL, err)
 		}
 		sanitizedURL := fmt.Sprintf("%d:secret-url", i+1)
+		if at != nil {
+			// Construct full remote_write url for the given tenant according to https://docs.victoriametrics.com/cluster-victoriametrics/#url-format
+			remoteWriteURL.Path = fmt.Sprintf("%s/insert/%d:%d/prometheus/api/v1/write", remoteWriteURL.Path, at.AccountID, at.ProjectID)
+			sanitizedURL = fmt.Sprintf("%s:%d:%d", sanitizedURL, at.AccountID, at.ProjectID)
+		}
 		if *showRemoteWriteURL {
 			sanitizedURL = fmt.Sprintf("%d:%s", i+1, remoteWriteURL)
 		}
@@ -402,6 +411,11 @@ func TryPush(at *auth.Token, wr *prompbmarshal.WriteRequest) bool {
 func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnFailure bool) bool {
 	tss := wr.Timeseries
 
+	if at == nil && MultitenancyEnabled() {
+		// Write data to default tenant if at isn't set when multitenancy is enabled.
+		at = defaultAuthToken
+	}
+
 	var tenantRctx *relabelCtx
 	if at != nil {
 		// Convert at to (vm_account_id, vm_project_id) labels.
@@ -471,15 +485,6 @@ func tryPush(at *auth.Token, wr *prompbmarshal.WriteRequest, forceDropSamplesOnF
 			rowsCountAfterRelabel := getRowsCount(tssBlock)
 			rowsDroppedByGlobalRelabel.Add(rowsCountBeforeRelabel - rowsCountAfterRelabel)
 		}
-		if timeserieslimits.Enabled() {
-			tmpBlock := tssBlock[:0]
-			for _, ts := range tssBlock {
-				if !timeserieslimits.IsExceeding(ts.Labels) {
-					tmpBlock = append(tmpBlock, ts)
-				}
-			}
-			tssBlock = tmpBlock
-		}
 		sortLabelsIfNeeded(tssBlock)
 		tssBlock = limitSeriesCardinality(tssBlock)
 		if sas.IsEnabled() {
@@ -724,14 +729,29 @@ func logSkippedSeries(labels []prompbmarshal.Label, flagName string, flagValue i
 	select {
 	case <-logSkippedSeriesTicker.C:
 		// Do not use logger.WithThrottler() here, since this will increase CPU usage
-		// because every call to logSkippedSeries will result to a call to prompbmarshal.LabelsToString.
-		logger.Warnf("skip series %s because %s=%d reached", prompbmarshal.LabelsToString(labels), flagName, flagValue)
+		// because every call to logSkippedSeries will result to a call to labelsToString.
+		logger.Warnf("skip series %s because %s=%d reached", labelsToString(labels), flagName, flagValue)
 	default:
 	}
 }
 
 var logSkippedSeriesTicker = time.NewTicker(5 * time.Second)
 
+func labelsToString(labels []prompbmarshal.Label) string {
+	var b []byte
+	b = append(b, '{')
+	for i, label := range labels {
+		b = append(b, label.Name...)
+		b = append(b, '=')
+		b = strconv.AppendQuote(b, label.Value)
+		if i+1 < len(labels) {
+			b = append(b, ',')
+		}
+	}
+	b = append(b, '}')
+	return string(b)
+}
+
 var (
 	globalRowsPushedBeforeRelabel = metrics.NewCounter("vmagent_remotewrite_global_rows_pushed_before_relabel_total")
 	rowsDroppedByGlobalRelabel    = metrics.NewCounter("vmagent_remotewrite_global_relabel_metrics_dropped_total")

app/vmalert-tool/main.go

@@ -51,14 +51,9 @@ Examples:
 						Usage:    `Optional external URL to template in rule's labels or annotations.`,
 						Required: false,
 					},
-					&cli.StringFlag{
-						Name:     "loggerLevel",
-						Usage:    `Minimum level of errors to log. Possible values: INFO, WARN, ERROR, FATAL, PANIC (default "ERROR").`,
-						Required: false,
-					},
 				},
 				Action: func(c *cli.Context) error {
-					if failed := unittest.UnitTest(c.StringSlice("files"), c.Bool("disableAlertgroupLabel"), c.StringSlice("external.label"), c.String("external.url"), c.String("loggerLevel")); failed {
+					if failed := unittest.UnitTest(c.StringSlice("files"), c.Bool("disableAlertgroupLabel"), c.StringSlice("external.label"), c.String("external.url")); failed {
 						return fmt.Errorf("unittest failed")
 					}
 					return nil

app/vmalert-tool/unittest/recording.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 	"github.com/VictoriaMetrics/metricsql"
 )
@@ -49,7 +48,7 @@ Outer:
 		}
 		var expSamples []parsedSample
 		for _, s := range mt.ExpSamples {
-			expLb := []prompbmarshal.Label{}
+			expLb := datasource.Labels{}
 			if s.Labels != "" {
 				metricsqlExpr, err := metricsql.Parse(s.Labels)
 				if err != nil {
@@ -65,7 +64,7 @@ Outer:
 				}
 				if len(metricsqlMetricExpr.LabelFilterss) > 0 {
 					for _, l := range metricsqlMetricExpr.LabelFilterss[0] {
-						expLb = append(expLb, prompbmarshal.Label{
+						expLb = append(expLb, datasource.Label{
 							Name:  l.Label,
 							Value: l.Value,
 						})

app/vmalert-tool/unittest/testdata/failed-test-with-missing-rulefile.yaml

@@ -1,4 +0,0 @@
-rule_files:
-  - non-existing-file.yaml
-
-tests: []

app/vmalert-tool/unittest/unittest.go

@@ -5,7 +5,6 @@ import (
 	"flag"
 	"fmt"
 	"net/http"
-	"net/url"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -47,24 +46,17 @@ var (
 	testRemoteWritePath   = "http://127.0.0.1" + httpListenAddr
 	testHealthHTTPPath    = "http://127.0.0.1" + httpListenAddr + "/health"
 
-	testLogLevel           = "ERROR"
 	disableAlertgroupLabel bool
 )
 
 const (
 	testStoragePath = "vmalert-unittest"
+	testLogLevel    = "ERROR"
 )
 
 // UnitTest runs unittest for files
-func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, externalURL, logLevel string) bool {
-	if logLevel != "" {
-		testLogLevel = logLevel
-	}
-	eu, err := url.Parse(externalURL)
-	if err != nil {
-		logger.Fatalf("failed to parse external URL: %w", err)
-	}
-	if err := templates.Load([]string{}, *eu); err != nil {
+func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, externalURL string) bool {
+	if err := templates.Load([]string{}, true); err != nil {
 		logger.Fatalf("failed to load template: %v", err)
 	}
 	storagePath = filepath.Join(os.TempDir(), testStoragePath)
@@ -82,7 +74,8 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 		logger.Fatalf("failed to load test files %q: %v", files, err)
 	}
 	if len(testfiles) == 0 {
-		logger.Fatalf("no test file found")
+		fmt.Println("no test file found")
+		return false
 	}
 
 	labels := make(map[string]string)
@@ -104,8 +97,8 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	var failed bool
 	for fileName, file := range testfiles {
 		if err := ruleUnitTest(fileName, file, labels); err != nil {
-			fmt.Println("FAILED")
-			fmt.Printf("failed to run unit test for file %q: \n%v", fileName, err)
+			fmt.Println("  FAILED")
+			fmt.Printf("\nfailed to run unit test for file %q: \n%v", fileName, err)
 			failed = true
 		} else {
 			fmt.Println("  SUCCESS")
@@ -116,7 +109,7 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 }
 
 func ruleUnitTest(filename string, content []byte, externalLabels map[string]string) []error {
-	fmt.Println("\n\nUnit Testing: ", filename)
+	fmt.Println("\nUnit Testing: ", filename)
 	var unitTestInp unitTestFile
 	if err := yaml.UnmarshalStrict(content, &unitTestInp); err != nil {
 		return []error{fmt.Errorf("failed to unmarshal file: %w", err)}
@@ -146,9 +139,6 @@ func ruleUnitTest(filename string, content []byte, externalLabels map[string]str
 	if err != nil {
 		return []error{fmt.Errorf("failed to parse `rule_files`: %w", err)}
 	}
-	if len(testGroups) == 0 {
-		return []error{fmt.Errorf("found no rule group in %v", unitTestInp.RuleFiles)}
-	}
 
 	var errs []error
 	for _, t := range unitTestInp.Tests {

app/vmalert-tool/unittest/unittest_test.go

@@ -1,21 +1,30 @@
 package unittest
 
 import (
+	"os"
 	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
 )
 
+func TestMain(m *testing.M) {
+	if err := templates.Load([]string{}, true); err != nil {
+		os.Exit(1)
+	}
+	os.Exit(m.Run())
+}
+
 func TestUnitTest_Failure(t *testing.T) {
 	f := func(files []string) {
 		t.Helper()
 
-		failed := UnitTest(files, false, nil, "", "")
+		failed := UnitTest(files, false, nil, "")
 		if !failed {
 			t.Fatalf("expecting failed test")
 		}
 	}
 
-	f([]string{"./testdata/failed-test-with-missing-rulefile.yaml"})
-
+	// failing test
 	f([]string{"./testdata/failed-test.yaml"})
 }
 
@@ -23,7 +32,7 @@ func TestUnitTest_Success(t *testing.T) {
 	f := func(disableGroupLabel bool, files []string, externalLabels []string, externalURL string) {
 		t.Helper()
 
-		failed := UnitTest(files, disableGroupLabel, externalLabels, externalURL, "")
+		failed := UnitTest(files, disableGroupLabel, externalLabels, externalURL)
 		if failed {
 			t.Fatalf("unexpected failed test")
 		}

app/vmalert/config/config_test.go

@@ -16,7 +16,7 @@ import (
 )
 
 func TestMain(m *testing.M) {
-	if err := templates.Load([]string{"testdata/templates/*good.tmpl"}, url.URL{}); err != nil {
+	if err := templates.Load([]string{"testdata/templates/*good.tmpl"}, true); err != nil {
 		os.Exit(1)
 	}
 	os.Exit(m.Run())

app/vmalert/config/testdata/dir/rules0-bad.rules

@@ -7,7 +7,7 @@ groups:
         labels:
           label: bar
         annotations:
-          summary: "{{ }}"
+          summary: "{{ $value }"
           description: "{{$labels}}"
       - alert: UnkownAnnotationsFunction
         for: 5m

app/vmalert/datasource/client_graphite.go

@@ -46,8 +46,8 @@ const (
 	graphitePrefix = "/graphite"
 )
 
-func (c *Client) setGraphiteReqParams(r *http.Request, query string) {
-	if c.appendTypePrefix {
+func (s *Client) setGraphiteReqParams(r *http.Request, query string) {
+	if s.appendTypePrefix {
 		r.URL.Path += graphitePrefix
 	}
 	r.URL.Path += graphitePath
@@ -58,7 +58,7 @@ func (c *Client) setGraphiteReqParams(r *http.Request, query string) {
 	q.Set("target", query)
 	q.Set("until", "now")
 
-	for k, vs := range c.extraParams {
+	for k, vs := range s.extraParams {
 		if q.Has(k) { // extraParams are prior to params in URL
 			q.Del(k)
 		}

app/vmalert/datasource/client_prom.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 
 	"github.com/valyala/fastjson"
 )
@@ -17,8 +16,7 @@ import (
 var (
 	disablePathAppend = flag.Bool("remoteRead.disablePathAppend", false, "Whether to disable automatic appending of '/api/v1/query' or '/select/logsql/stats_query' path "+
 		"to the configured -datasource.url and -remoteRead.url")
-	disableStepParam = flag.Bool("datasource.disableStepParam", false, "Whether to disable adding 'step' param in instant queries to the configured -datasource.url and -remoteRead.url. "+
-		"Only valid for prometheus datasource. "+
+	disableStepParam = flag.Bool("datasource.disableStepParam", false, "Whether to disable adding 'step' param to the issued instant queries. "+
 		"This might be useful when using vmalert with datasources that do not support 'step' param for instant queries, like Google Managed Prometheus. "+
 		"It is not recommended to enable this flag if you use vmalert with VictoriaMetrics.")
 )
@@ -83,14 +81,14 @@ func (pi *promInstant) Unmarshal(b []byte) error {
 		labels := metric.GetObject()
 
 		r := &pi.ms[i]
-		r.Labels = make([]prompbmarshal.Label, 0, labels.Len())
+		r.Labels = make([]Label, 0, labels.Len())
 		labels.Visit(func(key []byte, v *fastjson.Value) {
 			lv, errLocal := v.StringBytes()
 			if errLocal != nil {
 				err = fmt.Errorf("error when parsing label value %q: %s", v, errLocal)
 				return
 			}
-			r.Labels = append(r.Labels, prompbmarshal.Label{
+			r.Labels = append(r.Labels, Label{
 				Name:  string(key),
 				Value: string(lv),
 			})
@@ -220,8 +218,8 @@ func parsePrometheusResponse(req *http.Request, resp *http.Response) (res Result
 	return res, nil
 }
 
-func (c *Client) setPrometheusInstantReqParams(r *http.Request, query string, timestamp time.Time) {
-	if c.appendTypePrefix {
+func (s *Client) setPrometheusInstantReqParams(r *http.Request, query string, timestamp time.Time) {
+	if s.appendTypePrefix {
 		r.URL.Path += "/prometheus"
 	}
 	if !*disablePathAppend {
@@ -229,22 +227,22 @@ func (c *Client) setPrometheusInstantReqParams(r *http.Request, query string, ti
 	}
 	q := r.URL.Query()
 	q.Set("time", timestamp.Format(time.RFC3339))
-	if !*disableStepParam && c.evaluationInterval > 0 { // set step as evaluationInterval by default
+	if !*disableStepParam && s.evaluationInterval > 0 { // set step as evaluationInterval by default
 		// always convert to seconds to keep compatibility with older
 		// Prometheus versions. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1943
-		q.Set("step", fmt.Sprintf("%ds", int(c.evaluationInterval.Seconds())))
+		q.Set("step", fmt.Sprintf("%ds", int(s.evaluationInterval.Seconds())))
 	}
-	if !*disableStepParam && c.queryStep > 0 { // override step with user-specified value
+	if !*disableStepParam && s.queryStep > 0 { // override step with user-specified value
 		// always convert to seconds to keep compatibility with older
 		// Prometheus versions. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1943
-		q.Set("step", fmt.Sprintf("%ds", int(c.queryStep.Seconds())))
+		q.Set("step", fmt.Sprintf("%ds", int(s.queryStep.Seconds())))
 	}
 	r.URL.RawQuery = q.Encode()
-	c.setReqParams(r, query)
+	s.setReqParams(r, query)
 }
 
-func (c *Client) setPrometheusRangeReqParams(r *http.Request, query string, start, end time.Time) {
-	if c.appendTypePrefix {
+func (s *Client) setPrometheusRangeReqParams(r *http.Request, query string, start, end time.Time) {
+	if s.appendTypePrefix {
 		r.URL.Path += "/prometheus"
 	}
 	if !*disablePathAppend {
@@ -253,11 +251,11 @@ func (c *Client) setPrometheusRangeReqParams(r *http.Request, query string, star
 	q := r.URL.Query()
 	q.Add("start", start.Format(time.RFC3339))
 	q.Add("end", end.Format(time.RFC3339))
-	if c.evaluationInterval > 0 { // set step as evaluationInterval by default
+	if s.evaluationInterval > 0 { // set step as evaluationInterval by default
 		// always convert to seconds to keep compatibility with older
 		// Prometheus versions. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1943
-		q.Set("step", fmt.Sprintf("%ds", int(c.evaluationInterval.Seconds())))
+		q.Set("step", fmt.Sprintf("%ds", int(s.evaluationInterval.Seconds())))
 	}
 	r.URL.RawQuery = q.Encode()
-	c.setReqParams(r, query)
+	s.setReqParams(r, query)
 }

app/vmalert/datasource/client_test.go

@@ -14,7 +14,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
 var (
@@ -145,12 +144,12 @@ func TestVMInstantQuery(t *testing.T) {
 	}
 	expected := []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Value: "vm_rows", Name: "__name__"}, {Value: "bar", Name: "foo"}},
+			Labels:     []Label{{Value: "vm_rows", Name: "__name__"}, {Value: "bar", Name: "foo"}},
 			Timestamps: []int64{1583786142},
 			Values:     []float64{13763},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Value: "vm_requests", Name: "__name__"}, {Value: "baz", Name: "foo"}},
+			Labels:     []Label{{Value: "vm_requests", Name: "__name__"}, {Value: "baz", Name: "foo"}},
 			Timestamps: []int64{1583786140},
 			Values:     []float64{2000},
 		},
@@ -215,7 +214,7 @@ func TestVMInstantQuery(t *testing.T) {
 	}
 	exp := []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Value: "constantLine(10)", Name: "name"}},
+			Labels:     []Label{{Value: "constantLine(10)", Name: "name"}},
 			Timestamps: []int64{1611758403},
 			Values:     []float64{10},
 		},
@@ -237,12 +236,12 @@ func TestVMInstantQuery(t *testing.T) {
 	}
 	expected = []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Value: "total", Name: "stats_result"}, {Value: "bar", Name: "foo"}},
+			Labels:     []Label{{Value: "total", Name: "stats_result"}, {Value: "bar", Name: "foo"}},
 			Timestamps: []int64{1583786142},
 			Values:     []float64{13763},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Value: "total", Name: "stats_result"}, {Value: "baz", Name: "foo"}},
+			Labels:     []Label{{Value: "total", Name: "stats_result"}, {Value: "baz", Name: "foo"}},
 			Timestamps: []int64{1583786140},
 			Values:     []float64{2000},
 		},
@@ -445,7 +444,7 @@ func TestVMRangeQuery(t *testing.T) {
 		t.Fatalf("expected 1 metric  got %d in %+v", len(m), m)
 	}
 	expected := Metric{
-		Labels:     []prompbmarshal.Label{{Value: "vm_rows", Name: "__name__"}},
+		Labels:     []Label{{Value: "vm_rows", Name: "__name__"}},
 		Timestamps: []int64{1583786142},
 		Values:     []float64{13763},
 	}
@@ -476,7 +475,7 @@ func TestVMRangeQuery(t *testing.T) {
 		t.Fatalf("expected 1 metric  got %d in %+v", len(m), m)
 	}
 	expected = Metric{
-		Labels:     []prompbmarshal.Label{{Value: "total", Name: "stats_result"}},
+		Labels:     []Label{{Value: "total", Name: "stats_result"}},
 		Timestamps: []int64{1583786142},
 		Values:     []float64{10},
 	}

app/vmalert/datasource/client_vlogs.go

@@ -6,7 +6,7 @@ import (
 	"time"
 )
 
-func (c *Client) setVLogsInstantReqParams(r *http.Request, query string, timestamp time.Time) {
+func (s *Client) setVLogsInstantReqParams(r *http.Request, query string, timestamp time.Time) {
 	// there is no type path prefix in victorialogs APIs right now, ignore appendTypePrefix.
 	if !*disablePathAppend {
 		r.URL.Path += "/select/logsql/stats_query"
@@ -16,15 +16,15 @@ func (c *Client) setVLogsInstantReqParams(r *http.Request, query string, timesta
 	q.Set("time", timestamp.Format(time.RFC3339))
 	// set the `start` and `end` params if applyIntervalAsTimeFilter is enabled(time filter is missing in the rule expr),
 	// so the query will be executed in time range [timestamp - evaluationInterval, timestamp].
-	if c.applyIntervalAsTimeFilter && c.evaluationInterval > 0 {
-		q.Set("start", timestamp.Add(-c.evaluationInterval).Format(time.RFC3339))
+	if s.applyIntervalAsTimeFilter && s.evaluationInterval > 0 {
+		q.Set("start", timestamp.Add(-s.evaluationInterval).Format(time.RFC3339))
 		q.Set("end", timestamp.Format(time.RFC3339))
 	}
 	r.URL.RawQuery = q.Encode()
-	c.setReqParams(r, query)
+	s.setReqParams(r, query)
 }
 
-func (c *Client) setVLogsRangeReqParams(r *http.Request, query string, start, end time.Time) {
+func (s *Client) setVLogsRangeReqParams(r *http.Request, query string, start, end time.Time) {
 	// there is no type path prefix in victorialogs APIs right now, ignore appendTypePrefix.
 	if !*disablePathAppend {
 		r.URL.Path += "/select/logsql/stats_query_range"
@@ -33,11 +33,11 @@ func (c *Client) setVLogsRangeReqParams(r *http.Request, query string, start, en
 	q.Add("start", start.Format(time.RFC3339))
 	q.Add("end", end.Format(time.RFC3339))
 	// set step as evaluationInterval by default
-	if c.evaluationInterval > 0 {
-		q.Set("step", fmt.Sprintf("%ds", int(c.evaluationInterval.Seconds())))
+	if s.evaluationInterval > 0 {
+		q.Set("step", fmt.Sprintf("%ds", int(s.evaluationInterval.Seconds())))
 	}
 	r.URL.RawQuery = q.Encode()
-	c.setReqParams(r, query)
+	s.setReqParams(r, query)
 }
 
 func parseVLogsResponse(req *http.Request, resp *http.Response) (res Result, err error) {

app/vmalert/datasource/datasource.go

@@ -8,8 +8,6 @@ import (
 	"sort"
 	"strconv"
 	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
 // Querier interface wraps Query and QueryRange methods
@@ -57,7 +55,7 @@ type QuerierParams struct {
 
 // Metric is the basic entity which should be return by datasource
 type Metric struct {
-	Labels     []prompbmarshal.Label
+	Labels     []Label
 	Timestamps []int64
 	Values     []float64
 }
@@ -74,9 +72,22 @@ func (m *Metric) SetLabel(key, value string) {
 	m.AddLabel(key, value)
 }
 
+// SetLabels sets the given map as Metric labels
+func (m *Metric) SetLabels(ls map[string]string) {
+	var i int
+	m.Labels = make([]Label, len(ls))
+	for k, v := range ls {
+		m.Labels[i] = Label{
+			Name:  k,
+			Value: v,
+		}
+		i++
+	}
+}
+
 // AddLabel appends the given label to the label set
 func (m *Metric) AddLabel(key, value string) {
-	m.Labels = append(m.Labels, prompbmarshal.Label{Name: key, Value: value})
+	m.Labels = append(m.Labels, Label{Name: key, Value: value})
 }
 
 // DelLabel deletes the given label from the label set
@@ -99,8 +110,14 @@ func (m *Metric) Label(key string) string {
 	return ""
 }
 
+// Label represents metric's label
+type Label struct {
+	Name  string
+	Value string
+}
+
 // Labels is collection of Label
-type Labels []prompbmarshal.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] }
@@ -155,7 +172,7 @@ func LabelCompare(a, b Labels) int {
 // ConvertToLabels convert map to Labels
 func ConvertToLabels(m map[string]string) (labelset Labels) {
 	for k, v := range m {
-		labelset = append(labelset, prompbmarshal.Label{
+		labelset = append(labelset, Label{
 			Name:  k,
 			Value: v,
 		})

app/vmalert/datasource/init.go

@@ -11,6 +11,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 )
 
@@ -47,15 +48,22 @@ var (
 	oauth2TokenURL = flag.String("datasource.oauth2.tokenUrl", "", "Optional OAuth2 tokenURL to use for -datasource.url")
 	oauth2Scopes   = flag.String("datasource.oauth2.scopes", "", "Optional OAuth2 scopes to use for -datasource.url. Scopes must be delimited by ';'")
 
-	queryStep = flag.Duration("datasource.queryStep", 5*time.Minute, "How far a value can fallback to when evaluating queries to the configured -datasource.url and -remoteRead.url. Only valid for prometheus datasource. "+
+	lookBack = flag.Duration("datasource.lookback", 0, `Deprecated: please adjust "-search.latencyOffset" at datasource side `+
+		`or specify "latency_offset" in rule group's params. Lookback defines how far into the past to look when evaluating queries. `+
+		`For example, if the datasource.lookback=5m then param "time" with value now()-5m will be added to every query.`)
+	queryStep = flag.Duration("datasource.queryStep", 5*time.Minute, "How far a value can fallback to when evaluating queries. "+
 		"For example, if -datasource.queryStep=15s then param \"step\" with value \"15s\" will be added to every query. "+
 		"If set to 0, rule's evaluation interval will be used instead.")
+	queryTimeAlignment = flag.Bool("datasource.queryTimeAlignment", true, `Deprecated: please use "eval_alignment" in rule group instead. `+
+		`Whether to align "time" parameter with evaluation interval. `+
+		"Alignment supposed to produce deterministic results despite number of vmalert replicas or time they were started. "+
+		"See more details at https://github.com/VictoriaMetrics/VictoriaMetrics/pull/1257")
 	maxIdleConnections    = flag.Int("datasource.maxIdleConnections", 100, `Defines the number of idle (keep-alive connections) to each configured datasource. Consider setting this value equal to the value: groups_total * group.concurrency. Too low a value may result in a high number of sockets in TIME_WAIT state.`)
 	idleConnectionTimeout = flag.Duration("datasource.idleConnTimeout", 50*time.Second, `Defines a duration for idle (keep-alive connections) to exist. Consider setting this value less than "-http.idleConnTimeout". It must prevent possible "write: broken pipe" and "read: connection reset by peer" errors.`)
 	disableKeepAlive      = flag.Bool("datasource.disableKeepAlive", false, `Whether to disable long-lived connections to the datasource. `+
 		`If true, disables HTTP keep-alive and will only use the connection to the server for a single HTTP request.`)
-	roundDigits = flag.Int("datasource.roundDigits", 0, `Adds "round_digits" GET param to datasource requests which limits the number of digits after the decimal point in response values. `+
-		`Only valid for VictoriaMetrics as the datasource.`)
+	roundDigits = flag.Int("datasource.roundDigits", 0, `Adds "round_digits" GET param to datasource requests. `+
+		`In VM "round_digits" limits the number of digits after the decimal point in response values.`)
 )
 
 // InitSecretFlags must be called after flag.Parse and before any logging
@@ -82,6 +90,12 @@ func Init(extraParams url.Values) (QuerierBuilder, error) {
 	if *addr == "" {
 		return nil, fmt.Errorf("datasource.url is empty")
 	}
+	if !*queryTimeAlignment {
+		logger.Warnf("flag `-datasource.queryTimeAlignment` is deprecated and will be removed in next releases. Please use `eval_alignment` in rule group instead.")
+	}
+	if *lookBack != 0 {
+		logger.Warnf("flag `-datasource.lookback` is deprecated and will be removed in next releases. Please adjust `-search.latencyOffset` at datasource side or specify `latency_offset` in rule group's params. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5155 for details.")
+	}
 
 	tr, err := httputils.Transport(*addr, *tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify)
 	if err != nil {

app/vmalert/datasource/vm_prom_api_test.go

@@ -3,8 +3,6 @@ package datasource
 import (
 	"reflect"
 	"testing"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
 func TestPromInstant_UnmarshalPositive(t *testing.T) {
@@ -23,7 +21,7 @@ func TestPromInstant_UnmarshalPositive(t *testing.T) {
 
 	f(`[{"metric":{"__name__":"up"},"value":[1583780000,"42"]}]`, []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "up"}},
+			Labels:     []Label{{Name: "__name__", Value: "up"}},
 			Timestamps: []int64{1583780000},
 			Values:     []float64{42},
 		},
@@ -33,17 +31,17 @@ func TestPromInstant_UnmarshalPositive(t *testing.T) {
 {"metric":{"__name__":"foo"},"value":[1583780001,"7"]},
 {"metric":{"__name__":"baz", "instance":"bar"},"value":[1583780002,"8"]}]`, []Metric{
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "up"}},
+			Labels:     []Label{{Name: "__name__", Value: "up"}},
 			Timestamps: []int64{1583780000},
 			Values:     []float64{42},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "foo"}},
+			Labels:     []Label{{Name: "__name__", Value: "foo"}},
 			Timestamps: []int64{1583780001},
 			Values:     []float64{7},
 		},
 		{
-			Labels:     []prompbmarshal.Label{{Name: "__name__", Value: "baz"}, {Name: "instance", Value: "bar"}},
+			Labels:     []Label{{Name: "__name__", Value: "baz"}, {Name: "instance", Value: "bar"}},
 			Timestamps: []int64{1583780002},
 			Values:     []float64{8},
 		},

app/vmalert/main.go

@@ -78,13 +78,12 @@ absolute path to all .tpl files in root.
 	externalLabels = flagutil.NewArrayString("external.label", "Optional label in the form 'Name=value' to add to all generated recording rules and alerts. "+
 		"In case of conflicts, original labels are kept with prefix `exported_`.")
 
+	remoteReadIgnoreRestoreErrors = flag.Bool("remoteRead.ignoreRestoreErrors", true, "Whether to ignore errors from remote storage when restoring alerts state on startup. DEPRECATED - this flag has no effect and will be removed in the next releases.")
+
 	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.")
 )
 
-var (
-	alertURLGeneratorFn notifier.AlertURLGenerator
-	extURL              *url.URL
-)
+var alertURLGeneratorFn notifier.AlertURLGenerator
 
 func main() {
 	// Write flags and help message to stdout, since it is easier to grep or pipe.
@@ -98,15 +97,13 @@ func main() {
 	buildinfo.Init()
 	logger.Init()
 
-	var err error
-	extURL, err = getExternalURL(*externalURL)
-	if err != nil {
-		logger.Fatalf("failed to init external.url %q: %s", *externalURL, err)
+	if !*remoteReadIgnoreRestoreErrors {
+		logger.Warnf("flag `remoteRead.ignoreRestoreErrors` is deprecated and will be removed in next releases.")
 	}
 
-	err = templates.Load(*ruleTemplatesPath, *extURL)
+	err := templates.Load(*ruleTemplatesPath, true)
 	if err != nil {
-		logger.Fatalf("failed to load template %q: %s", *ruleTemplatesPath, err)
+		logger.Fatalf("failed to parse %q: %s", *ruleTemplatesPath, err)
 	}
 
 	if *dryRun {
@@ -120,7 +117,12 @@ func main() {
 		return
 	}
 
-	alertURLGeneratorFn, err = getAlertURLGenerator(extURL, *externalAlertSource, *validateTemplates)
+	eu, err := getExternalURL(*externalURL)
+	if err != nil {
+		logger.Fatalf("failed to init `-external.url`: %s", err)
+	}
+
+	alertURLGeneratorFn, err = getAlertURLGenerator(eu, *externalAlertSource, *validateTemplates)
 	if err != nil {
 		logger.Fatalf("failed to init `external.alert.source`: %s", err)
 	}
@@ -308,7 +310,7 @@ func getAlertURLGenerator(externalURL *url.URL, externalAlertSource string, vali
 		}
 		templated, err := alert.ExecTemplate(qFn, alert.Labels, m)
 		if err != nil {
-			logger.Errorf("cannot template alert source: %s", err)
+			logger.Errorf("can not exec source template %s", err)
 		}
 		return fmt.Sprintf("%s/%s", externalURL, templated["tpl"])
 	}, nil
@@ -363,7 +365,7 @@ func configReload(ctx context.Context, m *manager, groupsCfg []config.Group, sig
 			logger.Errorf("failed to reload notifier config: %s", err)
 			continue
 		}
-		err := templates.Load(*ruleTemplatesPath, *extURL)
+		err := templates.Load(*ruleTemplatesPath, false)
 		if err != nil {
 			setConfigError(err)
 			logger.Errorf("failed to load new templates: %s", err)

app/vmalert/main_test.go

@@ -74,10 +74,7 @@ func TestGetAlertURLGenerator(t *testing.T) {
 
 func TestConfigReload(t *testing.T) {
 	originalRulePath := *rulePath
-	originalExternalURL := extURL
-	extURL = &url.URL{}
 	defer func() {
-		extURL = originalExternalURL
 		*rulePath = originalRulePath
 	}()
 

app/vmalert/manager_test.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"math/rand"
-	"net/url"
 	"os"
 	"strings"
 	"sync"
@@ -19,7 +18,7 @@ import (
 )
 
 func TestMain(m *testing.M) {
-	if err := templates.Load([]string{"testdata/templates/*good.tmpl"}, url.URL{}); err != nil {
+	if err := templates.Load([]string{"testdata/templates/*good.tmpl"}, true); err != nil {
 		os.Exit(1)
 	}
 	os.Exit(m.Run())

app/vmalert/notifier/alert.go

@@ -127,7 +127,7 @@ func ExecTemplate(q templates.QueryFn, annotations map[string]string, tplData Al
 
 // ValidateTemplates validate annotations for possible template error, uses empty data for template population
 func ValidateTemplates(annotations map[string]string) error {
-	tmpl, err := templates.GetWithFuncs(nil)
+	tmpl, err := templates.Get()
 	if err != nil {
 		return err
 	}
@@ -146,21 +146,12 @@ func templateAnnotations(annotations map[string]string, data AlertTplData, tmpl
 	tData := tplData{data, externalLabels, externalURL}
 	header := strings.Join(tplHeaders, "")
 	for key, text := range annotations {
-		// simple check to skip text without template
-		if !strings.Contains(text, "{{") || !strings.Contains(text, "}}") {
-			r[key] = text
-			continue
-		}
-
 		buf.Reset()
 		builder.Reset()
 		builder.Grow(len(header) + len(text))
 		builder.WriteString(header)
 		builder.WriteString(text)
-		// clone a new template for each parse to avoid collision
-		ctmpl, _ := tmpl.Clone()
-		ctmpl = ctmpl.Option("missingkey=zero")
-		if err := templateAnnotation(&buf, builder.String(), tData, ctmpl, execute); err != nil {
+		if err := templateAnnotation(&buf, builder.String(), tData, tmpl, execute); err != nil {
 			r[key] = text
 			eg.Add(fmt.Errorf("key %q, template %q: %w", key, text, err))
 			continue
@@ -176,8 +167,14 @@ type tplData struct {
 	ExternalURL    string
 }
 
-func templateAnnotation(dst io.Writer, text string, data tplData, tpl *textTpl.Template, execute bool) error {
-	tpl, err := tpl.Parse(text)
+func templateAnnotation(dst io.Writer, text string, data tplData, tmpl *textTpl.Template, execute bool) error {
+	tpl, err := tmpl.Clone()
+	if err != nil {
+		return fmt.Errorf("error cloning template before parse annotation: %w", err)
+	}
+	// Clone() doesn't copy tpl Options, so we set them manually
+	tpl = tpl.Option("missingkey=zero")
+	tpl, err = tpl.Parse(text)
 	if err != nil {
 		return fmt.Errorf("error parsing annotation template: %w", err)
 	}

app/vmalert/notifier/alert_test.go

@@ -33,7 +33,7 @@ func TestAlertExecTemplate(t *testing.T) {
 		qFn := func(_ string) ([]datasource.Metric, error) {
 			return []datasource.Metric{
 				{
-					Labels: []prompbmarshal.Label{
+					Labels: []datasource.Label{
 						{Name: "foo", Value: "bar"},
 						{Name: "baz", Value: "qux"},
 					},
@@ -41,7 +41,7 @@ func TestAlertExecTemplate(t *testing.T) {
 					Timestamps: []int64{1},
 				},
 				{
-					Labels: []prompbmarshal.Label{
+					Labels: []datasource.Label{
 						{Name: "foo", Value: "garply"},
 						{Name: "baz", Value: "fred"},
 					},
@@ -75,13 +75,7 @@ func TestAlertExecTemplate(t *testing.T) {
 		Labels: map[string]string{
 			"instance": "localhost",
 		},
-	}, map[string]string{
-		"summary":     "it's a test summary",
-		"description": "it's a test description",
-	}, map[string]string{
-		"summary":     "it's a test summary",
-		"description": "it's a test description",
-	})
+	}, map[string]string{}, map[string]string{})
 
 	// label-template
 	f(&Alert{
@@ -99,19 +93,6 @@ func TestAlertExecTemplate(t *testing.T) {
 		"description": "It is 10000 connections for localhost for more than 5m0s",
 	})
 
-	// label template override
-	f(&Alert{
-		Value: 1e4,
-	}, map[string]string{
-		"summary":     `{{- define "default.template" -}} {{ printf "summary" }} {{- end -}} {{ template "default.template" . }}`,
-		"description": `{{ template "default.template" . }}`,
-		"value":       `{{$value }}`,
-	}, map[string]string{
-		"summary":     "summary",
-		"description": "",
-		"value":       "10000",
-	})
-
 	// expression-template
 	f(&Alert{
 		Expr: `vm_rows{"label"="bar"}<0`,

app/vmalert/notifier/init.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
@@ -92,11 +93,13 @@ var (
 func Init(gen AlertURLGenerator, extLabels map[string]string, extURL string) (func() []Notifier, error) {
 	externalURL = extURL
 	externalLabels = extLabels
-	_, err := url.Parse(externalURL)
+	eu, err := url.Parse(externalURL)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse external URL: %w", err)
 	}
 
+	templates.UpdateWithFuncs(templates.FuncsWithExternalURL(eu))
+
 	if *blackHole {
 		if len(*addrs) > 0 || *configPath != "" {
 			return nil, fmt.Errorf("only one of -notifier.blackhole, -notifier.url and -notifier.config flags must be specified")

app/vmalert/notifier/package_test.go

@@ -1,15 +1,13 @@
 package notifier
 
 import (
-	"net/url"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
 	"os"
 	"testing"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
 )
 
 func TestMain(m *testing.M) {
-	if err := templates.Load([]string{"testdata/templates/*good.tmpl"}, url.URL{}); err != nil {
+	if err := templates.Load([]string{"testdata/templates/*good.tmpl"}, true); err != nil {
 		os.Exit(1)
 	}
 	os.Exit(m.Run())

app/vmalert/remoteread/init.go

@@ -14,7 +14,7 @@ import (
 )
 
 var (
-	addr = flag.String("remoteRead.url", "", "Optional URL to datasource compatible with MetricsQL. It can be single node VictoriaMetrics or vmselect."+
+	addr = flag.String("remoteRead.url", "", "Optional URL to datasource compatible with Prometheus HTTP API. It can be single node VictoriaMetrics or vmselect."+
 		"Remote read is used to restore alerts state."+
 		"This configuration makes sense only if `vmalert` was configured with `remoteWrite.url` before and has been successfully persisted its state. "+
 		"Supports address in the form of IP address with a port (e.g., http://127.0.0.1:8428) or DNS SRV record. "+

app/vmalert/remotewrite/client.go

@@ -27,7 +27,7 @@ var defaultConcurrency = cgroup.AvailableCPUs() * 2
 
 const (
 	defaultMaxBatchSize  = 1e4
-	defaultMaxQueueSize  = 1e5
+	defaultMaxQueueSize  = 1e6
 	defaultFlushInterval = 2 * time.Second
 	defaultWriteTimeout  = 30 * time.Second
 )

app/vmalert/rule/alerting.go

@@ -14,10 +14,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
 // AlertingRule is basic alert entity
@@ -456,16 +454,13 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 		ar.logDebugf(ts, a, "created in state PENDING")
 	}
 	var numActivePending int
-	var tss []prompbmarshal.TimeSeries
 	for h, a := range ar.alerts {
 		// if alert wasn't updated in this iteration
 		// means it is resolved already
 		if _, ok := updated[h]; !ok {
 			if a.State == notifier.StatePending {
-				// alert was in Pending state - it is not active anymore
-				// add stale time series
-				tss = append(tss, pendingAlertStaleTimeSeries(a.Labels, ts.Unix(), true)...)
-
+				// alert was in Pending state - it is not
+				// active anymore
 				delete(ar.alerts, h)
 				ar.logDebugf(ts, a, "PENDING => DELETED: is absent in current evaluation round")
 				continue
@@ -483,9 +478,6 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 				if ts.Sub(a.KeepFiringSince) >= ar.KeepFiringFor {
 					a.State = notifier.StateInactive
 					a.ResolvedAt = ts
-					// add stale time series
-					tss = append(tss, firingAlertStaleTimeSeries(a.Labels, ts.Unix())...)
-
 					ar.logDebugf(ts, a, "FIRING => INACTIVE: is absent in current evaluation round")
 					continue
 				}
@@ -497,10 +489,6 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 			a.State = notifier.StateFiring
 			a.Start = ts
 			alertsFired.Inc()
-			if ar.For > 0 {
-				// add stale time series
-				tss = append(tss, pendingAlertStaleTimeSeries(a.Labels, ts.Unix(), false)...)
-			}
 			ar.logDebugf(ts, a, "PENDING => FIRING: %s since becoming active at %v", ts.Sub(a.ActiveAt), a.ActiveAt)
 		}
 	}
@@ -509,7 +497,7 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 		curState.Err = fmt.Errorf("exec exceeded limit of %d with %d alerts", limit, numActivePending)
 		return nil, curState.Err
 	}
-	return append(tss, ar.toTimeSeries(ts.Unix())...), nil
+	return ar.toTimeSeries(ts.Unix()), nil
 }
 
 func (ar *AlertingRule) expandTemplates(m datasource.Metric, qFn templates.QueryFn, ts time.Time) (*labelSet, map[string]string, error) {
@@ -534,7 +522,6 @@ func (ar *AlertingRule) expandTemplates(m datasource.Metric, qFn templates.Query
 	return ls, as, nil
 }
 
-// toTimeSeries creates `ALERTS` and `ALERTS_FOR_STATE` for active alerts
 func (ar *AlertingRule) toTimeSeries(timestamp int64) []prompbmarshal.TimeSeries {
 	var tss []prompbmarshal.TimeSeries
 	for _, a := range ar.alerts {
@@ -614,83 +601,26 @@ func (ar *AlertingRule) alertToTimeSeries(a *notifier.Alert, timestamp int64) []
 }
 
 func alertToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSeries {
-	var labels []prompbmarshal.Label
+	labels := make(map[string]string)
 	for k, v := range a.Labels {
-		labels = append(labels, prompbmarshal.Label{
-			Name:  k,
-			Value: v,
-		})
-	}
-	// __name__ already been dropped, no need to check duplication
-	labels = append(labels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
-	if ol := promrelabel.GetLabelByName(labels, alertStateLabel); ol != nil {
-		ol.Value = a.State.String()
-	} else {
-		labels = append(labels, prompbmarshal.Label{Name: alertStateLabel, Value: a.State.String()})
+		labels[k] = v
 	}
+	labels["__name__"] = alertMetricName
+	labels[alertStateLabel] = a.State.String()
 	return newTimeSeries([]float64{1}, []int64{timestamp}, labels)
 }
 
-// alertForToTimeSeries returns a time series that represents
+// alertForToTimeSeries returns a timeseries that represents
 // state of active alerts, where value is time when alert become active
 func alertForToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSeries {
-	var labels []prompbmarshal.Label
+	labels := make(map[string]string)
 	for k, v := range a.Labels {
-		labels = append(labels, prompbmarshal.Label{
-			Name:  k,
-			Value: v,
-		})
+		labels[k] = v
 	}
-	// __name__ already been dropped, no need to check duplication
-	labels = append(labels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
+	labels["__name__"] = alertForStateMetricName
 	return newTimeSeries([]float64{float64(a.ActiveAt.Unix())}, []int64{timestamp}, labels)
 }
 
-// pendingAlertStaleTimeSeries returns stale `ALERTS` and `ALERTS_FOR_STATE` time series
-// for alerts which changed their state from Pending to Inactive or Firing.
-func pendingAlertStaleTimeSeries(ls map[string]string, timestamp int64, includeAlertForState bool) []prompbmarshal.TimeSeries {
-	var result []prompbmarshal.TimeSeries
-	var baseLabels []prompbmarshal.Label
-	for k, v := range ls {
-		baseLabels = append(baseLabels, prompbmarshal.Label{
-			Name:  k,
-			Value: v,
-		})
-	}
-	// __name__ already been dropped, no need to check duplication
-	alertsLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: alertStateLabel, Value: notifier.StatePending.String()})
-	result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsLabels))
-
-	if includeAlertForState {
-		alertsForStateLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
-		result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsForStateLabels))
-	}
-	return result
-}
-
-// firingAlertStaleTimeSeries returns stale `ALERTS` and `ALERTS_FOR_STATE` time series
-// for alerts which changed their state from Firing to Inactive.
-func firingAlertStaleTimeSeries(ls map[string]string, timestamp int64) []prompbmarshal.TimeSeries {
-	var baseLabels []prompbmarshal.Label
-	for k, v := range ls {
-		baseLabels = append(baseLabels, prompbmarshal.Label{
-			Name:  k,
-			Value: v,
-		})
-	}
-	// __name__ already been dropped, no need to check duplication
-	alertsLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: alertStateLabel, Value: notifier.StateFiring.String()})
-
-	alertsForStateLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
-
-	return []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsLabels),
-		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsForStateLabels),
-	}
-}
-
 // restore restores the value of ActiveAt field for active alerts,
 // based on previously written time series `alertForStateMetricName`.
 // Only rules with For > 0 can be restored.
@@ -711,8 +641,7 @@ func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts ti
 	for k, v := range ar.Labels {
 		labelsFilter += fmt.Sprintf(",%s=%q", k, v)
 	}
-	// use `default_rollup()` instead of `last_over_time()` here to accounts for possible staleness markers
-	expr := fmt.Sprintf("default_rollup(%s{%s%s}[%ds])",
+	expr := fmt.Sprintf("last_over_time(%s{%s%s}[%ds])",
 		alertForStateMetricName, nameStr, labelsFilter, int(lookback.Seconds()))
 
 	res, _, err := q.Query(ctx, expr, ts)

app/vmalert/rule/alerting_test.go

@@ -15,7 +15,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
@@ -29,7 +28,7 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		rule.alerts[alert.ID] = alert
 		tss := rule.toTimeSeries(timestamp.Unix())
 		if err := compareTimeSeries(t, tssExpected, tss); err != nil {
-			t.Fatalf("timeseries mismatch for rule %q: %s", rule.Name, err)
+			t.Fatalf("timeseries mismatch: %s", err)
 		}
 	}
 
@@ -37,23 +36,14 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		State:    notifier.StateFiring,
 		ActiveAt: timestamp.Add(time.Second),
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: alertMetricName,
-			},
-			{
-				Name:  alertStateLabel,
-				Value: notifier.StateFiring.String(),
-			},
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__":      alertMetricName,
+			alertStateLabel: notifier.StateFiring.String(),
 		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: alertForStateMetricName,
-				},
+			map[string]string{
+				"__name__": alertForStateMetricName,
 			}),
 	})
 
@@ -64,40 +54,18 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 			"instance": "bar",
 		},
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: alertMetricName,
-				},
-				{
-					Name:  alertStateLabel,
-					Value: notifier.StateFiring.String(),
-				},
-				{
-					Name:  "job",
-					Value: "foo",
-				},
-				{
-					Name:  "instance",
-					Value: "bar",
-				},
-			}),
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__":      alertMetricName,
+			alertStateLabel: notifier.StateFiring.String(),
+			"job":           "foo",
+			"instance":      "bar",
+		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: alertForStateMetricName,
-				},
-				{
-					Name:  "job",
-					Value: "foo",
-				},
-				{
-					Name:  "instance",
-					Value: "bar",
-				},
+			map[string]string{
+				"__name__": alertForStateMetricName,
+				"job":      "foo",
+				"instance": "bar",
 			}),
 	})
 
@@ -105,29 +73,18 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		State: notifier.StateFiring, ActiveAt: timestamp.Add(time.Second),
 		Labels: map[string]string{
 			alertStateLabel: "foo",
+			"__name__":      "bar",
 		},
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: alertMetricName,
-			},
-			{
-				Name:  alertStateLabel,
-				Value: notifier.StateFiring.String(),
-			},
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__":      alertMetricName,
+			alertStateLabel: notifier.StateFiring.String(),
 		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: alertForStateMetricName,
-				},
-				{
-					Name:  alertStateLabel,
-					Value: "foo",
-				},
+			map[string]string{
+				"__name__":      alertForStateMetricName,
+				alertStateLabel: "foo",
 			}),
 	})
 
@@ -135,23 +92,14 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		State:    notifier.StateFiring,
 		ActiveAt: timestamp.Add(time.Second),
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: alertMetricName,
-			},
-			{
-				Name:  alertStateLabel,
-				Value: notifier.StateFiring.String(),
-			},
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__":      alertMetricName,
+			alertStateLabel: notifier.StateFiring.String(),
 		}),
 		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())},
 			[]int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: alertForStateMetricName,
-				},
+			map[string]string{
+				"__name__": alertForStateMetricName,
 			}),
 	})
 
@@ -159,21 +107,12 @@ func TestAlertingRuleToTimeSeries(t *testing.T) {
 		State:    notifier.StatePending,
 		ActiveAt: timestamp.Add(time.Second),
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: alertMetricName,
-			},
-			{
-				Name:  alertStateLabel,
-				Value: notifier.StatePending.String(),
-			},
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__":      alertMetricName,
+			alertStateLabel: notifier.StatePending.String(),
 		}),
-		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: alertForStateMetricName,
-			},
+		newTimeSeries([]float64{float64(timestamp.Add(time.Second).Unix())}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": alertForStateMetricName,
 		}),
 	})
 }
@@ -185,9 +124,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		alert  *notifier.Alert
 	}
 
-	ts, _ := time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
-
-	f := func(rule *AlertingRule, steps [][]datasource.Metric, alertsExpected map[int][]testAlert, tssExpected map[int][]prompbmarshal.TimeSeries) {
+	f := func(rule *AlertingRule, steps [][]datasource.Metric, alertsExpected map[int][]testAlert) {
 		t.Helper()
 
 		fq := &datasource.FakeQuerier{}
@@ -197,19 +134,13 @@ func TestAlertingRule_Exec(t *testing.T) {
 			Name: "TestRule_Exec",
 		}
 		rule.GroupID = fakeGroup.ID()
+		ts := time.Now()
 		for i, step := range steps {
 			fq.Reset()
 			fq.Add(step...)
-			tss, err := rule.exec(context.TODO(), ts, 0)
-			if err != nil {
+			if _, err := rule.exec(context.TODO(), ts, 0); err != nil {
 				t.Fatalf("unexpected error: %s", err)
 			}
-			// check generate time series
-			if _, ok := tssExpected[i]; ok {
-				if err := compareTimeSeries(t, tssExpected[i], tss); err != nil {
-					t.Fatalf("generated time series mismatch for rule %q in step %d: %s", rule.Name, i, err)
-				}
-			}
 
 			// shift the execution timestamp before the next iteration
 			ts = ts.Add(defaultStep)
@@ -243,21 +174,13 @@ func TestAlertingRule_Exec(t *testing.T) {
 		}
 	}
 
-	f(newTestAlertingRule("empty", 0), [][]datasource.Metric{}, nil, nil)
+	f(newTestAlertingRule("empty", 0), [][]datasource.Metric{}, nil)
 
-	f(newTestAlertingRule("empty_labels", 0), [][]datasource.Metric{
+	f(newTestAlertingRule("empty labels", 0), [][]datasource.Metric{
 		{datasource.Metric{Values: []float64{1}, Timestamps: []int64{1}}},
 	}, map[int][]testAlert{
 		0: {{alert: &notifier.Alert{State: notifier.StateFiring}}},
-	},
-		map[int][]prompbmarshal.TimeSeries{
-			0: {
-				{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "empty_labels"}, {Name: "alertstate", Value: "firing"}},
-					Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
-				{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "empty_labels"}},
-					Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
-			},
-		})
+	})
 
 	f(newTestAlertingRule("single-firing=>inactive=>firing=>inactive=>inactive", 0), [][]datasource.Metric{
 		{metricWithLabels(t, "name", "foo")},
@@ -271,25 +194,6 @@ func TestAlertingRule_Exec(t *testing.T) {
 		2: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
 		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
-	}, map[int][]prompbmarshal.TimeSeries{
-		0: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
-		},
-		1: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-		},
-		2: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-		},
 	})
 
 	f(newTestAlertingRule("single-firing=>inactive=>firing=>inactive=>inactive=>firing", 0), [][]datasource.Metric{
@@ -306,7 +210,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 		5: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-	}, nil)
+	})
 
 	f(newTestAlertingRule("multiple-firing", 0), [][]datasource.Metric{
 		{
@@ -320,7 +224,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 			{labels: []string{"name", "foo1"}, alert: &notifier.Alert{State: notifier.StateFiring}},
 			{labels: []string{"name", "foo2"}, alert: &notifier.Alert{State: notifier.StateFiring}},
 		},
-	}, nil)
+	})
 
 	// 1: fire first alert
 	// 2: fire second alert, set first inactive
@@ -329,57 +233,27 @@ func TestAlertingRule_Exec(t *testing.T) {
 		{metricWithLabels(t, "name", "foo")},
 		{metricWithLabels(t, "name", "foo1")},
 		{metricWithLabels(t, "name", "foo2")},
-	}, map[int][]testAlert{
-		0: {
-			{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}},
-		},
-		1: {
-			{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}},
-			{labels: []string{"name", "foo1"}, alert: &notifier.Alert{State: notifier.StateFiring}},
-		},
-		2: {
-			{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}},
-			{labels: []string{"name", "foo1"}, alert: &notifier.Alert{State: notifier.StateInactive}},
-			{labels: []string{"name", "foo2"}, alert: &notifier.Alert{State: notifier.StateFiring}},
-		},
-	}, map[int][]prompbmarshal.TimeSeries{
-		0: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
-		},
-		1: {
-			// stale time series for foo, `firing -> inactive`
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			// new time series for foo1
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-		},
-		2: {
-			// stale time series for foo1
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-			// new time series for foo2
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo2"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo2"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-		},
-	})
+	},
+		map[int][]testAlert{
+			0: {
+				{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}},
+			},
+			1: {
+				{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}},
+				{labels: []string{"name", "foo1"}, alert: &notifier.Alert{State: notifier.StateFiring}},
+			},
+			2: {
+				{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}},
+				{labels: []string{"name", "foo1"}, alert: &notifier.Alert{State: notifier.StateInactive}},
+				{labels: []string{"name", "foo2"}, alert: &notifier.Alert{State: notifier.StateFiring}},
+			},
+		})
 
 	f(newTestAlertingRule("for-pending", time.Minute), [][]datasource.Metric{
 		{metricWithLabels(t, "name", "foo")},
 	}, map[int][]testAlert{
 		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
-	}, nil)
+	})
 
 	f(newTestAlertingRule("for-fired", defaultStep), [][]datasource.Metric{
 		{metricWithLabels(t, "name", "foo")},
@@ -387,22 +261,6 @@ func TestAlertingRule_Exec(t *testing.T) {
 	}, map[int][]testAlert{
 		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-	}, map[int][]prompbmarshal.TimeSeries{
-		0: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
-		},
-		1: {
-			// stale time series for `pending -> firing`
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-		},
 	})
 
 	f(newTestAlertingRule("for-pending=>empty", time.Second), [][]datasource.Metric{
@@ -414,26 +272,6 @@ func TestAlertingRule_Exec(t *testing.T) {
 		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		2: {},
-	}, map[int][]prompbmarshal.TimeSeries{
-		0: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
-		},
-		1: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
-		},
-		// stale time series for `pending -> inactive`
-		2: {
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
-		},
 	})
 
 	f(newTestAlertingRule("for-pending=>firing=>inactive=>pending=>firing", defaultStep), [][]datasource.Metric{
@@ -449,7 +287,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		2: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-	}, nil)
+	})
 
 	f(newTestAlertingRuleWithCustomFields("for-pending=>firing=>keepfiring=>firing", defaultStep, 0, defaultStep, nil), [][]datasource.Metric{
 		{metricWithLabels(t, "name", "foo")},
@@ -462,7 +300,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
 		2: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
 		3: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-	}, nil)
+	})
 
 	f(newTestAlertingRuleWithCustomFields("for-pending=>firing=>keepfiring=>keepfiring=>inactive=>pending=>firing", defaultStep, 0, 2*defaultStep, nil), [][]datasource.Metric{
 		{metricWithLabels(t, "name", "foo")},
@@ -483,7 +321,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 		5: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		6: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
-	}, nil)
+	})
 }
 
 func TestAlertingRuleExecRange(t *testing.T) {
@@ -639,7 +477,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{Values: []float64{1, 1, 1}, Timestamps: []int64{1, 3, 5}},
 		{
 			Values: []float64{1, 1}, Timestamps: []int64{1, 5},
-			Labels: []prompbmarshal.Label{{Name: "foo", Value: "bar"}},
+			Labels: []datasource.Label{{Name: "foo", Value: "bar"}},
 		},
 	}, []*notifier.Alert{
 		{State: notifier.StatePending, ActiveAt: time.Unix(1, 0)},
@@ -685,7 +523,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{Values: []float64{1, 1}, Timestamps: []int64{1, 100}},
 		{
 			Values: []float64{1, 1}, Timestamps: []int64{1, 5},
-			Labels: []prompbmarshal.Label{{Name: "foo", Value: "bar"}},
+			Labels: []datasource.Label{{Name: "foo", Value: "bar"}},
 		},
 	}, []*notifier.Alert{
 		{
@@ -791,7 +629,7 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert with state restore
 	ts := time.Now().Truncate(time.Hour)
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
 		stateMetric("foo", ts))
 	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)}},
@@ -804,7 +642,7 @@ func TestGroup_Restore(t *testing.T) {
 
 	// two rules, two active alerts, one with state restored
 	ts = time.Now().Truncate(time.Hour)
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
 		stateMetric("bar", ts))
 	fn(
 		[]config.Rule{
@@ -824,9 +662,9 @@ func TestGroup_Restore(t *testing.T) {
 
 	// two rules, two active alerts, two with state restored
 	ts = time.Now().Truncate(time.Hour)
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
 		stateMetric("foo", ts))
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="bar"}[3600s])`,
 		stateMetric("bar", ts))
 	fn(
 		[]config.Rule{
@@ -846,7 +684,7 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert but wrong state restore
 	ts = time.Now().Truncate(time.Hour)
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertname="bar",alertgroup="TestRestore"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertname="bar",alertgroup="TestRestore"}[3600s])`,
 		stateMetric("wrong alert", ts))
 	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)}},
@@ -859,7 +697,7 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert with labels
 	ts = time.Now().Truncate(time.Hour)
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev"))
 	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutils.NewDuration(time.Second)}},
@@ -872,7 +710,7 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert with restore labels missmatch
 	ts = time.Now().Truncate(time.Hour)
-	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
+	fqr.Set(`last_over_time(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev", "team", "foo"))
 	fn(
 		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutils.NewDuration(time.Second)}},
@@ -1209,7 +1047,7 @@ func newTestAlertingRuleWithCustomFields(name string, waitFor, evalInterval, kee
 
 func TestAlertingRule_ToLabels(t *testing.T) {
 	metric := datasource.Metric{
-		Labels: []prompbmarshal.Label{
+		Labels: []datasource.Label{
 			{Name: "instance", Value: "0.0.0.0:8800"},
 			{Name: "group", Value: "vmalert"},
 			{Name: "alertname", Value: "ConfigurationReloadFailure"},

app/vmalert/rule/group.go

@@ -8,9 +8,12 @@ import (
 	"fmt"
 	"hash/fnv"
 	"net/url"
+	"strconv"
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+
 	"github.com/cheggaaa/pb/v3"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
@@ -18,6 +21,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/metrics"
@@ -346,9 +350,10 @@ func (g *Group) Start(ctx context.Context, nts func() []notifier.Notifier, rw re
 	}
 
 	e := &executor{
-		Rw:              rw,
-		Notifiers:       nts,
-		notifierHeaders: g.NotifierHeaders,
+		Rw:                       rw,
+		Notifiers:                nts,
+		notifierHeaders:          g.NotifierHeaders,
+		previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
 	}
 
 	g.infof("started")
@@ -421,6 +426,8 @@ func (g *Group) Start(ctx context.Context, nts func() []notifier.Notifier, rw re
 				continue
 			}
 
+			// ensure that staleness is tracked for existing rules only
+			e.purgeStaleSeries(g.Rules)
 			e.notifierHeaders = g.NotifierHeaders
 			g.mu.Unlock()
 
@@ -532,9 +539,10 @@ func (g *Group) Replay(start, end time.Time, rw remotewrite.RWClient, maxDataPoi
 // ExecOnce evaluates all the rules under group for once with given timestamp.
 func (g *Group) ExecOnce(ctx context.Context, nts func() []notifier.Notifier, rw remotewrite.RWClient, evalTS time.Time) chan error {
 	e := &executor{
-		Rw:              rw,
-		Notifiers:       nts,
-		notifierHeaders: g.NotifierHeaders,
+		Rw:                       rw,
+		Notifiers:                nts,
+		notifierHeaders:          g.NotifierHeaders,
+		previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
 	}
 	if len(g.Rules) < 1 {
 		return nil
@@ -625,6 +633,13 @@ type executor struct {
 	notifierHeaders map[string]string
 
 	Rw remotewrite.RWClient
+
+	previouslySentSeriesToRWMu sync.Mutex
+	// previouslySentSeriesToRW stores series sent to RW on previous iteration
+	// map[ruleID]map[ruleLabels][]prompb.Label
+	// where `ruleID` is ID of the Rule within a Group
+	// and `ruleLabels` is []prompb.Label marshalled to a string
+	previouslySentSeriesToRW map[uint64]map[string][]prompbmarshal.Label
 }
 
 // execConcurrently executes rules concurrently if concurrency>1
@@ -691,6 +706,11 @@ func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDurati
 		if err := pushToRW(tss); err != nil {
 			return err
 		}
+
+		staleSeries := e.getStaleSeries(r, tss, ts)
+		if err := pushToRW(staleSeries); err != nil {
+			return err
+		}
 	}
 
 	ar, ok := r.(*AlertingRule)
@@ -717,3 +737,79 @@ func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDurati
 	wg.Wait()
 	return errGr.Err()
 }
+
+var bbPool bytesutil.ByteBufferPool
+
+// getStaleSeries checks whether there are stale series from previously sent ones.
+func (e *executor) getStaleSeries(r Rule, tss []prompbmarshal.TimeSeries, timestamp time.Time) []prompbmarshal.TimeSeries {
+	bb := bbPool.Get()
+	defer bbPool.Put(bb)
+
+	ruleLabels := make(map[string][]prompbmarshal.Label, len(tss))
+	for _, ts := range tss {
+		// convert labels to strings, so we can compare with previously sent series
+		bb.B = labelsToString(bb.B, ts.Labels)
+		ruleLabels[string(bb.B)] = ts.Labels
+		bb.Reset()
+	}
+
+	rID := r.ID()
+	var staleS []prompbmarshal.TimeSeries
+	// check whether there are series which disappeared and need to be marked as stale
+	e.previouslySentSeriesToRWMu.Lock()
+	for key, labels := range e.previouslySentSeriesToRW[rID] {
+		if _, ok := ruleLabels[key]; ok {
+			continue
+		}
+		// previously sent series are missing in current series, so we mark them as stale
+		ss := newTimeSeriesPB([]float64{decimal.StaleNaN}, []int64{timestamp.Unix()}, labels)
+		staleS = append(staleS, ss)
+	}
+	// set previous series to current
+	e.previouslySentSeriesToRW[rID] = ruleLabels
+	e.previouslySentSeriesToRWMu.Unlock()
+
+	return staleS
+}
+
+// purgeStaleSeries deletes references in tracked
+// previouslySentSeriesToRW list to Rules which aren't present
+// in the given activeRules list. The method is used when the list
+// of loaded rules has changed and executor has to remove
+// references to non-existing rules.
+func (e *executor) purgeStaleSeries(activeRules []Rule) {
+	newPreviouslySentSeriesToRW := make(map[uint64]map[string][]prompbmarshal.Label)
+
+	e.previouslySentSeriesToRWMu.Lock()
+
+	for _, rule := range activeRules {
+		id := rule.ID()
+		prev, ok := e.previouslySentSeriesToRW[id]
+		if ok {
+			// keep previous series for staleness detection
+			newPreviouslySentSeriesToRW[id] = prev
+		}
+	}
+	e.previouslySentSeriesToRW = nil
+	e.previouslySentSeriesToRW = newPreviouslySentSeriesToRW
+
+	e.previouslySentSeriesToRWMu.Unlock()
+}
+
+func labelsToString(dst []byte, labels []prompbmarshal.Label) []byte {
+	dst = append(dst, '{')
+	for i, label := range labels {
+		if len(label.Name) == 0 {
+			dst = append(dst, "__name__"...)
+		} else {
+			dst = append(dst, label.Name...)
+		}
+		dst = append(dst, '=')
+		dst = strconv.AppendQuote(dst, label.Value)
+		if i < len(labels)-1 {
+			dst = append(dst, ',')
+		}
+	}
+	dst = append(dst, '}')
+	return dst
+}

app/vmalert/rule/group_test.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"net/url"
 	"os"
+	"reflect"
 	"sort"
 	"testing"
 	"time"
@@ -17,6 +17,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
@@ -27,7 +29,7 @@ func init() {
 }
 
 func TestMain(m *testing.M) {
-	if err := templates.Load([]string{}, url.URL{}); err != nil {
+	if err := templates.Load([]string{}, true); err != nil {
 		fmt.Println("failed to load template for test")
 		os.Exit(1)
 	}
@@ -381,6 +383,153 @@ func TestGetResolveDuration(t *testing.T) {
 	f(2*time.Minute, 0, 1*time.Minute, 8*time.Minute)
 }
 
+func TestGetStaleSeries(t *testing.T) {
+	ts := time.Now()
+	e := &executor{
+		previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
+	}
+	f := func(r Rule, labels, expLabels [][]prompbmarshal.Label) {
+		t.Helper()
+
+		var tss []prompbmarshal.TimeSeries
+		for _, l := range labels {
+			tss = append(tss, newTimeSeriesPB([]float64{1}, []int64{ts.Unix()}, l))
+		}
+		staleS := e.getStaleSeries(r, tss, ts)
+		if staleS == nil && expLabels == nil {
+			return
+		}
+		if len(staleS) != len(expLabels) {
+			t.Fatalf("expected to get %d stale series, got %d",
+				len(expLabels), len(staleS))
+		}
+		for i, exp := range expLabels {
+			got := staleS[i]
+			if !reflect.DeepEqual(exp, got.Labels) {
+				t.Fatalf("expected to get labels: \n%v;\ngot instead: \n%v",
+					exp, got.Labels)
+			}
+			if len(got.Samples) != 1 {
+				t.Fatalf("expected to have 1 sample; got %d", len(got.Samples))
+			}
+			if !decimal.IsStaleNaN(got.Samples[0].Value) {
+				t.Fatalf("expected sample value to be %v; got %v", decimal.StaleNaN, got.Samples[0].Value)
+			}
+		}
+	}
+
+	// warn: keep in mind, that executor holds the state, so sequence of f calls matters
+
+	// single series
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "foo")},
+		nil)
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "foo")},
+		nil)
+	f(&AlertingRule{RuleID: 1},
+		nil,
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "foo")})
+	f(&AlertingRule{RuleID: 1},
+		nil,
+		nil)
+
+	// multiple series
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{
+			toPromLabels(t, "__name__", "job:foo", "job", "foo"),
+			toPromLabels(t, "__name__", "job:foo", "job", "bar"),
+		},
+		nil)
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "bar")},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "foo")})
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "bar")},
+		nil)
+	f(&AlertingRule{RuleID: 1},
+		nil,
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "bar")})
+
+	// multiple rules and series
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{
+			toPromLabels(t, "__name__", "job:foo", "job", "foo"),
+			toPromLabels(t, "__name__", "job:foo", "job", "bar"),
+		},
+		nil)
+	f(&AlertingRule{RuleID: 2},
+		[][]prompbmarshal.Label{
+			toPromLabels(t, "__name__", "job:foo", "job", "foo"),
+			toPromLabels(t, "__name__", "job:foo", "job", "bar"),
+		},
+		nil)
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "bar")},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "foo")})
+	f(&AlertingRule{RuleID: 1},
+		[][]prompbmarshal.Label{toPromLabels(t, "__name__", "job:foo", "job", "bar")},
+		nil)
+}
+
+func TestPurgeStaleSeries(t *testing.T) {
+	ts := time.Now()
+	labels := toPromLabels(t, "__name__", "job:foo", "job", "foo")
+	tss := []prompbmarshal.TimeSeries{newTimeSeriesPB([]float64{1}, []int64{ts.Unix()}, labels)}
+
+	f := func(curRules, newRules, expStaleRules []Rule) {
+		t.Helper()
+		e := &executor{
+			previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
+		}
+		// seed executor with series for
+		// current rules
+		for _, rule := range curRules {
+			e.getStaleSeries(rule, tss, ts)
+		}
+
+		e.purgeStaleSeries(newRules)
+
+		if len(e.previouslySentSeriesToRW) != len(expStaleRules) {
+			t.Fatalf("expected to get %d stale series, got %d",
+				len(expStaleRules), len(e.previouslySentSeriesToRW))
+		}
+
+		for _, exp := range expStaleRules {
+			if _, ok := e.previouslySentSeriesToRW[exp.ID()]; !ok {
+				t.Fatalf("expected to have rule %d; got nil instead", exp.ID())
+			}
+		}
+	}
+
+	f(nil, nil, nil)
+	f(
+		nil,
+		[]Rule{&AlertingRule{RuleID: 1}},
+		nil,
+	)
+	f(
+		[]Rule{&AlertingRule{RuleID: 1}},
+		nil,
+		nil,
+	)
+	f(
+		[]Rule{&AlertingRule{RuleID: 1}},
+		[]Rule{&AlertingRule{RuleID: 2}},
+		nil,
+	)
+	f(
+		[]Rule{&AlertingRule{RuleID: 1}, &AlertingRule{RuleID: 2}},
+		[]Rule{&AlertingRule{RuleID: 2}},
+		[]Rule{&AlertingRule{RuleID: 2}},
+	)
+	f(
+		[]Rule{&AlertingRule{RuleID: 1}, &AlertingRule{RuleID: 2}},
+		[]Rule{&AlertingRule{RuleID: 1}, &AlertingRule{RuleID: 2}},
+		[]Rule{&AlertingRule{RuleID: 1}, &AlertingRule{RuleID: 2}},
+	)
+}
+
 func TestFaultyNotifier(t *testing.T) {
 	fq := &datasource.FakeQuerier{}
 	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
@@ -431,7 +580,8 @@ func TestFaultyRW(t *testing.T) {
 	}
 
 	e := &executor{
-		Rw: &remotewrite.Client{},
+		Rw:                       &remotewrite.Client{},
+		previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
 	}
 
 	err := e.exec(context.Background(), r, time.Now(), 0, 10)

app/vmalert/rule/group_timing_test.go

@@ -0,0 +1,36 @@
+package rule
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+)
+
+func BenchmarkGetStaleSeries(b *testing.B) {
+	ts := time.Now()
+	n := 100
+	payload := make([]prompbmarshal.TimeSeries, 0, n)
+	for i := 0; i < n; i++ {
+		s := fmt.Sprintf("%d", i)
+		labels := toPromLabels(b,
+			"__name__", "foo", ""+
+				"instance", s,
+			"job", s,
+			"state", s,
+		)
+		payload = append(payload, newTimeSeriesPB([]float64{1}, []int64{ts.Unix()}, labels))
+	}
+
+	e := &executor{
+		previouslySentSeriesToRW: make(map[uint64]map[string][]prompbmarshal.Label),
+	}
+	ar := &AlertingRule{RuleID: 1}
+
+	b.ResetTimer()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		e.getStaleSeries(ar, payload, ts)
+	}
+}

app/vmalert/rule/recording.go

@@ -3,17 +3,16 @@ package rule
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
 // RecordingRule is a Rule that supposed
@@ -35,8 +34,6 @@ type RecordingRule struct {
 	// during evaluations
 	state *ruleState
 
-	lastEvaluation map[string]struct{}
-
 	metrics *recordingRuleMetrics
 }
 
@@ -116,7 +113,7 @@ func (rr *RecordingRule) execRange(ctx context.Context, start, end time.Time) ([
 	var tss []prompbmarshal.TimeSeries
 	for _, s := range res.Data {
 		ts := rr.toTimeSeries(s)
-		key := stringifyLabels(ts.Labels)
+		key := stringifyLabels(ts)
 		if _, ok := duplicates[key]; ok {
 			return nil, fmt.Errorf("original metric %v; resulting labels %q: %w", s.Labels, key, errDuplicate)
 		}
@@ -158,47 +155,28 @@ func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]p
 		return nil, curState.Err
 	}
 
-	curEvaluation := make(map[string]struct{}, len(qMetrics))
-	lastEvaluation := rr.lastEvaluation
+	duplicates := make(map[string]struct{}, len(qMetrics))
 	var tss []prompbmarshal.TimeSeries
 	for _, r := range qMetrics {
 		ts := rr.toTimeSeries(r)
-		key := stringifyLabels(ts.Labels)
-		if _, ok := curEvaluation[key]; ok {
+		key := stringifyLabels(ts)
+		if _, ok := duplicates[key]; ok {
 			curState.Err = fmt.Errorf("original metric %v; resulting labels %q: %w", r, key, errDuplicate)
 			return nil, curState.Err
 		}
-		curEvaluation[key] = struct{}{}
-		delete(lastEvaluation, key)
+		duplicates[key] = struct{}{}
 		tss = append(tss, ts)
 	}
-	// check for stale time series
-	for k := range lastEvaluation {
-		tss = append(tss, prompbmarshal.TimeSeries{
-			Labels: stringToLabels(k),
-			Samples: []prompbmarshal.Sample{
-				{Value: decimal.StaleNaN, Timestamp: ts.UnixNano() / 1e6},
-			}})
-	}
-	rr.lastEvaluation = curEvaluation
 	return tss, nil
 }
 
-func stringToLabels(s string) []prompbmarshal.Label {
-	labels := strings.Split(s, ",")
-	rLabels := make([]prompbmarshal.Label, 0, len(labels))
-	for i := range labels {
-		if label := strings.Split(labels[i], "="); len(label) == 2 {
-			rLabels = append(rLabels, prompbmarshal.Label{
-				Name:  label[0],
-				Value: label[1],
-			})
-		}
+func stringifyLabels(ts prompbmarshal.TimeSeries) string {
+	labels := ts.Labels
+	if len(labels) > 1 {
+		sort.Slice(labels, func(i, j int) bool {
+			return labels[i].Name < labels[j].Name
+		})
 	}
-	return rLabels
-}
-
-func stringifyLabels(labels []prompbmarshal.Label) string {
 	b := strings.Builder{}
 	for i, l := range labels {
 		b.WriteString(l.Name)
@@ -212,27 +190,19 @@ func stringifyLabels(labels []prompbmarshal.Label) string {
 }
 
 func (rr *RecordingRule) toTimeSeries(m datasource.Metric) prompbmarshal.TimeSeries {
-	if preN := promrelabel.GetLabelByName(m.Labels, "__name__"); preN != nil {
-		preN.Value = rr.Name
-	} else {
-		m.Labels = append(m.Labels, prompbmarshal.Label{
-			Name:  "__name__",
-			Value: rr.Name,
-		})
+	labels := make(map[string]string)
+	for _, l := range m.Labels {
+		labels[l.Name] = l.Value
 	}
-	for k := range rr.Labels {
-		prevLabel := promrelabel.GetLabelByName(m.Labels, k)
-		if prevLabel != nil && prevLabel.Value != rr.Labels[k] {
-			// Rename the prevLabel to "exported_" + label.Name
-			prevLabel.Name = fmt.Sprintf("exported_%s", prevLabel.Name)
+	labels["__name__"] = rr.Name
+	// override existing labels with configured ones
+	for k, v := range rr.Labels {
+		if _, ok := labels[k]; ok && labels[k] != v {
+			labels[fmt.Sprintf("exported_%s", k)] = labels[k]
 		}
-		m.Labels = append(m.Labels, prompbmarshal.Label{
-			Name:  k,
-			Value: rr.Labels[k],
-		})
+		labels[k] = v
 	}
-	ts := newTimeSeries(m.Values, m.Timestamps, m.Labels)
-	return ts
+	return newTimeSeries(m.Values, m.Timestamps, labels)
 }
 
 // updateWith copies all significant fields.

app/vmalert/rule/recording_test.go

@@ -9,131 +9,59 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
 func TestRecordingRule_Exec(t *testing.T) {
-	ts, _ := time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
-	const defaultStep = 5 * time.Millisecond
-
-	f := func(rule *RecordingRule, steps [][]datasource.Metric, tssExpected [][]prompbmarshal.TimeSeries) {
+	f := func(rule *RecordingRule, metrics []datasource.Metric, tssExpected []prompbmarshal.TimeSeries) {
 		t.Helper()
 
 		fq := &datasource.FakeQuerier{}
-		for i, step := range steps {
-			fq.Reset()
-			fq.Add(step...)
-			rule.q = fq
-			rule.state = &ruleState{
-				entries: make([]StateEntry, 10),
-			}
-			tss, err := rule.exec(context.TODO(), ts, 0)
-			if err != nil {
-				t.Fatalf("fail to test rule %s: unexpected error: %s", rule.Name, err)
-			}
-			if err := compareTimeSeries(t, tssExpected[i], tss); err != nil {
-				t.Fatalf("fail to test rule %s: time series mismatch on step %d: %s", rule.Name, i, err)
-			}
-
-			ts = ts.Add(defaultStep)
+		fq.Add(metrics...)
+		rule.q = fq
+		rule.state = &ruleState{
+			entries: make([]StateEntry, 10),
+		}
+		tss, err := rule.exec(context.TODO(), time.Now(), 0)
+		if err != nil {
+			t.Fatalf("unexpected RecordingRule.exec error: %s", err)
+		}
+		if err := compareTimeSeries(t, tssExpected, tss); err != nil {
+			t.Fatalf("timeseries missmatch: %s", err)
 		}
 	}
 
+	timestamp := time.Now()
+
 	f(&RecordingRule{
 		Name: "foo",
-	}, [][]datasource.Metric{{
+	}, []datasource.Metric{
 		metricWithValueAndLabels(t, 10, "__name__", "bar"),
-	}}, [][]prompbmarshal.TimeSeries{{
-		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: "foo",
-			},
+	}, []prompbmarshal.TimeSeries{
+		newTimeSeries([]float64{10}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "foo",
 		}),
-	}})
+	})
 
 	f(&RecordingRule{
 		Name: "foobarbaz",
-	}, [][]datasource.Metric{
-		{
-			metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "foo"),
-			metricWithValueAndLabels(t, 2, "__name__", "bar", "job", "bar"),
-		},
-		{
-			metricWithValueAndLabels(t, 10, "__name__", "foo", "job", "foo"),
-		},
-		{
-			metricWithValueAndLabels(t, 10, "__name__", "foo", "job", "bar"),
-		},
-	}, [][]prompbmarshal.TimeSeries{
-		{
-			newTimeSeries([]float64{1}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "foo",
-				},
-			}),
-			newTimeSeries([]float64{2}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "bar",
-				},
-			}),
-		},
-		{
-			newTimeSeries([]float64{10}, []int64{ts.Add(defaultStep).UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "foo",
-				},
-			}),
-			// stale time series
-			newTimeSeries([]float64{decimal.StaleNaN}, []int64{ts.Add(defaultStep).UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "bar",
-				},
-			}),
-		},
-		{
-			newTimeSeries([]float64{10}, []int64{ts.Add(2 * defaultStep).UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "bar",
-				},
-			}),
-			newTimeSeries([]float64{decimal.StaleNaN}, []int64{ts.Add(2 * defaultStep).UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "foo",
-				},
-			}),
-		},
+	}, []datasource.Metric{
+		metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "foo"),
+		metricWithValueAndLabels(t, 2, "__name__", "bar", "job", "bar"),
+		metricWithValueAndLabels(t, 3, "__name__", "baz", "job", "baz"),
+	}, []prompbmarshal.TimeSeries{
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "foobarbaz",
+			"job":      "foo",
+		}),
+		newTimeSeries([]float64{2}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "foobarbaz",
+			"job":      "bar",
+		}),
+		newTimeSeries([]float64{3}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "foobarbaz",
+			"job":      "baz",
+		}),
 	})
 
 	f(&RecordingRule{
@@ -141,44 +69,22 @@ func TestRecordingRule_Exec(t *testing.T) {
 		Labels: map[string]string{
 			"source": "test",
 		},
-	}, [][]datasource.Metric{{
+	}, []datasource.Metric{
 		metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "foo"),
 		metricWithValueAndLabels(t, 1, "__name__", "bar", "job", "bar", "source", "origin"),
-	}}, [][]prompbmarshal.TimeSeries{{
-		newTimeSeries([]float64{2}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: "job:foo",
-			},
-			{
-				Name:  "job",
-				Value: "foo",
-			},
-			{
-				Name:  "source",
-				Value: "test",
-			},
+	}, []prompbmarshal.TimeSeries{
+		newTimeSeries([]float64{2}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "job:foo",
+			"job":      "foo",
+			"source":   "test",
 		}),
-		newTimeSeries([]float64{1}, []int64{ts.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "job:foo",
-				},
-				{
-					Name:  "job",
-					Value: "bar",
-				},
-				{
-					Name:  "source",
-					Value: "test",
-				},
-				{
-					Name:  "exported_source",
-					Value: "origin",
-				},
-			}),
-	}})
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__":        "job:foo",
+			"job":             "bar",
+			"source":          "test",
+			"exported_source": "origin",
+		}),
+	})
 }
 
 func TestRecordingRule_ExecRange(t *testing.T) {
@@ -204,13 +110,9 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 	}, []datasource.Metric{
 		metricWithValuesAndLabels(t, []float64{10, 20, 30}, "__name__", "bar"),
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{10, 20, 30}, []int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foo",
-				},
-			}),
+		newTimeSeries([]float64{10, 20, 30}, []int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()}, map[string]string{
+			"__name__": "foo",
+		}),
 	})
 
 	f(&RecordingRule{
@@ -220,36 +122,18 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 		metricWithValuesAndLabels(t, []float64{2, 3}, "__name__", "bar", "job", "bar"),
 		metricWithValuesAndLabels(t, []float64{4, 5, 6}, "__name__", "baz", "job", "baz"),
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: "foobarbaz",
-			},
-			{
-				Name:  "job",
-				Value: "foo",
-			},
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "foobarbaz",
+			"job":      "foo",
 		}),
-		newTimeSeries([]float64{2, 3}, []int64{timestamp.UnixNano(), timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: "foobarbaz",
-			},
-			{
-				Name:  "job",
-				Value: "bar",
-			},
+		newTimeSeries([]float64{2, 3}, []int64{timestamp.UnixNano(), timestamp.UnixNano()}, map[string]string{
+			"__name__": "foobarbaz",
+			"job":      "bar",
 		}),
 		newTimeSeries([]float64{4, 5, 6},
-			[]int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()}, []prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "foobarbaz",
-				},
-				{
-					Name:  "job",
-					Value: "baz",
-				},
+			[]int64{timestamp.UnixNano(), timestamp.UnixNano(), timestamp.UnixNano()}, map[string]string{
+				"__name__": "foobarbaz",
+				"job":      "baz",
 			}),
 	})
 
@@ -262,35 +146,16 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 		metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "foo"),
 		metricWithValueAndLabels(t, 1, "__name__", "bar", "job", "bar"),
 	}, []prompbmarshal.TimeSeries{
-		newTimeSeries([]float64{2}, []int64{timestamp.UnixNano()}, []prompbmarshal.Label{
-			{
-				Name:  "__name__",
-				Value: "job:foo",
-			},
-			{
-				Name:  "job",
-				Value: "foo",
-			},
-			{
-				Name:  "source",
-				Value: "test",
-			},
+		newTimeSeries([]float64{2}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "job:foo",
+			"job":      "foo",
+			"source":   "test",
+		}),
+		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()}, map[string]string{
+			"__name__": "job:foo",
+			"job":      "bar",
+			"source":   "test",
 		}),
-		newTimeSeries([]float64{1}, []int64{timestamp.UnixNano()},
-			[]prompbmarshal.Label{
-				{
-					Name:  "__name__",
-					Value: "job:foo",
-				},
-				{
-					Name:  "job",
-					Value: "bar",
-				},
-				{
-					Name:  "source",
-					Value: "test",
-				},
-			}),
 	})
 }
 

app/vmalert/rule/test_helpers.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
@@ -88,7 +87,7 @@ func metricWithLabels(t *testing.T, labels ...string) datasource.Metric {
 	}
 	m := datasource.Metric{Values: []float64{1}, Timestamps: []int64{1}}
 	for i := 0; i < len(labels); i += 2 {
-		m.Labels = append(m.Labels, prompbmarshal.Label{
+		m.Labels = append(m.Labels, datasource.Label{
 			Name:  labels[i],
 			Value: labels[i+1],
 		})
@@ -96,6 +95,21 @@ func metricWithLabels(t *testing.T, labels ...string) datasource.Metric {
 	return m
 }
 
+func toPromLabels(t testing.TB, labels ...string) []prompbmarshal.Label {
+	t.Helper()
+	if len(labels) == 0 || len(labels)%2 != 0 {
+		t.Fatalf("expected to get even number of labels")
+	}
+	var ls []prompbmarshal.Label
+	for i := 0; i < len(labels); i += 2 {
+		ls = append(ls, prompbmarshal.Label{
+			Name:  labels[i],
+			Value: labels[i+1],
+		})
+	}
+	return ls
+}
+
 func compareTimeSeries(t *testing.T, a, b []prompbmarshal.TimeSeries) error {
 	t.Helper()
 	if len(a) != len(b) {
@@ -108,7 +122,7 @@ func compareTimeSeries(t *testing.T, a, b []prompbmarshal.TimeSeries) error {
 		}
 		for i, exp := range expTS.Samples {
 			got := gotTS.Samples[i]
-			if got.Value != exp.Value && (!decimal.IsStaleNaN(got.Value) || !decimal.IsStaleNaN(exp.Value)) {
+			if got.Value != exp.Value {
 				return fmt.Errorf("expected value %.2f; got %.2f", exp.Value, got.Value)
 			}
 			// timestamp validation isn't always correct for now.

app/vmalert/rule/utils.go

@@ -9,14 +9,10 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
-// newTimeSeries first sorts given labels, then returns new time series.
-func newTimeSeries(values []float64, timestamps []int64, labels []prompbmarshal.Label) prompbmarshal.TimeSeries {
-	promrelabel.SortLabels(labels)
+func newTimeSeries(values []float64, timestamps []int64, labels map[string]string) prompbmarshal.TimeSeries {
 	ts := prompbmarshal.TimeSeries{
-		Labels:  labels,
 		Samples: make([]prompbmarshal.Sample, len(values)),
 	}
 	for i := range values {
@@ -25,6 +21,34 @@ func newTimeSeries(values []float64, timestamps []int64, labels []prompbmarshal.
 			Timestamp: time.Unix(timestamps[i], 0).UnixNano() / 1e6,
 		}
 	}
+	keys := make([]string, 0, len(labels))
+	for k := range labels {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys) // make order deterministic
+	for _, key := range keys {
+		ts.Labels = append(ts.Labels, prompbmarshal.Label{
+			Name:  key,
+			Value: labels[key],
+		})
+	}
+	return ts
+}
+
+// newTimeSeriesPB creates prompbmarshal.TimeSeries with given
+// values, timestamps and labels.
+// It expects that labels are already sorted.
+func newTimeSeriesPB(values []float64, timestamps []int64, labels []prompbmarshal.Label) prompbmarshal.TimeSeries {
+	ts := prompbmarshal.TimeSeries{
+		Samples: make([]prompbmarshal.Sample, len(values)),
+	}
+	for i := range values {
+		ts.Samples[i] = prompbmarshal.Sample{
+			Value:     values[i],
+			Timestamp: time.Unix(timestamps[i], 0).UnixNano() / 1e6,
+		}
+	}
+	ts.Labels = labels
 	return ts
 }
 

app/vmalert/templates/template.go

@@ -54,9 +54,10 @@ func newTemplate() *textTpl.Template {
 }
 
 // Load func loads templates from multiple globs specified in pathPatterns and either
-// sets them directly to current template if it's the first init;
-// or sets replacement templates and wait for Reload() to replace current template with replacement.
-func Load(pathPatterns []string, externalURL url.URL) error {
+// sets them directly to current template if it's undefined or with overwrite=true
+// or sets replacement templates and adds templates with new names to a current
+func Load(pathPatterns []string, overwrite bool) error {
+	var err error
 	tmpl := newTemplate()
 	for _, tp := range pathPatterns {
 		p, err := doublestar.FilepathGlob(tp)
@@ -78,12 +79,36 @@ func Load(pathPatterns []string, externalURL url.URL) error {
 	}
 	tplMu.Lock()
 	defer tplMu.Unlock()
-	tmpl = tmpl.Funcs(funcsWithExternalURL(externalURL))
-
-	if masterTmpl.current == nil {
-		masterTmpl.current = tmpl
+	if masterTmpl.current == nil || overwrite {
+		masterTmpl.replacement = nil
+		masterTmpl.current = newTemplate()
 	} else {
-		masterTmpl.replacement = tmpl
+		masterTmpl.replacement = newTemplate()
+		if err = copyTemplates(tmpl, masterTmpl.replacement, overwrite); err != nil {
+			return err
+		}
+	}
+	return copyTemplates(tmpl, masterTmpl.current, overwrite)
+}
+
+func copyTemplates(from *textTpl.Template, to *textTpl.Template, overwrite bool) error {
+	if from == nil {
+		return nil
+	}
+	if to == nil {
+		to = newTemplate()
+	}
+	tmpl, err := from.Clone()
+	if err != nil {
+		return err
+	}
+	for _, t := range tmpl.Templates() {
+		if to.Lookup(t.Name()) == nil || overwrite {
+			to, err = to.AddParseTree(t.Name(), t.Tree)
+			if err != nil {
+				return fmt.Errorf("failed to add template %q: %w", t.Name(), err)
+			}
+		}
 	}
 	return nil
 }
@@ -128,6 +153,13 @@ func datasourceMetricsToTemplateMetrics(ms []datasource.Metric) []metric {
 // for templating functions.
 type QueryFn func(query string) ([]datasource.Metric, error)
 
+// UpdateWithFuncs updates existing or sets a new function map for a template
+func UpdateWithFuncs(funcs textTpl.FuncMap) {
+	tplMu.Lock()
+	defer tplMu.Unlock()
+	masterTmpl.current = masterTmpl.current.Funcs(funcs)
+}
+
 // GetWithFuncs returns a copy of current template with additional FuncMap
 // provided with funcs argument
 func GetWithFuncs(funcs textTpl.FuncMap) (*textTpl.Template, error) {
@@ -137,11 +169,16 @@ func GetWithFuncs(funcs textTpl.FuncMap) (*textTpl.Template, error) {
 	if err != nil {
 		return nil, err
 	}
-	// Clone() doesn't copy tpl Options, so we set them manually
-	tmpl = tmpl.Option("missingkey=zero")
 	return tmpl.Funcs(funcs), nil
 }
 
+// Get returns a copy of a template
+func Get() (*textTpl.Template, error) {
+	tplMu.RLock()
+	defer tplMu.RUnlock()
+	return masterTmpl.current.Clone()
+}
+
 // FuncsWithQuery returns a function map that depends on metric data
 func FuncsWithQuery(query QueryFn) textTpl.FuncMap {
 	return textTpl.FuncMap{
@@ -159,8 +196,8 @@ func FuncsWithQuery(query QueryFn) textTpl.FuncMap {
 	}
 }
 
-// funcsWithExternalURL returns a function map that depends on externalURL value
-func funcsWithExternalURL(externalURL url.URL) textTpl.FuncMap {
+// FuncsWithExternalURL returns a function map that depends on externalURL value
+func FuncsWithExternalURL(externalURL *url.URL) textTpl.FuncMap {
 	return textTpl.FuncMap{
 		"externalURL": func() string {
 			return externalURL.String()

app/vmalert/templates/template_test.go

@@ -2,7 +2,6 @@ package templates
 
 import (
 	"math"
-	"net/url"
 	"strings"
 	"testing"
 	textTpl "text/template"
@@ -153,7 +152,7 @@ func TestTemplatesLoad_Failure(t *testing.T) {
 	f := func(pathPatterns []string, expectedErrStr string) {
 		t.Helper()
 
-		err := Load(pathPatterns, url.URL{})
+		err := Load(pathPatterns, false)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -172,17 +171,128 @@ func TestTemplatesLoad_Failure(t *testing.T) {
 }
 
 func TestTemplatesLoad_Success(t *testing.T) {
-	f := func(pathPatterns []string, expectedTmpl textTemplate) {
+	f := func(initialTmpl textTemplate, pathPatterns []string, overwrite bool, expectedTmpl textTemplate) {
 		t.Helper()
 
 		masterTmplOrig := masterTmpl
+		masterTmpl = initialTmpl
 		defer func() {
 			masterTmpl = masterTmplOrig
 		}()
 
-		if err := Load(pathPatterns, url.URL{}); err != nil {
+		if err := Load(pathPatterns, overwrite); err != nil {
 			t.Fatalf("cannot load templates: %s", err)
 		}
+
+		if !equalTemplates(masterTmpl.replacement, expectedTmpl.replacement) {
+			t.Fatalf("unexpected replacement template\ngot\n%+v\nwant\n%+v", masterTmpl.replacement, expectedTmpl.replacement)
+		}
+		if !equalTemplates(masterTmpl.current, expectedTmpl.current) {
+			t.Fatalf("unexpected current template\ngot\n%+v\nwant\n%+v", masterTmpl.current, expectedTmpl.current)
+		}
+	}
+
+	// non existing path undefined template override
+	initialTmpl := mkTemplate(nil, nil)
+	pathPatterns := []string{
+		"templates/non-existing/good-*.tpl",
+		"templates/absent/good-*.tpl",
+	}
+	overwrite := true
+	expectedTmpl := mkTemplate(``, nil)
+	f(initialTmpl, pathPatterns, overwrite, expectedTmpl)
+
+	// non existing path defined template override
+	initialTmpl = mkTemplate(`
+		{{- define "test.1" -}}
+			{{- printf "value" -}}
+		{{- end -}}
+	`, nil)
+	pathPatterns = []string{
+		"templates/non-existing/good-*.tpl",
+		"templates/absent/good-*.tpl",
+	}
+	overwrite = true
+	expectedTmpl = mkTemplate(``, nil)
+	f(initialTmpl, pathPatterns, overwrite, expectedTmpl)
+
+	// existing path undefined template override
+	initialTmpl = mkTemplate(nil, nil)
+	pathPatterns = []string{
+		"templates/other/nested/good0-*.tpl",
+		"templates/test/good0-*.tpl",
+	}
+	overwrite = false
+	expectedTmpl = mkTemplate(`
+		{{- define "good0-test.tpl" -}}{{- end -}}
+		{{- define "test.0" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.1" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.2" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.3" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+	`, nil)
+	f(initialTmpl, pathPatterns, overwrite, expectedTmpl)
+
+	// existing path defined template override
+	initialTmpl = mkTemplate(`
+		{{- define "test.1" -}}
+			{{ printf "Hello %s!" "world" }}
+		{{- end -}}
+	`, nil)
+	pathPatterns = []string{
+		"templates/other/nested/good0-*.tpl",
+		"templates/test/good0-*.tpl",
+	}
+	overwrite = false
+	expectedTmpl = mkTemplate(`
+		{{- define "good0-test.tpl" -}}{{- end -}}
+		{{- define "test.0" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.1" -}}
+			{{ printf "Hello %s!" "world" }}
+		{{- end -}}
+		{{- define "test.2" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.3" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+	`, `
+		{{- define "good0-test.tpl" -}}{{- end -}}
+		{{- define "test.0" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.1" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.2" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+		{{- define "test.3" -}}
+			{{ printf "Hello %s!" externalURL }}
+		{{- end -}}
+	`)
+	f(initialTmpl, pathPatterns, overwrite, expectedTmpl)
+}
+
+func TestTemplatesReload(t *testing.T) {
+	f := func(initialTmpl, expectedTmpl textTemplate) {
+		t.Helper()
+
+		masterTmplOrig := masterTmpl
+		masterTmpl = initialTmpl
+		defer func() {
+			masterTmpl = masterTmplOrig
+		}()
+
 		Reload()
 
 		if !equalTemplates(masterTmpl.replacement, expectedTmpl.replacement) {
@@ -193,47 +303,46 @@ func TestTemplatesLoad_Success(t *testing.T) {
 		}
 	}
 
-	// non existing path
-	pathPatterns := []string{
-		"templates/non-existing/good-*.tpl",
-		"templates/absent/good-*.tpl",
-	}
-	expectedTmpl := mkTemplate(``, nil)
-	f(pathPatterns, expectedTmpl)
+	// empty current and replacement templates
+	f(mkTemplate(nil, nil), mkTemplate(nil, nil))
 
-	// existing path
-	pathPatterns = []string{
-		"templates/test/good0-*.tpl",
-	}
-	expectedTmpl = mkTemplate(`
-		{{- define "good0-test.tpl" -}}{{- end -}}
-		{{- define "test.0" -}}
-			{{ printf "Hello %s!" externalURL }}
+	// empty current template only
+	f(mkTemplate(`
+		{{- define "test.1" -}}
+			{{- printf "value" -}}
 		{{- end -}}
-		{{- define "test.2" -}}
-			{{ printf "Hello %s!" externalURL }}
+	`, nil), mkTemplate(`
+		{{- define "test.1" -}}
+			{{- printf "value" -}}
 		{{- end -}}
-		{{- define "test.3" -}}
-			{{ printf "Hello %s!" externalURL }}
-		{{- end -}}
-	`, nil)
-	f(pathPatterns, expectedTmpl)
+	`, nil))
 
-	// existing path defined template override
-	pathPatterns = []string{
-		"templates/other/nested/good0-*.tpl",
-	}
-	expectedTmpl = mkTemplate(`
-		{{- define "good0-test.tpl" -}}{{- end -}}
+	// empty replacement template only
+	f(mkTemplate(nil, `
+		{{- define "test.1" -}}
+			{{- printf "value" -}}
+		{{- end -}}
+	`), mkTemplate(`
+		{{- define "test.1" -}}
+			{{- printf "value" -}}
+		{{- end -}}
+	`, nil))
+
+	// defined both templates
+	f(mkTemplate(`
 		{{- define "test.0" -}}
-			{{ printf "Hello %s!" externalURL }}
+			{{- printf "value" -}}
 		{{- end -}}
 		{{- define "test.1" -}}
-			{{ printf "Hello %s!" externalURL }}
+			{{- printf "before" -}}
 		{{- end -}}
-		{{- define "test.3" -}}
-			{{ printf "Hello %s!" externalURL }}
+	`, `
+		{{- define "test.1" -}}
+			{{- printf "after" -}}
 		{{- end -}}
-	`, nil)
-	f(pathPatterns, expectedTmpl)
+	`), mkTemplate(`
+		{{- define "test.1" -}}
+			{{- printf "after" -}}
+		{{- end -}}
+	`, nil))
 }

app/vmalert/web_types_test.go

@@ -2,12 +2,11 @@ package main
 
 import (
 	"fmt"
-	"reflect"
-	"testing"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/rule"
+	"reflect"
+	"testing"
 )
 
 func TestRecordingToApi(t *testing.T) {

app/vmauth/auth_config.go

@@ -67,7 +67,6 @@ type UserInfo struct {
 	URLPrefix              *URLPrefix  `yaml:"url_prefix,omitempty"`
 	DiscoverBackendIPs     *bool       `yaml:"discover_backend_ips,omitempty"`
 	URLMaps                []URLMap    `yaml:"url_map,omitempty"`
-	DumpRequestOnErrors    bool        `yaml:"dump_request_on_errors,omitempty"`
 	HeadersConf            HeadersConf `yaml:",inline"`
 	MaxConcurrentRequests  int         `yaml:"max_concurrent_requests,omitempty"`
 	DefaultURL             *URLPrefix  `yaml:"default_url,omitempty"`
@@ -463,12 +462,17 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 
 	// Slow path - select other backend urls.
 	n := atomicCounter.Add(1) - 1
+	buMin := bus[n%uint32(len(bus))]
 	for i := uint32(0); i < uint32(len(bus)); i++ {
 		idx := (n + i) % uint32(len(bus))
 		bu := bus[idx]
 		if bu.isBroken() {
 			continue
 		}
+		if buMin.isBroken() {
+			// verify that buMin isn't set as broken
+			buMin = bu
+		}
 		if bu.concurrentRequests.Load() == 0 {
 			// Fast path - return the backend with zero concurrently executed requests.
 			// Do not use CompareAndSwap() instead of Load(), since it is much slower on systems with many CPU cores.
@@ -478,13 +482,12 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 	}
 
 	// Slow path - return the backend with the minimum number of concurrently executed requests.
-	buMin := bus[n%uint32(len(bus))]
 	minRequests := buMin.concurrentRequests.Load()
 	for _, bu := range bus {
 		if bu.isBroken() {
 			continue
 		}
-		if n := bu.concurrentRequests.Load(); n < minRequests || buMin.isBroken() {
+		if n := bu.concurrentRequests.Load(); n < minRequests {
 			buMin = bu
 			minRequests = n
 		}
@@ -783,11 +786,10 @@ func parseAuthConfig(data []byte) (*AuthConfig, error) {
 
 func parseAuthConfigUsers(ac *AuthConfig) (map[string]*UserInfo, error) {
 	uis := ac.Users
-	byAuthToken := make(map[string]*UserInfo, len(uis))
 	if len(uis) == 0 && ac.UnauthorizedUser == nil {
-		// fast path for empty configuration
-		return byAuthToken, nil
+		return nil, fmt.Errorf("Missing `users` or `unauthorized_user` sections")
 	}
+	byAuthToken := make(map[string]*UserInfo, len(uis))
 	for i := range uis {
 		ui := &uis[i]
 		ats, err := getAuthTokens(ui.AuthToken, ui.BearerToken, ui.Username, ui.Password)
@@ -862,23 +864,22 @@ func (ui *UserInfo) initURLs() error {
 	loadBalancingPolicy := *defaultLoadBalancingPolicy
 	dropSrcPathPrefixParts := 0
 	discoverBackendIPs := *discoverBackendIPsGlobal
-	if ui.RetryStatusCodes != nil {
-		retryStatusCodes = ui.RetryStatusCodes
-	}
-	if ui.LoadBalancingPolicy != "" {
-		loadBalancingPolicy = ui.LoadBalancingPolicy
-	}
-	if ui.DropSrcPathPrefixParts != nil {
-		dropSrcPathPrefixParts = *ui.DropSrcPathPrefixParts
-	}
-	if ui.DiscoverBackendIPs != nil {
-		discoverBackendIPs = *ui.DiscoverBackendIPs
-	}
-
 	if ui.URLPrefix != nil {
 		if err := ui.URLPrefix.sanitizeAndInitialize(); err != nil {
 			return err
 		}
+		if ui.RetryStatusCodes != nil {
+			retryStatusCodes = ui.RetryStatusCodes
+		}
+		if ui.LoadBalancingPolicy != "" {
+			loadBalancingPolicy = ui.LoadBalancingPolicy
+		}
+		if ui.DropSrcPathPrefixParts != nil {
+			dropSrcPathPrefixParts = *ui.DropSrcPathPrefixParts
+		}
+		if ui.DiscoverBackendIPs != nil {
+			discoverBackendIPs = *ui.DiscoverBackendIPs
+		}
 		ui.URLPrefix.retryStatusCodes = retryStatusCodes
 		ui.URLPrefix.dropSrcPathPrefixParts = dropSrcPathPrefixParts
 		ui.URLPrefix.discoverBackendIPs = discoverBackendIPs

app/vmauth/auth_config_test.go

@@ -24,10 +24,16 @@ func TestParseAuthConfigFailure(t *testing.T) {
 		}
 	}
 
+	// Empty config
+	f(``)
+
 	// Invalid entry
 	f(`foobar`)
 	f(`foobar: baz`)
 
+	// Empty users
+	f(`users: []`)
+
 	// Missing url_prefix
 	f(`
 users:
@@ -296,12 +302,6 @@ func TestParseAuthConfigSuccess(t *testing.T) {
 
 	insecureSkipVerifyTrue := true
 
-	// Empty config
-	f(``, map[string]*UserInfo{})
-
-	// Empty users
-	f(`users: []`, map[string]*UserInfo{})
-
 	// Single user
 	f(`
 users:

app/vmauth/example_config_ent.yml

@@ -22,26 +22,26 @@ users:
   #  - or http://default2:8888/unsupported_url_handler?request_path=/non/existing/path
   #
   # Regular expressions are allowed in `src_paths` entries.
-  - username: "foobar"
+- username: "foobar"
+  url_map:
+  - src_paths:
+    - "/api/v1/query"
+    - "/api/v1/query_range"
+    - "/api/v1/label/[^/]+/values"
+    url_prefix:
+    - "http://vmselect1:8481/select/42/prometheus"
+    - "http://vmselect2:8481/select/42/prometheus"
+  - src_paths: ["/api/v1/write"]
+    url_prefix: "http://vminsert:8480/insert/42/prometheus"
+    headers:
+    - "X-Scope-OrgID: abc"
     ip_filters:
       deny_list: [127.0.0.1]
-    url_map:
-      - src_paths:
-          - "/api/v1/query"
-          - "/api/v1/query_range"
-          - "/api/v1/label/[^/]+/values"
-        url_prefix:
-          - "http://vmselect1:8481/select/42/prometheus"
-          - "http://vmselect2:8481/select/42/prometheus"
-      - src_paths: ["/api/v1/write"]
-        url_prefix: "http://vminsert:8480/insert/42/prometheus"
-        headers:
-          - "X-Scope-OrgID: abc"
-    default_url:
-      - "http://default1:8888/unsupported_url_handler"
-      - "http://default2:8888/unsupported_url_handler"
+  default_url:
+  - "http://default1:8888/unsupported_url_handler"
+  - "http://default2:8888/unsupported_url_handler"
 
 ip_filters:
   allow_list: ["1.2.3.0/24", "127.0.0.1"]
   deny_list:
-    - 10.1.0.1
+  - 10.1.0.1

app/vmauth/main.go

@@ -61,9 +61,6 @@ var (
 		"See https://docs.victoriametrics.com/vmauth/#backend-tls-setup")
 	backendTLSServerName = flag.String("backend.TLSServerName", "", "Optional TLS ServerName, which must be sent to HTTPS backend. "+
 		"See https://docs.victoriametrics.com/vmauth/#backend-tls-setup")
-	dryRun                   = flag.Bool("dryRun", false, "Whether to check only config files without running vmauth. The auth configuration file is validated. The -auth.config flag must be specified.")
-	removeXFFHTTPHeaderValue = flag.Bool(`removeXFFHTTPHeaderValue`, false, "Whether to remove the X-Forwarded-For HTTP header value from client requests before forwarding them to the backend. "+
-		"Recommended when vmauth is exposed to the internet.")
 )
 
 func main() {
@@ -74,16 +71,6 @@ func main() {
 	buildinfo.Init()
 	logger.Init()
 
-	if *dryRun {
-		if len(*authConfigPath) == 0 {
-			logger.Fatalf("missing required `-auth.config` command-line flag")
-		}
-		if _, err := reloadAuthConfig(); err != nil {
-			logger.Fatalf("failed to parse %q: %s", *authConfigPath, err)
-		}
-		return
-	}
-
 	listenAddrs := *httpListenAddrs
 	if len(listenAddrs) == 0 {
 		listenAddrs = []string{":8427"}
@@ -136,12 +123,6 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 
 	ui := getUserInfoByAuthTokens(ats)
 	if ui == nil {
-		uu := authConfig.Load().UnauthorizedUser
-		if uu != nil {
-			processUserRequest(w, r, uu)
-			return true
-		}
-
 		invalidAuthTokenRequests.Inc()
 		if *logInvalidAuthTokens {
 			err := fmt.Errorf("cannot authorize request with auth tokens %q", ats)
@@ -199,7 +180,7 @@ func processUserRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 
 func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 	u := normalizeURL(r.URL)
-	up, hc := ui.getURLPrefixAndHeaders(u, r.Host, r.Header)
+	up, hc := ui.getURLPrefixAndHeaders(u, r.Header)
 	isDefault := false
 	if up == nil {
 		if ui.DefaultURL == nil {
@@ -211,11 +192,7 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 				return
 			}
 			missingRouteRequests.Inc()
-			var di string
-			if ui.DumpRequestOnErrors {
-				di = debugInfo(u, r)
-			}
-			httpserver.Errorf(w, r, "missing route for %q%s", u.String(), di)
+			httpserver.Errorf(w, r, "missing route for %s", u.String())
 			return
 		}
 		up, hc = ui.DefaultURL, ui.HeadersConf
@@ -394,7 +371,7 @@ func sanitizeRequestHeaders(r *http.Request) *http.Request {
 		// X-Forwarded-For information as a comma+space
 		// separated list and fold multiple headers into one.
 		prior := req.Header["X-Forwarded-For"]
-		if len(prior) > 0 && !*removeXFFHTTPHeaderValue {
+		if len(prior) > 0 {
 			clientIP = strings.Join(prior, ", ") + ", " + clientIP
 		}
 		req.Header.Set("X-Forwarded-For", clientIP)
@@ -667,14 +644,3 @@ func (rtb *readTrackingBody) Close() error {
 
 	return nil
 }
-
-func debugInfo(u *url.URL, r *http.Request) string {
-	s := &strings.Builder{}
-	fmt.Fprintf(s, " (host: %q; ", r.Host)
-	fmt.Fprintf(s, "path: %q; ", u.Path)
-	fmt.Fprintf(s, "args: %q; ", u.Query().Encode())
-	fmt.Fprint(s, "headers:")
-	_ = r.Header.WriteSubset(s, nil)
-	fmt.Fprint(s, ")")
-	return s.String()
-}

app/vmauth/main_test.go

@@ -90,20 +90,6 @@ User-Agent: vmauth
 X-Forwarded-For: 12.34.56.78, 42.2.3.84`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 
-	// routing of all failed to authorize requests to unauthorized_user (issue #7543)
-	cfgStr = `
-unauthorized_user:
-  url_prefix: "{BACKEND}/foo"
-  keep_original_host: true`
-	requestURL = "http://foo:invalid-secret@some-host.com/abc/def"
-	backendHandler = func(w http.ResponseWriter, r *http.Request) {
-		fmt.Fprintf(w, "requested_url=http://%s%s", r.Host, r.URL)
-	}
-	responseExpected = `
-statusCode=200
-requested_url=http://some-host.com/foo/abc/def`
-	f(cfgStr, requestURL, backendHandler, responseExpected)
-
 	// keep_original_host
 	cfgStr = `
 unauthorized_user:
@@ -360,27 +346,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=400
-remoteAddr: "42.2.3.84:6789, X-Forwarded-For: 12.34.56.78"; requestURI: /abc?de=fg; missing route for "http://some-host.com/abc?de=fg"`
-	f(cfgStr, requestURL, backendHandler, responseExpected)
-
-	// missing default_url and default url_prefix for unauthorized user with dump_request_on_errors enabled
-	cfgStr = `
-unauthorized_user:
-  dump_request_on_errors: true
-  url_map:
-  - src_paths: ["/foo/.+"]
-    url_prefix: {BACKEND}/x-foo/`
-	requestURL = "http://some-host.com/abc?de=fg"
-	backendHandler = func(_ http.ResponseWriter, _ *http.Request) {
-		panic(fmt.Errorf("backend handler shouldn't be called"))
-	}
-	responseExpected = `
-statusCode=400
-remoteAddr: "42.2.3.84:6789, X-Forwarded-For: 12.34.56.78"; requestURI: /abc?de=fg; missing route for "http://some-host.com/abc?de=fg" (host: "some-host.com"; path: "/abc"; args: "de=fg"; headers:Connection: Some-Header,Other-Header
-Pass-Header: abc
-Some-Header: foobar
-X-Forwarded-For: 12.34.56.78
-)`
+remoteAddr: "42.2.3.84:6789, X-Forwarded-For: 12.34.56.78"; requestURI: /abc?de=fg; missing route for http://some-host.com/abc?de=fg`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 
 	// missing default_url and default url_prefix for unauthorized user when there are configs for authorized users

app/vmauth/target_url.go

@@ -51,9 +51,9 @@ func dropPrefixParts(path string, parts int) string {
 	return path
 }
 
-func (ui *UserInfo) getURLPrefixAndHeaders(u *url.URL, host string, h http.Header) (*URLPrefix, HeadersConf) {
+func (ui *UserInfo) getURLPrefixAndHeaders(u *url.URL, h http.Header) (*URLPrefix, HeadersConf) {
 	for _, e := range ui.URLMaps {
-		if !matchAnyRegex(e.SrcHosts, host) {
+		if !matchAnyRegex(e.SrcHosts, u.Host) {
 			continue
 		}
 		if !matchAnyRegex(e.SrcPaths, u.Path) {

app/vmauth/target_url_test.go

@@ -95,7 +95,7 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 			t.Fatalf("cannot parse %q: %s", requestURI, err)
 		}
 		u = normalizeURL(u)
-		up, hc := ui.getURLPrefixAndHeaders(u, u.Host, nil)
+		up, hc := ui.getURLPrefixAndHeaders(u, nil)
 		if up == nil {
 			t.Fatalf("cannot match available backend: %s", err)
 		}
@@ -187,10 +187,6 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 				RetryStatusCodes:       []int{},
 				DropSrcPathPrefixParts: intp(0),
 			},
-			{
-				SrcPaths:  getRegexs([]string{"/metrics"}),
-				URLPrefix: mustParseURL("http://metrics-server"),
-			},
 		},
 		URLPrefix: mustParseURL("http://default-server"),
 		HeadersConf: HeadersConf{
@@ -210,35 +206,6 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 		"bb: aaa", "x: y", []int{502}, "least_loaded", 2)
 	f(ui, "https://foo-host/api/v1/write", "http://vminsert/0/prometheus/api/v1/write", "", "", []int{}, "least_loaded", 0)
 	f(ui, "https://foo-host/foo/bar/api/v1/query_range", "http://default-server/api/v1/query_range", "bb: aaa", "x: y", []int{502}, "least_loaded", 2)
-	f(ui, "https://foo-host/metrics", "http://metrics-server", "", "", []int{502}, "least_loaded", 2)
-
-	// Complex routing with `url_map` without global url_prefix
-	ui = &UserInfo{
-		URLMaps: []URLMap{
-			{
-				SrcPaths:               getRegexs([]string{"/api/v1/write"}),
-				URLPrefix:              mustParseURL("http://vminsert/0/prometheus"),
-				RetryStatusCodes:       []int{},
-				DropSrcPathPrefixParts: intp(0),
-			},
-			{
-				SrcPaths:  getRegexs([]string{"/metrics/a/b"}),
-				URLPrefix: mustParseURL("http://metrics-server"),
-			},
-		},
-		HeadersConf: HeadersConf{
-			RequestHeaders: []*Header{
-				mustNewHeader("'bb: aaa'"),
-			},
-			ResponseHeaders: []*Header{
-				mustNewHeader("'x: y'"),
-			},
-		},
-		RetryStatusCodes:       []int{502},
-		DropSrcPathPrefixParts: intp(2),
-	}
-	f(ui, "https://foo-host/api/v1/write", "http://vminsert/0/prometheus/api/v1/write", "", "", []int{}, "least_loaded", 0)
-	f(ui, "https://foo-host/metrics/a/b", "http://metrics-server/b", "", "", []int{502}, "least_loaded", 2)
 
 	// Complex routing regexp paths in `url_map`
 	ui = &UserInfo{
@@ -306,7 +273,7 @@ func TestUserInfoGetBackendURL_SRV(t *testing.T) {
 			t.Fatalf("cannot parse %q: %s", requestURI, err)
 		}
 		u = normalizeURL(u)
-		up, _ := ui.getURLPrefixAndHeaders(u, u.Host, nil)
+		up, _ := ui.getURLPrefixAndHeaders(u, nil)
 		if up == nil {
 			t.Fatalf("cannot match available backend: %s", err)
 		}
@@ -384,7 +351,7 @@ func TestUserInfoGetBackendURL_SRVZeroBackends(t *testing.T) {
 			t.Fatalf("cannot parse %q: %s", requestURI, err)
 		}
 		u = normalizeURL(u)
-		up, _ := ui.getURLPrefixAndHeaders(u, u.Host, nil)
+		up, _ := ui.getURLPrefixAndHeaders(u, nil)
 		if up == nil {
 			t.Fatalf("cannot match available backend: %s", err)
 		}
@@ -432,7 +399,7 @@ func TestCreateTargetURLFailure(t *testing.T) {
 			t.Fatalf("cannot parse %q: %s", requestURI, err)
 		}
 		u = normalizeURL(u)
-		up, hc := ui.getURLPrefixAndHeaders(u, u.Host, nil)
+		up, hc := ui.getURLPrefixAndHeaders(u, nil)
 		if up != nil {
 			t.Fatalf("unexpected non-empty up=%#v", up)
 		}

app/vmctl/flags.go

@@ -616,7 +616,7 @@ var (
 		},
 		&cli.BoolFlag{
 			Name: vmNativeDisableBinaryProtocol,
-			Usage: "Whether to use https://docs.victoriametrics.com/#how-to-export-data-in-json-line-format " +
+			Usage: "Whether to use https://docs.victoriametrics.com/#how-to-export-data-in-json-line-format" +
 				"instead of https://docs.victoriametrics.com/#how-to-export-data-in-native-format API." +
 				"Binary export/import API protocol implies less network and resource usage, as it transfers compressed binary data blocks." +
 				"Non-binary export/import API is less efficient, but supports deduplication if it is configured on vm-native-src-addr side.",

app/vmctl/main.go

@@ -266,7 +266,7 @@ func main() {
 			},
 			{
 				Name:   "vm-native",
-				Usage:  "Migrate time series between VictoriaMetrics installations",
+				Usage:  "Migrate time series between VictoriaMetrics installations via native binary format",
 				Flags:  mergeFlags(globalFlags, vmNativeFlags),
 				Before: beforeFn,
 				Action: func(c *cli.Context) error {

app/vmctl/remoteread/remoteread.go

@@ -15,7 +15,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
-	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/prometheus/prometheus/storage/remote"
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
@@ -239,7 +238,7 @@ func processStreamResponse(body io.ReadCloser, callback StreamCallback) error {
 	bb := bbPool.Get()
 	defer func() { bbPool.Put(bb) }()
 
-	stream := remote.NewChunkedReader(body, config.DefaultChunkedReadLimit, bb.B)
+	stream := remote.NewChunkedReader(body, remote.DefaultChunkedReadLimit, bb.B)
 	for {
 		res := &prompb.ChunkedReadResponse{}
 		err := stream.NextProto(res)

app/vmctl/testdata/servers_integration_test/remote_read_server.go

@@ -12,14 +12,10 @@ import (
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/golang/snappy"
-	"github.com/prometheus/prometheus/model/histogram"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/prompb"
-	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/storage/remote"
-	"github.com/prometheus/prometheus/tsdb/chunkenc"
 	"github.com/prometheus/prometheus/tsdb/chunks"
-	"github.com/prometheus/prometheus/util/annotations"
 )
 
 const (
@@ -331,16 +327,13 @@ func NewMockStorage(series []*prompb.TimeSeries) *MockStorage {
 	return &MockStorage{store: series}
 }
 
-func (ms *MockStorage) Read(_ context.Context, query *prompb.Query, sortSeries bool) (storage.SeriesSet, error) {
-	if sortSeries {
-		return nil, fmt.Errorf("unexpected sortSeries=true")
-	}
+func (ms *MockStorage) Read(_ context.Context, query *prompb.Query) (*prompb.QueryResult, error) {
 	if ms.query != nil {
 		return nil, fmt.Errorf("expected only one call to remote client got: %v", query)
 	}
 	ms.query = query
 
-	tss := make([]*prompb.TimeSeries, 0, len(ms.store))
+	q := &prompb.QueryResult{Timeseries: make([]*prompb.TimeSeries, 0, len(ms.store))}
 	for _, s := range ms.store {
 		var samples []prompb.Sample
 		for _, sample := range s.Samples {
@@ -354,109 +347,15 @@ func (ms *MockStorage) Read(_ context.Context, query *prompb.Query, sortSeries b
 			series.Samples = samples
 		}
 
-		tss = append(tss, &series)
+		q.Timeseries = append(q.Timeseries, &series)
 	}
-	return &mockSeriesSet{
-		tss: tss,
-	}, nil
+	return q, nil
 }
 
 func (ms *MockStorage) Reset() {
 	ms.query = nil
 }
 
-type mockSeriesSet struct {
-	tss  []*prompb.TimeSeries
-	next int
-}
-
-func (ss *mockSeriesSet) Next() bool {
-	if ss.next >= len(ss.tss) {
-		return false
-	}
-	ss.next++
-	return true
-}
-
-func (ss *mockSeriesSet) At() storage.Series {
-	return &mockSeries{
-		s: ss.tss[ss.next-1],
-	}
-}
-
-func (ss *mockSeriesSet) Err() error {
-	return nil
-}
-
-func (ss *mockSeriesSet) Warnings() annotations.Annotations {
-	return nil
-}
-
-type mockSeries struct {
-	s *prompb.TimeSeries
-}
-
-func (s *mockSeries) Labels() labels.Labels {
-	a := make(labels.Labels, len(s.s.Labels))
-	for i, label := range s.s.Labels {
-		a[i] = labels.Label{
-			Name:  label.Name,
-			Value: label.Value,
-		}
-	}
-	return a
-}
-
-func (s *mockSeries) Iterator(chunkenc.Iterator) chunkenc.Iterator {
-	return &mockSamplesIterator{
-		samples: s.s.Samples,
-	}
-}
-
-type mockSamplesIterator struct {
-	samples []prompb.Sample
-	next    int
-}
-
-func (si *mockSamplesIterator) Next() chunkenc.ValueType {
-	if si.next >= len(si.samples) {
-		return chunkenc.ValNone
-	}
-	si.next++
-	return chunkenc.ValFloat
-}
-
-func (si *mockSamplesIterator) Seek(t int64) chunkenc.ValueType {
-	for i := range si.samples {
-		if si.samples[i].Timestamp >= t {
-			si.next = i + 1
-			return chunkenc.ValFloat
-		}
-	}
-	return chunkenc.ValNone
-}
-
-func (si *mockSamplesIterator) At() (int64, float64) {
-	s := si.samples[si.next-1]
-	return s.Timestamp, s.Value
-}
-
-func (si *mockSamplesIterator) AtHistogram(*histogram.Histogram) (int64, *histogram.Histogram) {
-	panic("BUG: musn't be called")
-}
-
-func (si *mockSamplesIterator) AtFloatHistogram(*histogram.FloatHistogram) (int64, *histogram.FloatHistogram) {
-	panic("BUG: mustn't be called")
-}
-
-func (si *mockSamplesIterator) AtT() int64 {
-	return si.samples[si.next-1].Timestamp
-}
-
-func (si *mockSamplesIterator) Err() error {
-	return nil
-}
-
 func labelsToLabelsProto(labels labels.Labels) []prompb.Label {
 	result := make([]prompb.Label, 0, len(labels))
 	for _, l := range labels {

app/vmctl/vm_native_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 )
 
@@ -214,15 +214,15 @@ func processFlags() {
 func fillStorage(series []vm.TimeSeries) error {
 	var mrs []storage.MetricRow
 	for _, series := range series {
-		var labels []prompbmarshal.Label
+		var labels []prompb.Label
 		for _, lp := range series.LabelPairs {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  lp.Name,
 				Value: lp.Value,
 			})
 		}
 		if series.Name != "" {
-			labels = append(labels, prompbmarshal.Label{
+			labels = append(labels, prompb.Label{
 				Name:  "__name__",
 				Value: series.Name,
 			})

app/vminsert/common/insert_ctx.go

@@ -4,46 +4,13 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
-)
-
-// StartIngestionRateLimiter starts ingestion rate limiter.
-//
-// Ingestion rate limiter must be started before Init() call.
-//
-// StopIngestionRateLimiter must be called before Stop() call in order to unblock all the callers
-// to ingestion rate limiter. Otherwise deadlock may occur at Stop() call.
-func StartIngestionRateLimiter(maxIngestionRate int) {
-	if maxIngestionRate <= 0 {
-		return
-	}
-	ingestionRateLimitReached := metrics.NewCounter(`vm_max_ingestion_rate_limit_reached_total`)
-	ingestionRateLimiterStopCh = make(chan struct{})
-	ingestionRateLimiter = ratelimiter.New(int64(maxIngestionRate), ingestionRateLimitReached, ingestionRateLimiterStopCh)
-}
-
-// StopIngestionRateLimiter stops ingestion rate limiter.
-func StopIngestionRateLimiter() {
-	if ingestionRateLimiterStopCh == nil {
-		return
-	}
-	close(ingestionRateLimiterStopCh)
-	ingestionRateLimiterStopCh = nil
-}
-
-var (
-	ingestionRateLimiter       *ratelimiter.RateLimiter
-	ingestionRateLimiterStopCh chan struct{}
 )
 
 // InsertCtx contains common bits for data points insertion.
@@ -63,7 +30,7 @@ type InsertCtx struct {
 func (ctx *InsertCtx) Reset(rowsLen int) {
 	labels := ctx.Labels
 	for i := range labels {
-		labels[i] = prompbmarshal.Label{}
+		labels[i] = prompb.Label{}
 	}
 	ctx.Labels = labels[:0]
 
@@ -84,7 +51,7 @@ func cleanMetricRow(mr *storage.MetricRow) {
 	mr.MetricNameRaw = nil
 }
 
-func (ctx *InsertCtx) marshalMetricNameRaw(prefix []byte, labels []prompbmarshal.Label) []byte {
+func (ctx *InsertCtx) marshalMetricNameRaw(prefix []byte, labels []prompb.Label) []byte {
 	start := len(ctx.metricNamesBuf)
 	ctx.metricNamesBuf = append(ctx.metricNamesBuf, prefix...)
 	ctx.metricNamesBuf = storage.MarshalMetricNameRaw(ctx.metricNamesBuf, labels)
@@ -92,38 +59,16 @@ func (ctx *InsertCtx) marshalMetricNameRaw(prefix []byte, labels []prompbmarshal
 	return metricNameRaw[:len(metricNameRaw):len(metricNameRaw)]
 }
 
-// TryPrepareLabels prepares context labels to the ingestion
-//
-// It returns false if timeseries should be skipped
-func (ctx *InsertCtx) TryPrepareLabels(hasRelabeling bool) bool {
-	if hasRelabeling {
-		ctx.ApplyRelabeling()
-	}
-	if len(ctx.Labels) == 0 {
-		return false
-	}
-	if timeserieslimits.Enabled() && timeserieslimits.IsExceeding(ctx.Labels) {
-		return false
-	}
-	ctx.sortLabelsIfNeeded()
-
-	return true
-}
-
 // WriteDataPoint writes (timestamp, value) with the given prefix and labels into ctx buffer.
-//
-// caller should invoke TryPrepareLabels before using this function if needed
-func (ctx *InsertCtx) WriteDataPoint(prefix []byte, labels []prompbmarshal.Label, timestamp int64, value float64) error {
+func (ctx *InsertCtx) WriteDataPoint(prefix []byte, labels []prompb.Label, timestamp int64, value float64) error {
 	metricNameRaw := ctx.marshalMetricNameRaw(prefix, labels)
 	return ctx.addRow(metricNameRaw, timestamp, value)
 }
 
 // WriteDataPointExt writes (timestamp, value) with the given metricNameRaw and labels into ctx buffer.
 //
-// caller must invoke TryPrepareLabels before using this function
-//
 // It returns metricNameRaw for the given labels if len(metricNameRaw) == 0.
-func (ctx *InsertCtx) WriteDataPointExt(metricNameRaw []byte, labels []prompbmarshal.Label, timestamp int64, value float64) ([]byte, error) {
+func (ctx *InsertCtx) WriteDataPointExt(metricNameRaw []byte, labels []prompb.Label, timestamp int64, value float64) ([]byte, error) {
 	if len(metricNameRaw) == 0 {
 		metricNameRaw = ctx.marshalMetricNameRaw(nil, labels)
 	}
@@ -161,7 +106,7 @@ func (ctx *InsertCtx) AddLabelBytes(name, value []byte) {
 		// Do not skip labels with empty name, since they are equal to __name__.
 		return
 	}
-	ctx.Labels = append(ctx.Labels, prompbmarshal.Label{
+	ctx.Labels = append(ctx.Labels, prompb.Label{
 		// Do not copy name and value contents for performance reasons.
 		// This reduces GC overhead on the number of objects and allocations.
 		Name:  bytesutil.ToUnsafeString(name),
@@ -179,7 +124,7 @@ func (ctx *InsertCtx) AddLabel(name, value string) {
 		// Do not skip labels with empty name, since they are equal to __name__.
 		return
 	}
-	ctx.Labels = append(ctx.Labels, prompbmarshal.Label{
+	ctx.Labels = append(ctx.Labels, prompb.Label{
 		// Do not copy name and value contents for performance reasons.
 		// This reduces GC overhead on the number of objects and allocations.
 		Name:  name,
@@ -204,12 +149,9 @@ func (ctx *InsertCtx) FlushBufs() error {
 		}
 		matchIdxsPool.Put(matchIdxs)
 	}
-	ingestionRateLimiter.Register(len(ctx.mrs))
-
 	// There is no need in limiting the number of concurrent calls to vmstorage.AddRows() here,
 	// since the number of concurrent FlushBufs() calls should be already limited via writeconcurrencylimiter
 	// used at every stream.Parse() call under lib/protoparser/*
-
 	err := vmstorage.AddRows(ctx.mrs)
 	ctx.Reset(0)
 	if err == nil {

app/vminsert/common/sort_labels.go

@@ -4,7 +4,7 @@ import (
 	"flag"
 	"sort"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
 var sortLabels = flag.Bool("sortLabels", false, `Whether to sort labels for incoming samples before writing them to storage. `+
@@ -12,14 +12,14 @@ var sortLabels = flag.Bool("sortLabels", false, `Whether to sort labels for inco
 	`For example, if m{k1="v1",k2="v2"} may be sent as m{k2="v2",k1="v1"}. `+
 	`Enabled sorting for labels can slow down ingestion performance a bit`)
 
-// sortLabelsIfNeeded sorts labels if -sortLabels command-line flag is set
-func (ctx *InsertCtx) sortLabelsIfNeeded() {
+// SortLabelsIfNeeded sorts labels if -sortLabels command-line flag is set
+func (ctx *InsertCtx) SortLabelsIfNeeded() {
 	if *sortLabels {
 		sort.Sort(&ctx.Labels)
 	}
 }
 
-type sortedLabels []prompbmarshal.Label
+type sortedLabels []prompb.Label
 
 func (sl *sortedLabels) Len() int { return len(*sl) }
 func (sl *sortedLabels) Less(i, j int) bool {

app/vminsert/csvimport/request_handler.go

@@ -46,9 +46,14 @@ func insertRows(rows []parser.Row, extraLabels []prompbmarshal.Label) error {
 			label := &extraLabels[j]
 			ctx.AddLabel(label.Name, label.Value)
 		}
-		if !ctx.TryPrepareLabels(hasRelabeling) {
+		if hasRelabeling {
+			ctx.ApplyRelabeling()
+		}
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
 			continue
 		}
+		ctx.SortLabelsIfNeeded()
 		if err := ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value); err != nil {
 			return err
 		}

app/vminsert/datadogsketches/request_handler.go

@@ -60,9 +60,14 @@ func insertRows(sketches []*datadogsketches.Sketch, extraLabels []prompbmarshal.
 				label := &extraLabels[j]
 				ctx.AddLabel(label.Name, label.Value)
 			}
-			if !ctx.TryPrepareLabels(hasRelabeling) {
+			if hasRelabeling {
+				ctx.ApplyRelabeling()
+			}
+			if len(ctx.Labels) == 0 {
+				// Skip metric without labels.
 				continue
 			}
+			ctx.SortLabelsIfNeeded()
 			var metricNameRaw []byte
 			var err error
 			for _, p := range m.Points {

app/vminsert/datadogv1/request_handler.go

@@ -63,9 +63,14 @@ func insertRows(series []datadogv1.Series, extraLabels []prompbmarshal.Label) er
 			label := &extraLabels[j]
 			ctx.AddLabel(label.Name, label.Value)
 		}
-		if !ctx.TryPrepareLabels(hasRelabeling) {
+		if hasRelabeling {
+			ctx.ApplyRelabeling()
+		}
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
 			continue
 		}
+		ctx.SortLabelsIfNeeded()
 		var metricNameRaw []byte
 		var err error
 		for _, pt := range ss.Points {

app/vminsert/datadogv2/request_handler.go

@@ -66,9 +66,14 @@ func insertRows(series []datadogv2.Series, extraLabels []prompbmarshal.Label) er
 			label := &extraLabels[j]
 			ctx.AddLabel(label.Name, label.Value)
 		}
-		if !ctx.TryPrepareLabels(hasRelabeling) {
+		if hasRelabeling {
+			ctx.ApplyRelabeling()
+		}
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
 			continue
 		}
+		ctx.SortLabelsIfNeeded()
 		var metricNameRaw []byte
 		var err error
 		for _, pt := range ss.Points {

app/vminsert/graphite/request_handler.go

@@ -36,9 +36,14 @@ func insertRows(rows []parser.Row) error {
 			tag := &r.Tags[j]
 			ctx.AddLabel(tag.Key, tag.Value)
 		}
-		if !ctx.TryPrepareLabels(hasRelabeling) {
+		if hasRelabeling {
+			ctx.ApplyRelabeling()
+		}
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
 			continue
 		}
+		ctx.SortLabelsIfNeeded()
 		if err := ctx.WriteDataPoint(nil, ctx.Labels, r.Timestamp, r.Value); err != nil {
 			return err
 		}

app/vminsert/influx/request_handler.go

@@ -9,12 +9,12 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vminsert/relabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -70,7 +70,6 @@ func insertRows(db string, rows []parser.Row, extraLabels []prompbmarshal.Label)
 	ic.Reset(rowsLen)
 	rowsTotal := 0
 	hasRelabeling := relabel.HasRelabeling()
-	hasLimitsEnabled := timeserieslimits.Enabled()
 	for i := range rows {
 		r := &rows[i]
 		rowsTotal += len(r.Fields)
@@ -110,17 +109,18 @@ func insertRows(db string, rows []parser.Row, extraLabels []prompbmarshal.Label)
 				metricGroup := bytesutil.ToUnsafeString(ctx.metricGroupBuf)
 				ic.Labels = append(ic.Labels[:0], ctx.originLabels...)
 				ic.AddLabel("", metricGroup)
-				if !ic.TryPrepareLabels(true) {
+				ic.ApplyRelabeling()
+				if len(ic.Labels) == 0 {
+					// Skip metric without labels.
 					continue
 				}
+				ic.SortLabelsIfNeeded()
 				if err := ic.WriteDataPoint(nil, ic.Labels, r.Timestamp, f.Value); err != nil {
 					return err
 				}
 			}
 		} else {
-			if !ic.TryPrepareLabels(false) {
-				continue
-			}
+			ic.SortLabelsIfNeeded()
 			ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf[:0], ic.Labels)
 			labelsLen := len(ic.Labels)
 			for j := range r.Fields {
@@ -131,12 +131,9 @@ func insertRows(db string, rows []parser.Row, extraLabels []prompbmarshal.Label)
 				metricGroup := bytesutil.ToUnsafeString(ctx.metricGroupBuf)
 				ic.Labels = ic.Labels[:labelsLen]
 				ic.AddLabel("", metricGroup)
-				if hasLimitsEnabled {
-					// special case for optimisation above
-					// check only __name__ label value limits
-					if timeserieslimits.IsExceeding(ic.Labels[len(ic.Labels)-1:]) {
-						continue
-					}
+				if len(ic.Labels) == 0 {
+					// Skip metric without labels.
+					continue
 				}
 				if err := ic.WriteDataPoint(ctx.metricNameBuf, ic.Labels[len(ic.Labels)-1:], r.Timestamp, f.Value); err != nil {
 					return err
@@ -153,7 +150,7 @@ type pushCtx struct {
 	Common         common.InsertCtx
 	metricNameBuf  []byte
 	metricGroupBuf []byte
-	originLabels   []prompbmarshal.Label
+	originLabels   []prompb.Label
 }
 
 func (ctx *pushCtx) reset() {
@@ -163,7 +160,7 @@ func (ctx *pushCtx) reset() {
 
 	originLabels := ctx.originLabels
 	for i := range originLabels {
-		originLabels[i] = prompbmarshal.Label{}
+		originLabels[i] = prompb.Label{}
 	}
 	ctx.originLabels = originLabels[:0]
 }

app/vminsert/main.go

@@ -41,8 +41,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/firehose"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 )
 
 var (
@@ -67,9 +67,8 @@ var (
 		"at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
 	configAuthKey          = flagutil.NewPassword("configAuthKey", "Authorization key for accessing /config page. It must be passed via authKey query arg. It overrides -httpAuth.*")
 	reloadAuthKey          = flagutil.NewPassword("reloadAuthKey", "Auth key for /-/reload http endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings.")
-	maxLabelsPerTimeseries = flag.Int("maxLabelsPerTimeseries", 40, "The maximum number of labels per time series to be accepted. Series with superfluous labels are ignored. In this case the vm_rows_ignored_total{reason=\"too_many_labels\"} metric at /metrics page is incremented")
-	maxLabelNameLen        = flag.Int("maxLabelNameLen", 256, "The maximum length of label name in the accepted time series. Series with longer label name are ignored. In this case the vm_rows_ignored_total{reason=\"too_long_label_name\"} metric at /metrics page is incremented")
-	maxLabelValueLen       = flag.Int("maxLabelValueLen", 4*1024, "The maximum length of label values in the accepted time series. Series with longer label value are ignored. In this case the vm_rows_ignored_total{reason=\"too_long_label_value\"} metric at /metrics page is incremented")
+	maxLabelsPerTimeseries = flag.Int("maxLabelsPerTimeseries", 30, "The maximum number of labels accepted per time series. Superfluous labels are dropped. In this case the vm_metrics_with_dropped_labels_total metric at /metrics page is incremented")
+	maxLabelValueLen       = flag.Int("maxLabelValueLen", 4*1024, "The maximum length of label values in the accepted time series. Longer label values are truncated. In this case the vm_too_long_label_values_total metric at /metrics page is incremented")
 )
 
 var (
@@ -88,6 +87,8 @@ var staticServer = http.FileServer(http.FS(staticFiles))
 func Init() {
 	relabel.Init()
 	vminsertCommon.InitStreamAggr()
+	storage.SetMaxLabelsPerTimeseries(*maxLabelsPerTimeseries)
+	storage.SetMaxLabelValueLen(*maxLabelValueLen)
 	common.StartUnmarshalWorkers()
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, *graphiteUseProxyProtocol, graphite.InsertHandler)
@@ -104,7 +105,6 @@ func Init() {
 	promscrape.Init(func(_ *auth.Token, wr *prompbmarshal.WriteRequest) {
 		prompush.Push(wr)
 	})
-	timeserieslimits.Init(*maxLabelsPerTimeseries, *maxLabelNameLen, *maxLabelValueLen)
 }
 
 // Stop stops vminsert.
@@ -439,4 +439,14 @@ var (
 	promscrapeStatusConfigRequests = metrics.NewCounter(`vm_http_requests_total{path="/api/v1/status/config"}`)
 
 	promscrapeConfigReloadRequests = metrics.NewCounter(`vm_http_requests_total{path="/-/reload"}`)
+
+	_ = metrics.NewGauge(`vm_metrics_with_dropped_labels_total`, func() float64 {
+		return float64(storage.MetricsWithDroppedLabels.Load())
+	})
+	_ = metrics.NewGauge(`vm_too_long_label_names_total`, func() float64 {
+		return float64(storage.TooLongLabelNames.Load())
+	})
+	_ = metrics.NewGauge(`vm_too_long_label_values_total`, func() float64 {
+		return float64(storage.TooLongLabelValues.Load())
+	})
 )

app/vminsert/native/request_handler.go

@@ -55,9 +55,14 @@ func insertRows(block *stream.Block, extraLabels []prompbmarshal.Label) error {
 		label := &extraLabels[j]
 		ic.AddLabel(label.Name, label.Value)
 	}
-	if !ic.TryPrepareLabels(hasRelabeling) {
+	if hasRelabeling {
+		ic.ApplyRelabeling()
+	}
+	if len(ic.Labels) == 0 {
+		// Skip metric without labels.
 		return nil
 	}
+	ic.SortLabelsIfNeeded()
 	ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf[:0], ic.Labels)
 	values := block.Values
 	timestamps := block.Timestamps
@@ -66,9 +71,7 @@ func insertRows(block *stream.Block, extraLabels []prompbmarshal.Label) error {
 	}
 	for j, value := range values {
 		timestamp := timestamps[j]
-		// TODO: @f41gh7 looks like it's better to use WriteDataPointExt
-		// since metricName never changes inside insertRows call
-		if err := ic.WriteDataPoint(ctx.metricNameBuf, ic.Labels, timestamp, value); err != nil {
+		if err := ic.WriteDataPoint(ctx.metricNameBuf, nil, timestamp, value); err != nil {
 			return err
 		}
 	}