CHANGELOG.md: cut v1.107.0 release

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

@@ -13,7 +13,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
 )
 
 var (
@@ -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 {

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 {

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"
 )
@@ -471,15 +472,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 +716,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/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)

app/vmalert-tool/unittest/unittest_test.go

@@ -1,14 +1,24 @@
 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")
 		}
@@ -23,7 +33,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/main.go

@@ -81,10 +81,7 @@ absolute path to all .tpl files in root.
 	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 +95,9 @@ func main() {
 	buildinfo.Init()
 	logger.Init()
 
-	var err error
-	extURL, err = getExternalURL(*externalURL)
+	err := templates.Load(*ruleTemplatesPath, true)
 	if err != nil {
-		logger.Fatalf("failed to init external.url %q: %s", *externalURL, err)
-	}
-
-	err = templates.Load(*ruleTemplatesPath, *extURL)
-	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 +111,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 +304,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 +359,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

app/vmalert/notifier/alert_test.go

@@ -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/rule/group_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"net/url"
 	"os"
 	"sort"
 	"testing"
@@ -27,7 +26,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)
 	}

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) {
@@ -142,6 +174,13 @@ func GetWithFuncs(funcs textTpl.FuncMap) (*textTpl.Template, error) {
 	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 +198,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

@@ -783,11 +783,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)

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/main.go

@@ -199,7 +199,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 {
@@ -213,7 +213,7 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 			missingRouteRequests.Inc()
 			var di string
 			if ui.DumpRequestOnErrors {
-				di = debugInfo(u, r)
+				di = debugInfo(u, r.Header)
 			}
 			httpserver.Errorf(w, r, "missing route for %q%s", u.String(), di)
 			return
@@ -668,13 +668,13 @@ func (rtb *readTrackingBody) Close() error {
 	return nil
 }
 
-func debugInfo(u *url.URL, r *http.Request) string {
+func debugInfo(u *url.URL, h http.Header) string {
 	s := &strings.Builder{}
-	fmt.Fprintf(s, " (host: %q; ", r.Host)
+	fmt.Fprintf(s, " (host: %q; ", u.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)
+	_ = h.WriteSubset(s, nil)
 	fmt.Fprint(s, ")")
 	return s.String()
 }

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)
 		}
@@ -306,7 +306,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 +384,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 +432,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/remote_read_test.go

@@ -1,348 +1,348 @@
 package main
 
-import (
-	"context"
-	"net/http"
-	"testing"
-	"time"
-
-	"github.com/prometheus/prometheus/prompb"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/backoff"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/remoteread"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/testdata/servers_integration_test"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
-)
-
-func TestRemoteRead(t *testing.T) {
-	barpool.Disable(true)
-	defer func() {
-		barpool.Disable(false)
-	}()
-	defer func() { isSilent = false }()
-
-	var testCases = []struct {
-		name             string
-		remoteReadConfig remoteread.Config
-		vmCfg            vm.Config
-		start            string
-		end              string
-		numOfSamples     int64
-		numOfSeries      int64
-		rrp              remoteReadProcessor
-		chunk            string
-		remoteReadSeries func(start, end, numOfSeries, numOfSamples int64) []*prompb.TimeSeries
-		expectedSeries   []vm.TimeSeries
-	}{
-		{
-			name:             "step minute on minute time range",
-			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*"},
-			vmCfg:            vm.Config{Addr: "", Concurrency: 1},
-			start:            "2022-11-26T11:23:05+02:00",
-			end:              "2022-11-26T11:24:05+02:00",
-			numOfSamples:     2,
-			numOfSeries:      3,
-			chunk:            stepper.StepMinute,
-			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
-			expectedSeries: []vm.TimeSeries{
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
-					Timestamps: []int64{1669454585000, 1669454615000},
-					Values:     []float64{0, 0},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
-					Timestamps: []int64{1669454585000, 1669454615000},
-					Values:     []float64{100, 100},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
-					Timestamps: []int64{1669454585000, 1669454615000},
-					Values:     []float64{200, 200},
-				},
-			},
-		},
-		{
-			name:             "step month on month time range",
-			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*"},
-			vmCfg: vm.Config{Addr: "", Concurrency: 1,
-				Transport: http.DefaultTransport.(*http.Transport)},
-			start:            "2022-09-26T11:23:05+02:00",
-			end:              "2022-11-26T11:24:05+02:00",
-			numOfSamples:     2,
-			numOfSeries:      3,
-			chunk:            stepper.StepMonth,
-			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
-			expectedSeries: []vm.TimeSeries{
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
-					Timestamps: []int64{1664184185000},
-					Values:     []float64{0},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
-					Timestamps: []int64{1664184185000},
-					Values:     []float64{100},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
-					Timestamps: []int64{1664184185000},
-					Values:     []float64{200},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
-					Timestamps: []int64{1666819415000},
-					Values:     []float64{0},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
-					Timestamps: []int64{1666819415000},
-					Values:     []float64{100},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
-					Timestamps: []int64{1666819415000},
-					Values:     []float64{200}},
-			},
-		},
-	}
-
-	for _, tt := range testCases {
-		t.Run(tt.name, func(t *testing.T) {
-			ctx := context.Background()
-			remoteReadServer := remote_read_integration.NewRemoteReadServer(t)
-			defer remoteReadServer.Close()
-			remoteWriteServer := remote_read_integration.NewRemoteWriteServer(t)
-			defer remoteWriteServer.Close()
-
-			tt.remoteReadConfig.Addr = remoteReadServer.URL()
-
-			rr, err := remoteread.NewClient(tt.remoteReadConfig)
-			if err != nil {
-				t.Fatalf("error create remote read client: %s", err)
-			}
-
-			start, err := time.Parse(time.RFC3339, tt.start)
-			if err != nil {
-				t.Fatalf("Error parse start time: %s", err)
-			}
-
-			end, err := time.Parse(time.RFC3339, tt.end)
-			if err != nil {
-				t.Fatalf("Error parse end time: %s", err)
-			}
-
-			rrs := tt.remoteReadSeries(start.Unix(), end.Unix(), tt.numOfSeries, tt.numOfSamples)
-
-			remoteReadServer.SetRemoteReadSeries(rrs)
-			remoteWriteServer.ExpectedSeries(tt.expectedSeries)
-
-			tt.vmCfg.Addr = remoteWriteServer.URL()
-
-			b, err := backoff.New(10, 1.8, time.Second*2)
-			if err != nil {
-				t.Fatalf("failed to create backoff: %s", err)
-			}
-			tt.vmCfg.Backoff = b
-
-			importer, err := vm.NewImporter(ctx, tt.vmCfg)
-			if err != nil {
-				t.Fatalf("failed to create VM importer: %s", err)
-			}
-			defer importer.Close()
-
-			rmp := remoteReadProcessor{
-				src: rr,
-				dst: importer,
-				filter: remoteReadFilter{
-					timeStart: &start,
-					timeEnd:   &end,
-					chunk:     tt.chunk,
-				},
-				cc:        1,
-				isVerbose: false,
-			}
-
-			err = rmp.run(ctx)
-			if err != nil {
-				t.Fatalf("failed to run remote read processor: %s", err)
-			}
-		})
-	}
-}
-
-func TestSteamRemoteRead(t *testing.T) {
-	barpool.Disable(true)
-	defer func() {
-		barpool.Disable(false)
-	}()
-	defer func() { isSilent = false }()
-
-	var testCases = []struct {
-		name             string
-		remoteReadConfig remoteread.Config
-		vmCfg            vm.Config
-		start            string
-		end              string
-		numOfSamples     int64
-		numOfSeries      int64
-		rrp              remoteReadProcessor
-		chunk            string
-		remoteReadSeries func(start, end, numOfSeries, numOfSamples int64) []*prompb.TimeSeries
-		expectedSeries   []vm.TimeSeries
-	}{
-		{
-			name:             "step minute on minute time range",
-			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*", UseStream: true},
-			vmCfg:            vm.Config{Addr: "", Concurrency: 1},
-			start:            "2022-11-26T11:23:05+02:00",
-			end:              "2022-11-26T11:24:05+02:00",
-			numOfSamples:     2,
-			numOfSeries:      3,
-			chunk:            stepper.StepMinute,
-			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
-			expectedSeries: []vm.TimeSeries{
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
-					Timestamps: []int64{1669454585000, 1669454615000},
-					Values:     []float64{0, 0},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
-					Timestamps: []int64{1669454585000, 1669454615000},
-					Values:     []float64{100, 100},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
-					Timestamps: []int64{1669454585000, 1669454615000},
-					Values:     []float64{200, 200},
-				},
-			},
-		},
-		{
-			name:             "step month on month time range",
-			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*", UseStream: true},
-			vmCfg:            vm.Config{Addr: "", Concurrency: 1},
-			start:            "2022-09-26T11:23:05+02:00",
-			end:              "2022-11-26T11:24:05+02:00",
-			numOfSamples:     2,
-			numOfSeries:      3,
-			chunk:            stepper.StepMonth,
-			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
-			expectedSeries: []vm.TimeSeries{
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
-					Timestamps: []int64{1664184185000},
-					Values:     []float64{0},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
-					Timestamps: []int64{1664184185000},
-					Values:     []float64{100},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
-					Timestamps: []int64{1664184185000},
-					Values:     []float64{200},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
-					Timestamps: []int64{1666819415000},
-					Values:     []float64{0},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
-					Timestamps: []int64{1666819415000},
-					Values:     []float64{100},
-				},
-				{
-					Name:       "vm_metric_1",
-					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
-					Timestamps: []int64{1666819415000},
-					Values:     []float64{200}},
-			},
-		},
-	}
-
-	for _, tt := range testCases {
-		t.Run(tt.name, func(t *testing.T) {
-			ctx := context.Background()
-			remoteReadServer := remote_read_integration.NewRemoteReadStreamServer(t)
-			defer remoteReadServer.Close()
-			remoteWriteServer := remote_read_integration.NewRemoteWriteServer(t)
-			defer remoteWriteServer.Close()
-
-			tt.remoteReadConfig.Addr = remoteReadServer.URL()
-
-			rr, err := remoteread.NewClient(tt.remoteReadConfig)
-			if err != nil {
-				t.Fatalf("error create remote read client: %s", err)
-			}
-
-			start, err := time.Parse(time.RFC3339, tt.start)
-			if err != nil {
-				t.Fatalf("Error parse start time: %s", err)
-			}
-
-			end, err := time.Parse(time.RFC3339, tt.end)
-			if err != nil {
-				t.Fatalf("Error parse end time: %s", err)
-			}
-
-			rrs := tt.remoteReadSeries(start.Unix(), end.Unix(), tt.numOfSeries, tt.numOfSamples)
-
-			remoteReadServer.InitMockStorage(rrs)
-			remoteWriteServer.ExpectedSeries(tt.expectedSeries)
-
-			tt.vmCfg.Addr = remoteWriteServer.URL()
-
-			b, err := backoff.New(10, 1.8, time.Second*2)
-			if err != nil {
-				t.Fatalf("failed to create backoff: %s", err)
-			}
-
-			tt.vmCfg.Backoff = b
-			importer, err := vm.NewImporter(ctx, tt.vmCfg)
-			if err != nil {
-				t.Fatalf("failed to create VM importer: %s", err)
-			}
-			defer importer.Close()
-
-			rmp := remoteReadProcessor{
-				src: rr,
-				dst: importer,
-				filter: remoteReadFilter{
-					timeStart: &start,
-					timeEnd:   &end,
-					chunk:     tt.chunk,
-				},
-				cc:        1,
-				isVerbose: false,
-			}
-
-			err = rmp.run(ctx)
-			if err != nil {
-				t.Fatalf("failed to run remote read processor: %s", err)
-			}
-		})
-	}
-}
+// import (
+// 	"context"
+// 	"net/http"
+// 	"testing"
+// 	"time"
+//
+// 	"github.com/prometheus/prometheus/prompb"
+//
+// 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/backoff"
+// 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
+// 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/remoteread"
+// 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
+// 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/testdata/servers_integration_test"
+// 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
+// )
+//
+// func TestRemoteRead(t *testing.T) {
+// 	barpool.Disable(true)
+// 	defer func() {
+// 		barpool.Disable(false)
+// 	}()
+// 	defer func() { isSilent = false }()
+//
+// 	var testCases = []struct {
+// 		name             string
+// 		remoteReadConfig remoteread.Config
+// 		vmCfg            vm.Config
+// 		start            string
+// 		end              string
+// 		numOfSamples     int64
+// 		numOfSeries      int64
+// 		rrp              remoteReadProcessor
+// 		chunk            string
+// 		remoteReadSeries func(start, end, numOfSeries, numOfSamples int64) []*prompb.TimeSeries
+// 		expectedSeries   []vm.TimeSeries
+// 	}{
+// 		{
+// 			name:             "step minute on minute time range",
+// 			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*"},
+// 			vmCfg:            vm.Config{Addr: "", Concurrency: 1},
+// 			start:            "2022-11-26T11:23:05+02:00",
+// 			end:              "2022-11-26T11:24:05+02:00",
+// 			numOfSamples:     2,
+// 			numOfSeries:      3,
+// 			chunk:            stepper.StepMinute,
+// 			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
+// 			expectedSeries: []vm.TimeSeries{
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
+// 					Timestamps: []int64{1669454585000, 1669454615000},
+// 					Values:     []float64{0, 0},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
+// 					Timestamps: []int64{1669454585000, 1669454615000},
+// 					Values:     []float64{100, 100},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
+// 					Timestamps: []int64{1669454585000, 1669454615000},
+// 					Values:     []float64{200, 200},
+// 				},
+// 			},
+// 		},
+// 		{
+// 			name:             "step month on month time range",
+// 			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*"},
+// 			vmCfg: vm.Config{Addr: "", Concurrency: 1,
+// 				Transport: http.DefaultTransport.(*http.Transport)},
+// 			start:            "2022-09-26T11:23:05+02:00",
+// 			end:              "2022-11-26T11:24:05+02:00",
+// 			numOfSamples:     2,
+// 			numOfSeries:      3,
+// 			chunk:            stepper.StepMonth,
+// 			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
+// 			expectedSeries: []vm.TimeSeries{
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
+// 					Timestamps: []int64{1664184185000},
+// 					Values:     []float64{0},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
+// 					Timestamps: []int64{1664184185000},
+// 					Values:     []float64{100},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
+// 					Timestamps: []int64{1664184185000},
+// 					Values:     []float64{200},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
+// 					Timestamps: []int64{1666819415000},
+// 					Values:     []float64{0},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
+// 					Timestamps: []int64{1666819415000},
+// 					Values:     []float64{100},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
+// 					Timestamps: []int64{1666819415000},
+// 					Values:     []float64{200}},
+// 			},
+// 		},
+// 	}
+//
+// 	for _, tt := range testCases {
+// 		t.Run(tt.name, func(t *testing.T) {
+// 			ctx := context.Background()
+// 			remoteReadServer := remote_read_integration.NewRemoteReadServer(t)
+// 			defer remoteReadServer.Close()
+// 			remoteWriteServer := remote_read_integration.NewRemoteWriteServer(t)
+// 			defer remoteWriteServer.Close()
+//
+// 			tt.remoteReadConfig.Addr = remoteReadServer.URL()
+//
+// 			rr, err := remoteread.NewClient(tt.remoteReadConfig)
+// 			if err != nil {
+// 				t.Fatalf("error create remote read client: %s", err)
+// 			}
+//
+// 			start, err := time.Parse(time.RFC3339, tt.start)
+// 			if err != nil {
+// 				t.Fatalf("Error parse start time: %s", err)
+// 			}
+//
+// 			end, err := time.Parse(time.RFC3339, tt.end)
+// 			if err != nil {
+// 				t.Fatalf("Error parse end time: %s", err)
+// 			}
+//
+// 			rrs := tt.remoteReadSeries(start.Unix(), end.Unix(), tt.numOfSeries, tt.numOfSamples)
+//
+// 			remoteReadServer.SetRemoteReadSeries(rrs)
+// 			remoteWriteServer.ExpectedSeries(tt.expectedSeries)
+//
+// 			tt.vmCfg.Addr = remoteWriteServer.URL()
+//
+// 			b, err := backoff.New(10, 1.8, time.Second*2)
+// 			if err != nil {
+// 				t.Fatalf("failed to create backoff: %s", err)
+// 			}
+// 			tt.vmCfg.Backoff = b
+//
+// 			importer, err := vm.NewImporter(ctx, tt.vmCfg)
+// 			if err != nil {
+// 				t.Fatalf("failed to create VM importer: %s", err)
+// 			}
+// 			defer importer.Close()
+//
+// 			rmp := remoteReadProcessor{
+// 				src: rr,
+// 				dst: importer,
+// 				filter: remoteReadFilter{
+// 					timeStart: &start,
+// 					timeEnd:   &end,
+// 					chunk:     tt.chunk,
+// 				},
+// 				cc:        1,
+// 				isVerbose: false,
+// 			}
+//
+// 			err = rmp.run(ctx)
+// 			if err != nil {
+// 				t.Fatalf("failed to run remote read processor: %s", err)
+// 			}
+// 		})
+// 	}
+// }
+//
+// func TestSteamRemoteRead(t *testing.T) {
+// 	barpool.Disable(true)
+// 	defer func() {
+// 		barpool.Disable(false)
+// 	}()
+// 	defer func() { isSilent = false }()
+//
+// 	var testCases = []struct {
+// 		name             string
+// 		remoteReadConfig remoteread.Config
+// 		vmCfg            vm.Config
+// 		start            string
+// 		end              string
+// 		numOfSamples     int64
+// 		numOfSeries      int64
+// 		rrp              remoteReadProcessor
+// 		chunk            string
+// 		remoteReadSeries func(start, end, numOfSeries, numOfSamples int64) []*prompb.TimeSeries
+// 		expectedSeries   []vm.TimeSeries
+// 	}{
+// 		{
+// 			name:             "step minute on minute time range",
+// 			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*", UseStream: true},
+// 			vmCfg:            vm.Config{Addr: "", Concurrency: 1},
+// 			start:            "2022-11-26T11:23:05+02:00",
+// 			end:              "2022-11-26T11:24:05+02:00",
+// 			numOfSamples:     2,
+// 			numOfSeries:      3,
+// 			chunk:            stepper.StepMinute,
+// 			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
+// 			expectedSeries: []vm.TimeSeries{
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
+// 					Timestamps: []int64{1669454585000, 1669454615000},
+// 					Values:     []float64{0, 0},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
+// 					Timestamps: []int64{1669454585000, 1669454615000},
+// 					Values:     []float64{100, 100},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
+// 					Timestamps: []int64{1669454585000, 1669454615000},
+// 					Values:     []float64{200, 200},
+// 				},
+// 			},
+// 		},
+// 		{
+// 			name:             "step month on month time range",
+// 			remoteReadConfig: remoteread.Config{Addr: "", LabelName: "__name__", LabelValue: ".*", UseStream: true},
+// 			vmCfg:            vm.Config{Addr: "", Concurrency: 1},
+// 			start:            "2022-09-26T11:23:05+02:00",
+// 			end:              "2022-11-26T11:24:05+02:00",
+// 			numOfSamples:     2,
+// 			numOfSeries:      3,
+// 			chunk:            stepper.StepMonth,
+// 			remoteReadSeries: remote_read_integration.GenerateRemoteReadSeries,
+// 			expectedSeries: []vm.TimeSeries{
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
+// 					Timestamps: []int64{1664184185000},
+// 					Values:     []float64{0},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
+// 					Timestamps: []int64{1664184185000},
+// 					Values:     []float64{100},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
+// 					Timestamps: []int64{1664184185000},
+// 					Values:     []float64{200},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "0"}},
+// 					Timestamps: []int64{1666819415000},
+// 					Values:     []float64{0},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "1"}},
+// 					Timestamps: []int64{1666819415000},
+// 					Values:     []float64{100},
+// 				},
+// 				{
+// 					Name:       "vm_metric_1",
+// 					LabelPairs: []vm.LabelPair{{Name: "job", Value: "2"}},
+// 					Timestamps: []int64{1666819415000},
+// 					Values:     []float64{200}},
+// 			},
+// 		},
+// 	}
+//
+// 	for _, tt := range testCases {
+// 		t.Run(tt.name, func(t *testing.T) {
+// 			ctx := context.Background()
+// 			remoteReadServer := remote_read_integration.NewRemoteReadStreamServer(t)
+// 			defer remoteReadServer.Close()
+// 			remoteWriteServer := remote_read_integration.NewRemoteWriteServer(t)
+// 			defer remoteWriteServer.Close()
+//
+// 			tt.remoteReadConfig.Addr = remoteReadServer.URL()
+//
+// 			rr, err := remoteread.NewClient(tt.remoteReadConfig)
+// 			if err != nil {
+// 				t.Fatalf("error create remote read client: %s", err)
+// 			}
+//
+// 			start, err := time.Parse(time.RFC3339, tt.start)
+// 			if err != nil {
+// 				t.Fatalf("Error parse start time: %s", err)
+// 			}
+//
+// 			end, err := time.Parse(time.RFC3339, tt.end)
+// 			if err != nil {
+// 				t.Fatalf("Error parse end time: %s", err)
+// 			}
+//
+// 			rrs := tt.remoteReadSeries(start.Unix(), end.Unix(), tt.numOfSeries, tt.numOfSamples)
+//
+// 			remoteReadServer.InitMockStorage(rrs)
+// 			remoteWriteServer.ExpectedSeries(tt.expectedSeries)
+//
+// 			tt.vmCfg.Addr = remoteWriteServer.URL()
+//
+// 			b, err := backoff.New(10, 1.8, time.Second*2)
+// 			if err != nil {
+// 				t.Fatalf("failed to create backoff: %s", err)
+// 			}
+//
+// 			tt.vmCfg.Backoff = b
+// 			importer, err := vm.NewImporter(ctx, tt.vmCfg)
+// 			if err != nil {
+// 				t.Fatalf("failed to create VM importer: %s", err)
+// 			}
+// 			defer importer.Close()
+//
+// 			rmp := remoteReadProcessor{
+// 				src: rr,
+// 				dst: importer,
+// 				filter: remoteReadFilter{
+// 					timeStart: &start,
+// 					timeEnd:   &end,
+// 					chunk:     tt.chunk,
+// 				},
+// 				cc:        1,
+// 				isVerbose: false,
+// 			}
+//
+// 			err = rmp.run(ctx)
+// 			if err != nil {
+// 				t.Fatalf("failed to run remote read processor: %s", err)
+// 			}
+// 		})
+// 	}
+// }

app/vmctl/remoteread/remoteread.go

@@ -18,6 +18,7 @@ import (
 	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/prometheus/prometheus/storage/remote"
+
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
 )
 

app/vmctl/testdata/servers_integration_test/remote_read_server.go

@@ -1,469 +1,368 @@
 package remote_read_integration
 
-import (
-	"context"
-	"fmt"
-	"io"
-	"net/http"
-	"net/http/httptest"
-	"strconv"
-	"strings"
-	"testing"
-
-	"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 (
-	maxBytesInFrame = 1024 * 1024
-)
-
-type RemoteReadServer struct {
-	server  *httptest.Server
-	series  []*prompb.TimeSeries
-	storage *MockStorage
-}
-
-// NewRemoteReadServer creates a remote read server. It exposes a single endpoint and responds with the
-// passed series based on the request to the read endpoint. It returns a server which should be closed after
-// being used.
-func NewRemoteReadServer(t *testing.T) *RemoteReadServer {
-	rrs := &RemoteReadServer{
-		series: make([]*prompb.TimeSeries, 0),
-	}
-	rrs.server = httptest.NewServer(rrs.getReadHandler(t))
-	return rrs
-}
-
-// Close closes the server.
-func (rrs *RemoteReadServer) Close() {
-	rrs.server.Close()
-}
-
-func (rrs *RemoteReadServer) URL() string {
-	return rrs.server.URL
-}
-
-func (rrs *RemoteReadServer) SetRemoteReadSeries(series []*prompb.TimeSeries) {
-	rrs.series = append(rrs.series, series...)
-}
-
-func (rrs *RemoteReadServer) getReadHandler(t *testing.T) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if !validateReadHeaders(t, r) {
-			t.Fatalf("invalid read headers")
-		}
-
-		compressed, err := io.ReadAll(r.Body)
-		if err != nil {
-			t.Fatalf("error read body: %s", err)
-		}
-
-		reqBuf, err := snappy.Decode(nil, compressed)
-		if err != nil {
-			t.Fatalf("error decode compressed data:%s", err)
-		}
-
-		var req prompb.ReadRequest
-		if err := proto.Unmarshal(reqBuf, &req); err != nil {
-			t.Fatalf("error unmarshal read request: %s", err)
-		}
-
-		resp := &prompb.ReadResponse{
-			Results: make([]*prompb.QueryResult, len(req.Queries)),
-		}
-
-		for i, r := range req.Queries {
-			startTs := r.StartTimestampMs
-			endTs := r.EndTimestampMs
-			ts := make([]*prompb.TimeSeries, len(rrs.series))
-			for i, s := range rrs.series {
-				var samples []prompb.Sample
-				for _, sample := range s.Samples {
-					if sample.Timestamp >= startTs && sample.Timestamp < endTs {
-						samples = append(samples, sample)
-					}
-				}
-				var series prompb.TimeSeries
-				if len(samples) > 0 {
-					series.Labels = s.Labels
-					series.Samples = samples
-				}
-				ts[i] = &series
-			}
-
-			resp.Results[i] = &prompb.QueryResult{Timeseries: ts}
-			data, err := proto.Marshal(resp)
-			if err != nil {
-				t.Fatalf("error marshal response: %s", err)
-			}
-
-			compressed = snappy.Encode(nil, data)
-
-			w.Header().Set("Content-Type", "application/x-protobuf")
-			w.Header().Set("Content-Encoding", "snappy")
-			w.WriteHeader(http.StatusOK)
-
-			if _, err := w.Write(compressed); err != nil {
-				t.Fatalf("snappy encode error: %s", err)
-			}
-		}
-	})
-}
-
-func NewRemoteReadStreamServer(t *testing.T) *RemoteReadServer {
-	rrs := &RemoteReadServer{
-		series: make([]*prompb.TimeSeries, 0),
-	}
-	rrs.server = httptest.NewServer(rrs.getStreamReadHandler(t))
-	return rrs
-}
-
-func (rrs *RemoteReadServer) InitMockStorage(series []*prompb.TimeSeries) {
-	rrs.storage = NewMockStorage(series)
-}
-
-func (rrs *RemoteReadServer) getStreamReadHandler(t *testing.T) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if !validateStreamReadHeaders(t, r) {
-			t.Fatalf("invalid read headers")
-		}
-
-		f, ok := w.(http.Flusher)
-		if !ok {
-			t.Fatalf("internal http.ResponseWriter does not implement http.Flusher interface")
-		}
-
-		stream := remote.NewChunkedWriter(w, f)
-
-		data, err := io.ReadAll(r.Body)
-		if err != nil {
-			t.Fatalf("error read body: %s", err)
-		}
-
-		decodedData, err := snappy.Decode(nil, data)
-		if err != nil {
-			t.Fatalf("error decode compressed data:%s", err)
-		}
-
-		var req prompb.ReadRequest
-		if err := proto.Unmarshal(decodedData, &req); err != nil {
-			t.Fatalf("error unmarshal read request: %s", err)
-		}
-
-		var chks []prompb.Chunk
-		ctx := context.Background()
-		for idx, r := range req.Queries {
-			startTs := r.StartTimestampMs
-			endTs := r.EndTimestampMs
-
-			var matchers []*labels.Matcher
-			cb := func() (int64, error) { return 0, nil }
-
-			c := remote.NewSampleAndChunkQueryableClient(rrs.storage, nil, matchers, true, cb)
-
-			q, err := c.ChunkQuerier(startTs, endTs)
-			if err != nil {
-				t.Fatalf("error init chunk querier: %s", err)
-			}
-
-			ss := q.Select(ctx, false, nil, matchers...)
-			var iter chunks.Iterator
-			for ss.Next() {
-				series := ss.At()
-				iter = series.Iterator(iter)
-				labels := remote.MergeLabels(labelsToLabelsProto(series.Labels()), nil)
-
-				frameBytesLeft := maxBytesInFrame
-				for _, lb := range labels {
-					frameBytesLeft -= lb.Size()
-				}
-
-				isNext := iter.Next()
-
-				for isNext {
-					chunk := iter.At()
-
-					if chunk.Chunk == nil {
-						t.Fatalf("error found not populated chunk returned by SeriesSet at ref: %v", chunk.Ref)
-					}
-
-					chks = append(chks, prompb.Chunk{
-						MinTimeMs: chunk.MinTime,
-						MaxTimeMs: chunk.MaxTime,
-						Type:      prompb.Chunk_Encoding(chunk.Chunk.Encoding()),
-						Data:      chunk.Chunk.Bytes(),
-					})
-
-					frameBytesLeft -= chks[len(chks)-1].Size()
-
-					// We are fine with minor inaccuracy of max bytes per frame. The inaccuracy will be max of full chunk size.
-					isNext = iter.Next()
-					if frameBytesLeft > 0 && isNext {
-						continue
-					}
-
-					resp := &prompb.ChunkedReadResponse{
-						ChunkedSeries: []*prompb.ChunkedSeries{
-							{Labels: labels, Chunks: chks},
-						},
-						QueryIndex: int64(idx),
-					}
-
-					b, err := proto.Marshal(resp)
-					if err != nil {
-						t.Fatalf("error marshal response: %s", err)
-					}
-
-					if _, err := stream.Write(b); err != nil {
-						t.Fatalf("error write to stream: %s", err)
-					}
-					chks = chks[:0]
-					rrs.storage.Reset()
-				}
-				if err := iter.Err(); err != nil {
-					t.Fatalf("error iterate over chunk series: %s", err)
-				}
-			}
-		}
-	})
-}
-
-func validateReadHeaders(t *testing.T, r *http.Request) bool {
-	if r.Method != http.MethodPost {
-		t.Fatalf("got %q method, expected %q", r.Method, http.MethodPost)
-	}
-	if r.Header.Get("Content-Encoding") != "snappy" {
-		t.Fatalf("got %q content encoding header, expected %q", r.Header.Get("Content-Encoding"), "snappy")
-	}
-	if r.Header.Get("Content-Type") != "application/x-protobuf" {
-		t.Fatalf("got %q content type header, expected %q", r.Header.Get("Content-Type"), "application/x-protobuf")
-	}
-
-	remoteReadVersion := r.Header.Get("X-Prometheus-Remote-Read-Version")
-	if remoteReadVersion == "" {
-		t.Fatalf("got empty prometheus remote read header")
-	}
-	if !strings.HasPrefix(remoteReadVersion, "0.1.") {
-		t.Fatalf("wrong remote version defined")
-	}
-
-	return true
-}
-
-func validateStreamReadHeaders(t *testing.T, r *http.Request) bool {
-	if r.Method != http.MethodPost {
-		t.Fatalf("got %q method, expected %q", r.Method, http.MethodPost)
-	}
-	if r.Header.Get("Content-Encoding") != "snappy" {
-		t.Fatalf("got %q content encoding header, expected %q", r.Header.Get("Content-Encoding"), "snappy")
-	}
-	if r.Header.Get("Content-Type") != "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse" {
-		t.Fatalf("got %q content type header, expected %q", r.Header.Get("Content-Type"), "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse")
-	}
-
-	remoteReadVersion := r.Header.Get("X-Prometheus-Remote-Read-Version")
-	if remoteReadVersion == "" {
-		t.Fatalf("got empty prometheus remote read header")
-	}
-	if !strings.HasPrefix(remoteReadVersion, "0.1.") {
-		t.Fatalf("wrong remote version defined")
-	}
-	return true
-}
-
-func GenerateRemoteReadSeries(start, end, numOfSeries, numOfSamples int64) []*prompb.TimeSeries {
-	var ts []*prompb.TimeSeries
-	j := 0
-	for i := 0; i < int(numOfSeries); i++ {
-		if i%3 == 0 {
-			j++
-		}
-
-		timeSeries := prompb.TimeSeries{
-			Labels: []prompb.Label{
-				{Name: labels.MetricName, Value: fmt.Sprintf("vm_metric_%d", j)},
-				{Name: "job", Value: strconv.Itoa(i)},
-			},
-		}
-
-		ts = append(ts, &timeSeries)
-	}
-
-	for i := range ts {
-		ts[i].Samples = generateRemoteReadSamples(i, start, end, numOfSamples)
-	}
-
-	return ts
-}
-
-func generateRemoteReadSamples(idx int, startTime, endTime, numOfSamples int64) []prompb.Sample {
-	samples := make([]prompb.Sample, 0)
-	delta := (endTime - startTime) / numOfSamples
-
-	t := startTime
-	for t != endTime {
-		v := 100 * int64(idx)
-		samples = append(samples, prompb.Sample{
-			Timestamp: t * 1000,
-			Value:     float64(v),
-		})
-		t = t + delta
-	}
-
-	return samples
-}
-
-type MockStorage struct {
-	query *prompb.Query
-	store []*prompb.TimeSeries
-}
-
-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")
-	}
-	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))
-	for _, s := range ms.store {
-		var samples []prompb.Sample
-		for _, sample := range s.Samples {
-			if sample.Timestamp >= query.StartTimestampMs && sample.Timestamp < query.EndTimestampMs {
-				samples = append(samples, sample)
-			}
-		}
-		var series prompb.TimeSeries
-		if len(samples) > 0 {
-			series.Labels = s.Labels
-			series.Samples = samples
-		}
-
-		tss = append(tss, &series)
-	}
-	return &mockSeriesSet{
-		tss: tss,
-	}, 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 {
-		result = append(result, prompb.Label{
-			Name:  l.Name,
-			Value: l.Value,
-		})
-	}
-	return result
-}
+// import (
+// 	"context"
+// 	"fmt"
+// 	"io"
+// 	"net/http"
+// 	"net/http/httptest"
+// 	"strconv"
+// 	"strings"
+// 	"testing"
+//
+// 	"github.com/gogo/protobuf/proto"
+// 	"github.com/golang/snappy"
+// 	"github.com/prometheus/prometheus/model/labels"
+// 	"github.com/prometheus/prometheus/prompb"
+// 	"github.com/prometheus/prometheus/storage/remote"
+// 	"github.com/prometheus/prometheus/tsdb/chunks"
+// )
+//
+// const (
+// 	maxBytesInFrame = 1024 * 1024
+// )
+//
+// type RemoteReadServer struct {
+// 	server  *httptest.Server
+// 	series  []*prompb.TimeSeries
+// 	storage *MockStorage
+// }
+//
+// // NewRemoteReadServer creates a remote read server. It exposes a single endpoint and responds with the
+// // passed series based on the request to the read endpoint. It returns a server which should be closed after
+// // being used.
+// func NewRemoteReadServer(t *testing.T) *RemoteReadServer {
+// 	rrs := &RemoteReadServer{
+// 		series: make([]*prompb.TimeSeries, 0),
+// 	}
+// 	rrs.server = httptest.NewServer(rrs.getReadHandler(t))
+// 	return rrs
+// }
+//
+// // Close closes the server.
+// func (rrs *RemoteReadServer) Close() {
+// 	rrs.server.Close()
+// }
+//
+// func (rrs *RemoteReadServer) URL() string {
+// 	return rrs.server.URL
+// }
+//
+// func (rrs *RemoteReadServer) SetRemoteReadSeries(series []*prompb.TimeSeries) {
+// 	rrs.series = append(rrs.series, series...)
+// }
+//
+// func (rrs *RemoteReadServer) getReadHandler(t *testing.T) http.Handler {
+// 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+// 		if !validateReadHeaders(t, r) {
+// 			t.Fatalf("invalid read headers")
+// 		}
+//
+// 		compressed, err := io.ReadAll(r.Body)
+// 		if err != nil {
+// 			t.Fatalf("error read body: %s", err)
+// 		}
+//
+// 		reqBuf, err := snappy.Decode(nil, compressed)
+// 		if err != nil {
+// 			t.Fatalf("error decode compressed data:%s", err)
+// 		}
+//
+// 		var req prompb.ReadRequest
+// 		if err := proto.Unmarshal(reqBuf, &req); err != nil {
+// 			t.Fatalf("error unmarshal read request: %s", err)
+// 		}
+//
+// 		resp := &prompb.ReadResponse{
+// 			Results: make([]*prompb.QueryResult, len(req.Queries)),
+// 		}
+//
+// 		for i, r := range req.Queries {
+// 			startTs := r.StartTimestampMs
+// 			endTs := r.EndTimestampMs
+// 			ts := make([]*prompb.TimeSeries, len(rrs.series))
+// 			for i, s := range rrs.series {
+// 				var samples []prompb.Sample
+// 				for _, sample := range s.Samples {
+// 					if sample.Timestamp >= startTs && sample.Timestamp < endTs {
+// 						samples = append(samples, sample)
+// 					}
+// 				}
+// 				var series prompb.TimeSeries
+// 				if len(samples) > 0 {
+// 					series.Labels = s.Labels
+// 					series.Samples = samples
+// 				}
+// 				ts[i] = &series
+// 			}
+//
+// 			resp.Results[i] = &prompb.QueryResult{Timeseries: ts}
+// 			data, err := proto.Marshal(resp)
+// 			if err != nil {
+// 				t.Fatalf("error marshal response: %s", err)
+// 			}
+//
+// 			compressed = snappy.Encode(nil, data)
+//
+// 			w.Header().Set("Content-Type", "application/x-protobuf")
+// 			w.Header().Set("Content-Encoding", "snappy")
+// 			w.WriteHeader(http.StatusOK)
+//
+// 			if _, err := w.Write(compressed); err != nil {
+// 				t.Fatalf("snappy encode error: %s", err)
+// 			}
+// 		}
+// 	})
+// }
+//
+// func NewRemoteReadStreamServer(t *testing.T) *RemoteReadServer {
+// 	rrs := &RemoteReadServer{
+// 		series: make([]*prompb.TimeSeries, 0),
+// 	}
+// 	rrs.server = httptest.NewServer(rrs.getStreamReadHandler(t))
+// 	return rrs
+// }
+//
+// func (rrs *RemoteReadServer) InitMockStorage(series []*prompb.TimeSeries) {
+// 	rrs.storage = NewMockStorage(series)
+// }
+//
+// func (rrs *RemoteReadServer) getStreamReadHandler(t *testing.T) http.Handler {
+// 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+// 		if !validateStreamReadHeaders(t, r) {
+// 			t.Fatalf("invalid read headers")
+// 		}
+//
+// 		f, ok := w.(http.Flusher)
+// 		if !ok {
+// 			t.Fatalf("internal http.ResponseWriter does not implement http.Flusher interface")
+// 		}
+//
+// 		stream := remote.NewChunkedWriter(w, f)
+//
+// 		data, err := io.ReadAll(r.Body)
+// 		if err != nil {
+// 			t.Fatalf("error read body: %s", err)
+// 		}
+//
+// 		decodedData, err := snappy.Decode(nil, data)
+// 		if err != nil {
+// 			t.Fatalf("error decode compressed data:%s", err)
+// 		}
+//
+// 		var req prompb.ReadRequest
+// 		if err := proto.Unmarshal(decodedData, &req); err != nil {
+// 			t.Fatalf("error unmarshal read request: %s", err)
+// 		}
+//
+// 		var chks []prompb.Chunk
+// 		ctx := context.Background()
+// 		for idx, r := range req.Queries {
+// 			startTs := r.StartTimestampMs
+// 			endTs := r.EndTimestampMs
+//
+// 			var matchers []*labels.Matcher
+// 			cb := func() (int64, error) { return 0, nil }
+//
+// 			c := remote.NewSampleAndChunkQueryableClient(rrs.storage, nil, matchers, true, cb)
+//
+// 			q, err := c.ChunkQuerier(startTs, endTs)
+// 			if err != nil {
+// 				t.Fatalf("error init chunk querier: %s", err)
+// 			}
+//
+// 			ss := q.Select(ctx, false, nil, matchers...)
+// 			var iter chunks.Iterator
+// 			for ss.Next() {
+// 				series := ss.At()
+// 				iter = series.Iterator(iter)
+// 				labels := remote.MergeLabels(labelsToLabelsProto(series.Labels()), nil)
+//
+// 				frameBytesLeft := maxBytesInFrame
+// 				for _, lb := range labels {
+// 					frameBytesLeft -= lb.Size()
+// 				}
+//
+// 				isNext := iter.Next()
+//
+// 				for isNext {
+// 					chunk := iter.At()
+//
+// 					if chunk.Chunk == nil {
+// 						t.Fatalf("error found not populated chunk returned by SeriesSet at ref: %v", chunk.Ref)
+// 					}
+//
+// 					chks = append(chks, prompb.Chunk{
+// 						MinTimeMs: chunk.MinTime,
+// 						MaxTimeMs: chunk.MaxTime,
+// 						Type:      prompb.Chunk_Encoding(chunk.Chunk.Encoding()),
+// 						Data:      chunk.Chunk.Bytes(),
+// 					})
+//
+// 					frameBytesLeft -= chks[len(chks)-1].Size()
+//
+// 					// We are fine with minor inaccuracy of max bytes per frame. The inaccuracy will be max of full chunk size.
+// 					isNext = iter.Next()
+// 					if frameBytesLeft > 0 && isNext {
+// 						continue
+// 					}
+//
+// 					resp := &prompb.ChunkedReadResponse{
+// 						ChunkedSeries: []*prompb.ChunkedSeries{
+// 							{Labels: labels, Chunks: chks},
+// 						},
+// 						QueryIndex: int64(idx),
+// 					}
+//
+// 					b, err := proto.Marshal(resp)
+// 					if err != nil {
+// 						t.Fatalf("error marshal response: %s", err)
+// 					}
+//
+// 					if _, err := stream.Write(b); err != nil {
+// 						t.Fatalf("error write to stream: %s", err)
+// 					}
+// 					chks = chks[:0]
+// 					rrs.storage.Reset()
+// 				}
+// 				if err := iter.Err(); err != nil {
+// 					t.Fatalf("error iterate over chunk series: %s", err)
+// 				}
+// 			}
+// 		}
+// 	})
+// }
+//
+// func validateReadHeaders(t *testing.T, r *http.Request) bool {
+// 	if r.Method != http.MethodPost {
+// 		t.Fatalf("got %q method, expected %q", r.Method, http.MethodPost)
+// 	}
+// 	if r.Header.Get("Content-Encoding") != "snappy" {
+// 		t.Fatalf("got %q content encoding header, expected %q", r.Header.Get("Content-Encoding"), "snappy")
+// 	}
+// 	if r.Header.Get("Content-Type") != "application/x-protobuf" {
+// 		t.Fatalf("got %q content type header, expected %q", r.Header.Get("Content-Type"), "application/x-protobuf")
+// 	}
+//
+// 	remoteReadVersion := r.Header.Get("X-Prometheus-Remote-Read-Version")
+// 	if remoteReadVersion == "" {
+// 		t.Fatalf("got empty prometheus remote read header")
+// 	}
+// 	if !strings.HasPrefix(remoteReadVersion, "0.1.") {
+// 		t.Fatalf("wrong remote version defined")
+// 	}
+//
+// 	return true
+// }
+//
+// func validateStreamReadHeaders(t *testing.T, r *http.Request) bool {
+// 	if r.Method != http.MethodPost {
+// 		t.Fatalf("got %q method, expected %q", r.Method, http.MethodPost)
+// 	}
+// 	if r.Header.Get("Content-Encoding") != "snappy" {
+// 		t.Fatalf("got %q content encoding header, expected %q", r.Header.Get("Content-Encoding"), "snappy")
+// 	}
+// 	if r.Header.Get("Content-Type") != "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse" {
+// 		t.Fatalf("got %q content type header, expected %q", r.Header.Get("Content-Type"), "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse")
+// 	}
+//
+// 	remoteReadVersion := r.Header.Get("X-Prometheus-Remote-Read-Version")
+// 	if remoteReadVersion == "" {
+// 		t.Fatalf("got empty prometheus remote read header")
+// 	}
+// 	if !strings.HasPrefix(remoteReadVersion, "0.1.") {
+// 		t.Fatalf("wrong remote version defined")
+// 	}
+// 	return true
+// }
+//
+// func GenerateRemoteReadSeries(start, end, numOfSeries, numOfSamples int64) []*prompb.TimeSeries {
+// 	var ts []*prompb.TimeSeries
+// 	j := 0
+// 	for i := 0; i < int(numOfSeries); i++ {
+// 		if i%3 == 0 {
+// 			j++
+// 		}
+//
+// 		timeSeries := prompb.TimeSeries{
+// 			Labels: []prompb.Label{
+// 				{Name: labels.MetricName, Value: fmt.Sprintf("vm_metric_%d", j)},
+// 				{Name: "job", Value: strconv.Itoa(i)},
+// 			},
+// 		}
+//
+// 		ts = append(ts, &timeSeries)
+// 	}
+//
+// 	for i := range ts {
+// 		ts[i].Samples = generateRemoteReadSamples(i, start, end, numOfSamples)
+// 	}
+//
+// 	return ts
+// }
+//
+// func generateRemoteReadSamples(idx int, startTime, endTime, numOfSamples int64) []prompb.Sample {
+// 	samples := make([]prompb.Sample, 0)
+// 	delta := (endTime - startTime) / numOfSamples
+//
+// 	t := startTime
+// 	for t != endTime {
+// 		v := 100 * int64(idx)
+// 		samples = append(samples, prompb.Sample{
+// 			Timestamp: t * 1000,
+// 			Value:     float64(v),
+// 		})
+// 		t = t + delta
+// 	}
+//
+// 	return samples
+// }
+//
+// type MockStorage struct {
+// 	query *prompb.Query
+// 	store []*prompb.TimeSeries
+// }
+//
+// func NewMockStorage(series []*prompb.TimeSeries) *MockStorage {
+// 	return &MockStorage{store: series}
+// }
+//
+// 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
+//
+// 	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 {
+// 			if sample.Timestamp >= query.StartTimestampMs && sample.Timestamp < query.EndTimestampMs {
+// 				samples = append(samples, sample)
+// 			}
+// 		}
+// 		var series prompb.TimeSeries
+// 		if len(samples) > 0 {
+// 			series.Labels = s.Labels
+// 			series.Samples = samples
+// 		}
+//
+// 		q.Timeseries = append(q.Timeseries, &series)
+// 	}
+// 	return q, nil
+// }
+//
+// func (ms *MockStorage) Reset() {
+// 	ms.query = nil
+// }
+//
+// func labelsToLabelsProto(labels labels.Labels) []prompb.Label {
+// 	result := make([]prompb.Label, 0, len(labels))
+// 	for _, l := range labels {
+// 		result = append(result, prompb.Label{
+// 			Name:  l.Name,
+// 			Value: l.Value,
+// 		})
+// 	}
+// 	return result
+// }

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/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.
@@ -92,27 +59,7 @@ 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 {
 	metricNameRaw := ctx.marshalMetricNameRaw(prefix, labels)
 	return ctx.addRow(metricNameRaw, timestamp, value)
@@ -120,8 +67,6 @@ func (ctx *InsertCtx) WriteDataPoint(prefix []byte, labels []prompbmarshal.Label
 
 // 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) {
 	if len(metricNameRaw) == 0 {
@@ -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

@@ -12,8 +12,8 @@ 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)
 	}

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

@@ -14,7 +14,6 @@ import (
 	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 +69,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 +108,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 +130,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

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
 		}
 	}

app/vminsert/newrelic/request_handler.go

@@ -58,9 +58,14 @@ func insertRows(rows []newrelic.Row, extraLabels []prompbmarshal.Label) error {
 				label := &extraLabels[k]
 				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, s.Value); err != nil {
 				return err
 			}

app/vminsert/opentelemetry/request_handler.go

@@ -59,9 +59,14 @@ func insertRows(tss []prompbmarshal.TimeSeries, extraLabels []prompbmarshal.Labe
 		for _, label := range extraLabels {
 			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
 		samples := ts.Samples

app/vminsert/opentsdb/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/opentsdbhttp/request_handler.go

@@ -54,9 +54,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/prometheusimport/request_handler.go

@@ -54,9 +54,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/prompush/push.go

@@ -57,9 +57,12 @@ func push(ctx *common.InsertCtx, tss []prompbmarshal.TimeSeries) {
 			label := &ts.Labels[j]
 			ctx.AddLabel(label.Name, label.Value)
 		}
-		if !ctx.TryPrepareLabels(false) {
+		ctx.ApplyRelabeling()
+		if len(ctx.Labels) == 0 {
+			// Skip metric without labels.
 			continue
 		}
+		ctx.SortLabelsIfNeeded()
 		var metricNameRaw []byte
 		var err error
 		for i := range ts.Samples {

app/vminsert/promremotewrite/request_handler.go

@@ -52,10 +52,14 @@ func insertRows(timeseries []prompb.TimeSeries, extraLabels []prompbmarshal.Labe
 			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
 		samples := ts.Samples

app/vminsert/vmimport/request_handler.go

@@ -58,9 +58,14 @@ func insertRows(rows []parser.Row, 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.
 			continue
 		}
+		ic.SortLabelsIfNeeded()
 		ctx.metricNameBuf = storage.MarshalMetricNameRaw(ctx.metricNameBuf[:0], ic.Labels)
 		values := r.Values
 		timestamps := r.Timestamps

app/vmselect/promql/rollup.go

@@ -699,13 +699,8 @@ func (rc *rollupConfig) doInternal(dstValues []float64, tsm *timeseriesMap, valu
 	// Extend dstValues in order to remove mallocs below.
 	dstValues = decimal.ExtendFloat64sCapacity(dstValues, len(rc.Timestamps))
 
-	// Use step as the scrape interval for instant queries (when start == end).
-	maxPrevInterval := rc.Step
-	if rc.Start < rc.End {
-		scrapeInterval := getScrapeInterval(timestamps, rc.Step)
-		maxPrevInterval = getMaxPrevInterval(scrapeInterval)
-	}
-
+	scrapeInterval := getScrapeInterval(timestamps, rc.Step)
+	maxPrevInterval := getMaxPrevInterval(scrapeInterval)
 	if rc.LookbackDelta > 0 && maxPrevInterval > rc.LookbackDelta {
 		maxPrevInterval = rc.LookbackDelta
 	}

app/vmselect/vmui/asset-manifest.json

@@ -1,13 +1,13 @@
 {
   "files": {
-    "main.css": "./static/css/main.876c56b7.css",
-    "main.js": "./static/js/main.caf36c39.js",
+    "main.css": "./static/css/main.b1929c64.css",
+    "main.js": "./static/js/main.a7d57628.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.876c56b7.css",
-    "static/js/main.caf36c39.js"
+    "static/css/main.b1929c64.css",
+    "static/js/main.a7d57628.js"
   ]
 }

app/vmselect/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 and troubleshoot your VictoriaMetrics data"/><link rel="manifest" href="./manifest.json"/><title>vmui</title><script src="./dashboards/index.js" type="module"></script><meta name="twitter:card" content="summary"><meta name="twitter:title" content="UI for VictoriaMetrics"><meta name="twitter:site" content="@https://victoriametrics.com/"><meta name="twitter:description" content="Explore and troubleshoot your VictoriaMetrics data"><meta name="twitter:image" content="./preview.jpg"><meta property="og:type" content="website"><meta property="og:title" content="UI for VictoriaMetrics"><meta property="og:url" content="https://victoriametrics.com/"><meta property="og:description" content="Explore and troubleshoot your VictoriaMetrics data"><script defer="defer" src="./static/js/main.caf36c39.js"></script><link href="./static/css/main.876c56b7.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 and troubleshoot your VictoriaMetrics data"/><link rel="manifest" href="./manifest.json"/><title>vmui</title><script src="./dashboards/index.js" type="module"></script><meta name="twitter:card" content="summary"><meta name="twitter:title" content="UI for VictoriaMetrics"><meta name="twitter:site" content="@https://victoriametrics.com/"><meta name="twitter:description" content="Explore and troubleshoot your VictoriaMetrics data"><meta name="twitter:image" content="./preview.jpg"><meta property="og:type" content="website"><meta property="og:title" content="UI for VictoriaMetrics"><meta property="og:url" content="https://victoriametrics.com/"><meta property="og:description" content="Explore and troubleshoot your VictoriaMetrics data"><script defer="defer" src="./static/js/main.a7d57628.js"></script><link href="./static/css/main.b1929c64.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>

app/vmui/Dockerfile-web

@@ -1,4 +1,4 @@
-FROM golang:1.23.4 AS build-web-stage
+FROM golang:1.23.3 AS build-web-stage
 COPY build /build
 
 WORKDIR /build
@@ -6,7 +6,7 @@ COPY web/ /build/
 RUN GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -o web-amd64 github.com/VictoriMetrics/vmui/ && \
     GOOS=windows GOARCH=amd64 CGO_ENABLED=0 go build -o web-windows github.com/VictoriMetrics/vmui/
 
-FROM alpine:3.21.0
+FROM alpine:3.20.3
 USER root
 
 COPY --from=build-web-stage /build/web-amd64 /app/web

app/vmui/packages/vmui/src/api/types.ts

@@ -48,11 +48,3 @@ export interface LogHits {
     [key: string]: string;
   };
 }
-
-export interface ReportMetaData {
-  id: number;
-  title: string;
-  endpoint: string;
-  comment: string;
-  params: Record<string, string>;
-}

app/vmui/packages/vmui/src/components/Configurators/QueryEditor/QueryEditor.tsx

@@ -43,8 +43,7 @@ const QueryEditor: FC<QueryEditorProps> = ({
   const { isMobile } = useDeviceDetect();
 
   const [openAutocomplete, setOpenAutocomplete] = useState(false);
-  const [caretPositionAutocomplete, setCaretPositionAutocomplete] = useState<[number, number]>([0, 0]);
-  const [caretPositionInput, setCaretPositionInput] = useState<[number, number]>([0, 0]);
+  const [caretPosition, setCaretPosition] = useState<[number, number]>([0, 0]);
   const autocompleteAnchorEl = useRef<HTMLInputElement>(null);
 
   const [showAutocomplete, setShowAutocomplete] = useState(autocomplete);
@@ -67,7 +66,7 @@ const QueryEditor: FC<QueryEditorProps> = ({
 
   const handleSelect = (val: string, caretPosition: number) => {
     onChange(val);
-    setCaretPositionInput([caretPosition, caretPosition]);
+    setCaretPosition([caretPosition, caretPosition]);
   };
 
   const handleKeyDown = (e: KeyboardEvent) => {
@@ -109,7 +108,7 @@ const QueryEditor: FC<QueryEditorProps> = ({
   };
 
   const handleChangeCaret = (val: [number, number]) => {
-    setCaretPositionAutocomplete(prev => prev[0] === val[0] && prev[1] === val[1] ? prev : val);
+    setCaretPosition(prev => prev[0] === val[0] && prev[1] === val[1] ? prev : val);
   };
 
   useEffect(() => {
@@ -119,7 +118,7 @@ const QueryEditor: FC<QueryEditorProps> = ({
   useEffect(() => {
     setShowAutocomplete(false);
     debouncedSetShowAutocomplete(true);
-  }, [caretPositionAutocomplete]);
+  }, [caretPosition]);
 
   return (
     <div
@@ -138,13 +137,13 @@ const QueryEditor: FC<QueryEditorProps> = ({
         onChangeCaret={handleChangeCaret}
         disabled={disabled}
         inputmode={"search"}
-        caretPosition={caretPositionInput}
+        caretPosition={caretPosition}
       />
       {showAutocomplete && autocomplete && (
         <QueryEditorAutocomplete
           value={value}
           anchorEl={autocompleteAnchorEl}
-          caretPosition={caretPositionAutocomplete}
+          caretPosition={caretPosition}
           hasHelperText={Boolean(warning || error)}
           includeFunctions={includeFunctions}
           onSelect={handleSelect}

app/vmui/packages/vmui/src/components/ExploreMetrics/ExploreMetricsHeader/style.scss

@@ -38,10 +38,6 @@
     align-items: flex-start;
     gap: $padding-small;
 
-    ul {
-      list-style-position: inside;
-    }
-
     button {
       color: inherit;
       min-height: 29px;

app/vmui/packages/vmui/src/components/Main/Accordion/Accordion.tsx

@@ -19,10 +19,6 @@ const Accordion: FC<AccordionProps> = ({
   const [isOpen, setIsOpen] = useState(defaultExpanded);
 
   const toggleOpen = () => {
-    const selection = window.getSelection();
-    if (selection && selection.toString()) {
-      return; // If the text is selected, cancel the execution of toggle.
-    }
     setIsOpen(prev => !prev);
   };
 

app/vmui/packages/vmui/src/components/Main/Alert/style.scss

@@ -46,8 +46,6 @@
     display: flex;
     align-items: center;
     justify-content: center;
-    align-self: flex-start;
-    min-height: 24px;
   }
 
   &__content {

app/vmui/packages/vmui/src/components/Main/Icons/index.tsx

@@ -570,14 +570,3 @@ export const SpinnerIcon = () => (
     </path>
   </svg>
 );
-
-export const CommentIcon = () => (
-  <svg
-    viewBox="0 0 24 24"
-    fill="currentColor"
-  >
-    <path
-      d="M21.99 4c0-1.1-.89-2-1.99-2H4c-1.1 0-2 .9-2 2v12c0 1.1.9 2 2 2h14l4 4zM18 14H6v-2h12zm0-3H6V9h12zm0-3H6V6h12z"
-    ></path>
-  </svg>
-);

app/vmui/packages/vmui/src/components/Main/MarkdownEditor/MarkdownEditor.tsx

@@ -1,62 +0,0 @@
-import React, { FC } from "preact/compat";
-import useBoolean from "../../../hooks/useBoolean";
-import classNames from "classnames";
-import TextField from "../TextField/TextField";
-import "./style.scss";
-import { marked } from "marked";
-
-interface Props {
-  value: string;
-  onChange: (value: string) => void;
-}
-
-const tabs = [
-  { title: "Write", value: false },
-  { title: "Preview", value: true },
-];
-
-const MarkdownEditor: FC<Props> = ({ value, onChange }) => {
-  const {
-    value: markdownPreview,
-    setTrue: setMarkdownPreviewTrue,
-    setFalse: setMarkdownPreviewFalse,
-  } = useBoolean(false);
-
-  return (
-    <div className="vm-markdown-editor">
-      <div className="vm-markdown-editor-header">
-        <div className="vm-markdown-editor-header-tabs">
-          {tabs.map(({ title, value }) => (
-            <div
-              key={title}
-              className={classNames({
-                "vm-markdown-editor-header-tabs__tab": true,
-                "vm-markdown-editor-header-tabs__tab_active": markdownPreview === value,
-              })}
-              onClick={value ? setMarkdownPreviewTrue : setMarkdownPreviewFalse}
-            >
-              {title}
-            </div>
-          ))}
-        </div>
-        <span className="vm-markdown-editor-header__info">
-          Markdown is supported
-        </span>
-      </div>
-      {markdownPreview ? (
-        <div
-          className="vm-markdown-editor-preview vm-markdown"
-          dangerouslySetInnerHTML={{ __html: marked(value) as string }}
-        />
-      ) : (
-        <TextField
-          type="textarea"
-          value={value}
-          onChange={onChange}
-        />
-      )}
-    </div>
-  );
-};
-
-export default MarkdownEditor;