CHANGELOG.md: cut v1.115.0 release

This commit properly reset aggregator state. Previously, it was not checked for `nil` and it lead to the panic on access.

Related issue:
https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8634

.github/workflows/docs.yaml

@@ -40,9 +40,8 @@ jobs:
       - name: Copy docs
         id: update
         run: |
-          rsync -zarv \
-            --exclude="Makefile" \
-            docs/ ../__vm-docs/content/victoriametrics
+          find docs -type d -maxdepth 1 -mindepth 1 -exec \
+            sh -c 'rsync -zarvh --delete {}/ ../__vm-docs/content/$(basename {})/' \;
           echo "SHORT_SHA=$(git rev-parse --short $GITHUB_SHA)" >> $GITHUB_OUTPUT
         working-directory: __vm
 

Makefile

@@ -18,7 +18,7 @@ TAR_OWNERSHIP ?= --owner=1000 --group=1000
 .PHONY: $(MAKECMDGOALS)
 
 include app/*/Makefile
-include cspell/Makefile
+include codespell/Makefile
 include docs/Makefile
 include deployment/*/Makefile
 include dashboards/Makefile
@@ -567,7 +567,7 @@ golangci-lint: install-golangci-lint
 	golangci-lint run
 
 install-golangci-lint:
-	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.64.5
+	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.64.7
 
 remove-golangci-lint:
 	rm -rf `which golangci-lint`

app/vlinsert/datadog/datadog.go

@@ -3,7 +3,6 @@ package datadog
 import (
 	"bytes"
 	"fmt"
-	"io"
 	"net/http"
 	"strconv"
 	"time"
@@ -11,20 +10,22 @@ import (
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/valyala/fastjson"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"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"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 )
 
 var (
-	datadogStreamFields = flagutil.NewArrayString("datadog.streamFields", "Datadog tags to be used as stream fields.")
-	datadogIgnoreFields = flagutil.NewArrayString("datadog.ignoreFields", "Datadog tags to ignore.")
+	datadogStreamFields = flagutil.NewArrayString("datadog.streamFields", "Comma-separated list of fields to use as log stream fields for logs ingested via DataDog protocol. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/datadog-agent/#stream-fields")
+	datadogIgnoreFields = flagutil.NewArrayString("datadog.ignoreFields", "Comma-separated list of fields to ignore for logs ingested via DataDog protocol. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/datadog-agent/#dropping-fields")
+
+	maxRequestSize = flagutil.NewBytes("datadog.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single DataDog request")
 )
 
 var parserPool fastjson.ParserPool
@@ -46,7 +47,6 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 	w.Header().Add("Content-Type", "application/json")
 	startTime := time.Now()
 	v2LogsRequestsTotal.Inc()
-	reader := r.Body
 
 	var ts int64
 	if tsValue := r.Header.Get("dd-message-timestamp"); tsValue != "" && tsValue != "0" {
@@ -61,25 +61,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 		ts = startTime.UnixNano()
 	}
 
-	if r.Header.Get("Content-Encoding") == "gzip" {
-		zr, err := common.GetGzipReader(reader)
-		if err != nil {
-			httpserver.Errorf(w, r, "cannot read gzipped logs request: %s", err)
-			return true
-		}
-		defer common.PutGzipReader(zr)
-		reader = zr
-	}
-
-	wcr := writeconcurrencylimiter.GetReader(reader)
-	data, err := io.ReadAll(wcr)
-	writeconcurrencylimiter.PutReader(wcr)
-	if err != nil {
-		httpserver.Errorf(w, r, "cannot read request body: %s", err)
-		return true
-	}
-
-	cp, err := insertutils.GetCommonParams(r)
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return true
@@ -97,11 +79,15 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 
-	lmp := cp.NewLogMessageProcessor("datadog")
-	err = readLogsRequest(ts, data, lmp)
-	lmp.MustClose()
+	encoding := r.Header.Get("Content-Encoding")
+	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
+		lmp := cp.NewLogMessageProcessor("datadog", false)
+		err := readLogsRequest(ts, data, lmp)
+		lmp.MustClose()
+		return err
+	})
 	if err != nil {
-		logger.Warnf("cannot decode log message in /api/v2/logs request: %s, stream fields: %s", err, cp.StreamFields)
+		httpserver.Errorf(w, r, "cannot read DataDog protocol data: %s", err)
 		return true
 	}
 
@@ -121,7 +107,7 @@ var (
 // datadog message field has two formats:
 //   - regular log message with string text
 //   - nested json format for serverless plugins
-//     which has folowing format:
+//     which has the following 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
@@ -184,7 +170,7 @@ func appendMsgFields(fields []logstorage.Field, v *fastjson.Value) ([]logstorage
 
 // 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, lmp insertutil.LogMessageProcessor) error {
 	p := parserPool.Get()
 	defer parserPool.Put(p)
 	v, err := p.ParseBytes(data)

app/vlinsert/datadog/datadog_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func TestReadLogsRequestFailure(t *testing.T) {
@@ -13,7 +13,7 @@ func TestReadLogsRequestFailure(t *testing.T) {
 
 		ts := time.Now().UnixNano()
 
-		lmp := &insertutils.TestLogMessageProcessor{}
+		lmp := &insertutil.TestLogMessageProcessor{}
 		if err := readLogsRequest(ts, []byte(data), lmp); err == nil {
 			t.Fatalf("expecting non-empty error")
 		}
@@ -37,7 +37,7 @@ func TestReadLogsRequestSuccess(t *testing.T) {
 		for i := 0; i < rowsExpected; i++ {
 			timestampsExpected = append(timestampsExpected, ts)
 		}
-		lmp := &insertutils.TestLogMessageProcessor{}
+		lmp := &insertutil.TestLogMessageProcessor{}
 		if err := readLogsRequest(ts, []byte(data), lmp); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}

app/vlinsert/elasticsearch/elasticsearch.go

@@ -10,14 +10,14 @@ import (
 
 	"github.com/VictoriaMetrics/metrics"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bufferedwriter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 )
 
@@ -90,7 +90,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 		startTime := time.Now()
 		bulkRequestsTotal.Inc()
 
-		cp, err := insertutils.GetCommonParams(r)
+		cp, err := insertutil.GetCommonParams(r)
 		if err != nil {
 			httpserver.Errorf(w, r, "%s", err)
 			return true
@@ -99,10 +99,10 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 			httpserver.Errorf(w, r, "%s", err)
 			return true
 		}
-		lmp := cp.NewLogMessageProcessor("elasticsearch_bulk")
-		isGzip := r.Header.Get("Content-Encoding") == "gzip"
+		lmp := cp.NewLogMessageProcessor("elasticsearch_bulk", true)
+		encoding := r.Header.Get("Content-Encoding")
 		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(streamName, r.Body, encoding, 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)
@@ -131,22 +131,19 @@ var (
 	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(streamName string, r io.Reader, encoding string, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) (int, error) {
 	// See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
 
-	if isGzip {
-		zr, err := common.GetGzipReader(r)
-		if err != nil {
-			return 0, fmt.Errorf("cannot read gzipped _bulk request: %w", err)
-		}
-		defer common.PutGzipReader(zr)
-		r = zr
+	reader, err := protoparserutil.GetUncompressedReader(r, encoding)
+	if err != nil {
+		return 0, fmt.Errorf("cannot decode Elasticsearch protocol data: %w", err)
 	}
+	defer protoparserutil.PutUncompressedReader(reader)
 
-	wcr := writeconcurrencylimiter.GetReader(r)
+	wcr := writeconcurrencylimiter.GetReader(reader)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
-	lr := insertutils.NewLineReader(streamName, wcr)
+	lr := insertutil.NewLineReader(streamName, wcr)
 
 	n := 0
 	for {
@@ -159,7 +156,7 @@ func readBulkRequest(streamName string, r io.Reader, isGzip bool, timeField stri
 	}
 }
 
-func readBulkLine(lr *insertutils.LineReader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (bool, error) {
+func readBulkLine(lr *insertutil.LineReader, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) (bool, error) {
 	var line []byte
 
 	// Read the command, must be "create" or "index"
@@ -231,7 +228,7 @@ func parseElasticsearchTimestamp(s string) (int64, error) {
 	}
 	if len(s) < len("YYYY-MM-DD") || s[len("YYYY")] != '-' {
 		// Try parsing timestamp in seconds or milliseconds
-		return insertutils.ParseUnixTimestamp(s)
+		return insertutil.ParseUnixTimestamp(s)
 	}
 	if len(s) == len("YYYY-MM-DD") {
 		t, err := time.Parse("2006-01-02", s)

app/vlinsert/elasticsearch/elasticsearch_test.go

@@ -2,20 +2,24 @@ package elasticsearch
 
 import (
 	"bytes"
-	"compress/gzip"
 	"fmt"
+	"github.com/golang/snappy"
+	"github.com/klauspost/compress/gzip"
+	"github.com/klauspost/compress/zlib"
+	"github.com/klauspost/compress/zstd"
+	"io"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func TestReadBulkRequest_Failure(t *testing.T) {
 	f := func(data string) {
 		t.Helper()
 
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
-		rows, err := readBulkRequest("test", r, false, "_time", []string{"_msg"}, tlp)
+		rows, err := readBulkRequest("test", r, "", "_time", []string{"_msg"}, tlp)
 		if err == nil {
 			t.Fatalf("expecting non-empty error")
 		}
@@ -33,15 +37,15 @@ foobar`)
 }
 
 func TestReadBulkRequest_Success(t *testing.T) {
-	f := func(data, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
+	f := func(data, encoding, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
 		msgFields := []string{"non_existing_foo", msgField, "non_exiting_bar"}
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 
 		// Read the request without compression
 		r := bytes.NewBufferString(data)
-		rows, err := readBulkRequest("test", r, false, timeField, msgFields, tlp)
+		rows, err := readBulkRequest("test", r, "", timeField, msgFields, tlp)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -53,10 +57,12 @@ func TestReadBulkRequest_Success(t *testing.T) {
 		}
 
 		// Read the request with compression
-		tlp = &insertutils.TestLogMessageProcessor{}
-		compressedData := compressData(data)
-		r = bytes.NewBufferString(compressedData)
-		rows, err = readBulkRequest("test", r, true, timeField, msgFields, tlp)
+		tlp = &insertutil.TestLogMessageProcessor{}
+		if encoding != "" {
+			data = compressData(data, encoding)
+		}
+		r = bytes.NewBufferString(data)
+		rows, err = readBulkRequest("test", r, encoding, timeField, msgFields, tlp)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -69,9 +75,9 @@ func TestReadBulkRequest_Success(t *testing.T) {
 	}
 
 	// Verify an empty data
-	f("", "_time", "_msg", nil, "")
-	f("\n", "_time", "_msg", nil, "")
-	f("\n\n", "_time", "_msg", nil, "")
+	f("", "gzip", "_time", "_msg", nil, "")
+	f("\n", "gzip", "_time", "_msg", nil, "")
+	f("\n\n", "gzip", "_time", "_msg", nil, "")
 
 	// Verify non-empty data
 	data := `{"create":{"_index":"filebeat-8.8.0"}}
@@ -82,20 +88,35 @@ func TestReadBulkRequest_Success(t *testing.T) {
 {"message":"xyz","@timestamp":"1686026893735","x":"y"}
 {"create":{"_index":"filebeat-8.8.0"}}
 {"message":"qwe rty","@timestamp":"1686026893"}
+{"create":{"_index":"filebeat-8.8.0"}}
+{"message":"qwe rty float","@timestamp":"1686026123.62"}
 `
 	timeField := "@timestamp"
 	msgField := "message"
-	timestampsExpected := []int64{1686026891735000000, 1686023292735000000, 1686026893735000000, 1686026893000000000}
+	timestampsExpected := []int64{1686026891735000000, 1686023292735000000, 1686026893735000000, 1686026893000000000, 1686026123620000000}
 	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)
+{"_msg":"qwe rty"}
+{"_msg":"qwe rty float"}`
+	f(data, "zstd", timeField, msgField, timestampsExpected, resultExpected)
 }
 
-func compressData(s string) string {
+func compressData(s string, encoding string) string {
 	var bb bytes.Buffer
-	zw := gzip.NewWriter(&bb)
+	var zw io.WriteCloser
+	switch encoding {
+	case "gzip":
+		zw = gzip.NewWriter(&bb)
+	case "zstd":
+		zw, _ = zstd.NewWriter(&bb)
+	case "snappy":
+		zw = snappy.NewBufferedWriter(&bb)
+	case "deflate":
+		zw = zlib.NewWriter(&bb)
+	default:
+		panic(fmt.Errorf("%q encoding is not supported", encoding))
+	}
 	if _, err := zw.Write([]byte(s)); err != nil {
 		panic(fmt.Errorf("unexpected error when compressing data: %w", err))
 	}

app/vlinsert/elasticsearch/elasticsearch_timing_test.go

@@ -5,20 +5,29 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 )
 
 func BenchmarkReadBulkRequest(b *testing.B) {
-	b.Run("gzip:off", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, false)
+	b.Run("encoding:none", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, "")
 	})
-	b.Run("gzip:on", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, true)
+	b.Run("encoding:gzip", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, "gzip")
+	})
+	b.Run("encoding:zstd", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, "zstd")
+	})
+	b.Run("encoding:deflate", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, "deflate")
+	})
+	b.Run("encoding:snappy", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, "snappy")
 	})
 }
 
-func benchmarkReadBulkRequest(b *testing.B, isGzip bool) {
+func benchmarkReadBulkRequest(b *testing.B, encoding string) {
 	data := `{"create":{"_index":"filebeat-8.8.0"}}
 {"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
 {"create":{"_index":"filebeat-8.8.0"}}
@@ -26,14 +35,14 @@ func benchmarkReadBulkRequest(b *testing.B, isGzip bool) {
 {"create":{"_index":"filebeat-8.8.0"}}
 {"message":"xyz","@timestamp":"2023-06-06T04:48:13.735Z","x":"y"}
 `
-	if isGzip {
-		data = compressData(data)
+	if encoding != "" {
+		data = compressData(data, encoding)
 	}
 	dataBytes := bytesutil.ToUnsafeBytes(data)
 
 	timeField := "@timestamp"
 	msgFields := []string{"message"}
-	blp := &insertutils.BenchmarkLogMessageProcessor{}
+	blp := &insertutil.BenchmarkLogMessageProcessor{}
 
 	b.ReportAllocs()
 	b.SetBytes(int64(len(data)))
@@ -41,7 +50,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("test", r, encoding, timeField, msgFields, blp)
 			if err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}

app/vlinsert/insertutil/common_params.go

@@ -1,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"flag"
@@ -13,7 +13,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
@@ -49,13 +49,13 @@ func GetCommonParams(r *http.Request) (*CommonParams, error) {
 	}
 
 	timeField := "_time"
-	if tf := httputils.GetRequestValue(r, "_time_field", "VL-Time-Field"); tf != "" {
+	if tf := httputil.GetRequestValue(r, "_time_field", "VL-Time-Field"); tf != "" {
 		timeField = tf
 	}
 
-	msgFields := httputils.GetArray(r, "_msg_field", "VL-Msg-Field")
-	streamFields := httputils.GetArray(r, "_stream_fields", "VL-Stream-Fields")
-	ignoreFields := httputils.GetArray(r, "ignore_fields", "VL-Ignore-Fields")
+	msgFields := httputil.GetArray(r, "_msg_field", "VL-Msg-Field")
+	streamFields := httputil.GetArray(r, "_stream_fields", "VL-Stream-Fields")
+	ignoreFields := httputil.GetArray(r, "ignore_fields", "VL-Ignore-Fields")
 
 	extraFields, err := getExtraFields(r)
 	if err != nil {
@@ -63,7 +63,7 @@ func GetCommonParams(r *http.Request) (*CommonParams, error) {
 	}
 
 	debug := false
-	if dv := httputils.GetRequestValue(r, "debug", "VL-Debug"); dv != "" {
+	if dv := httputil.GetRequestValue(r, "debug", "VL-Debug"); dv != "" {
 		debug, err = strconv.ParseBool(dv)
 		if err != nil {
 			return nil, fmt.Errorf("cannot parse debug=%q: %w", dv, err)
@@ -92,7 +92,7 @@ func GetCommonParams(r *http.Request) (*CommonParams, error) {
 }
 
 func getExtraFields(r *http.Request) ([]logstorage.Field, error) {
-	efs := httputils.GetArray(r, "extra_fields", "VL-Extra-Fields")
+	efs := httputil.GetArray(r, "extra_fields", "VL-Extra-Fields")
 	if len(efs) == 0 {
 		return nil, nil
 	}
@@ -141,7 +141,7 @@ type LogMessageProcessor interface {
 	//
 	// If streamFields is non-nil, then the given streamFields must be used as log stream fields instead of pre-configured fields.
 	//
-	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can re-use them.
+	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can reuse them.
 	AddRow(timestamp int64, fields, streamFields []logstorage.Field)
 
 	// MustClose() must flush all the remaining fields and free up resources occupied by LogMessageProcessor.
@@ -191,9 +191,6 @@ func (lmp *logMessageProcessor) initPeriodicFlush() {
 //
 // 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) {
-	lmp.mu.Lock()
-	defer lmp.mu.Unlock()
-
 	lmp.rowsIngestedTotal.Inc()
 	n := logstorage.EstimatedJSONRowLen(fields)
 	lmp.bytesIngestedTotal.Add(n)
@@ -205,7 +202,11 @@ func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []l
 		return
 	}
 
+	lmp.mu.Lock()
+	defer lmp.mu.Unlock()
+
 	lmp.lr.MustAdd(lmp.cp.TenantID, timestamp, fields, streamFields)
+
 	if lmp.cp.Debug {
 		s := lmp.lr.GetRowString(0)
 		lmp.lr.ResetKeepSettings()
@@ -238,7 +239,7 @@ 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(protocolName string, isStreamMode bool) 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))
@@ -251,7 +252,10 @@ func (cp *CommonParams) NewLogMessageProcessor(protocolName string) LogMessagePr
 
 		stopCh: make(chan struct{}),
 	}
-	lmp.initPeriodicFlush()
+
+	if isStreamMode {
+		lmp.initPeriodicFlush()
+	}
 
 	return lmp
 }

app/vlinsert/insertutil/flags.go

@@ -1,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"flag"

app/vlinsert/insertutil/line_reader.go

@@ -1,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"bytes"

app/vlinsert/insertutil/line_reader_test.go

@@ -1,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"bytes"

app/vlinsert/insertutil/testutils.go

@@ -1,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"fmt"

app/vlinsert/insertutil/timestamp.go

@@ -1,8 +1,10 @@
-package insertutils
+package insertutil
 
 import (
 	"fmt"
+	"math"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
@@ -49,6 +51,35 @@ func parseTimestamp(s string) (int64, error) {
 
 // ParseUnixTimestamp parses s as unix timestamp in seconds, milliseconds, microseconds or nanoseconds and returns the parsed timestamp in nanoseconds.
 func ParseUnixTimestamp(s string) (int64, error) {
+	if strings.IndexByte(s, '.') >= 0 {
+		// Parse timestamp as floating-point value
+		f, err := strconv.ParseFloat(s, 64)
+		if err != nil {
+			return 0, fmt.Errorf("cannot parse unix timestamp from %q: %w", s, err)
+		}
+		if f < (1<<31) && f >= (-1<<31) {
+			// The timestamp is in seconds.
+			return int64(f * 1e9), nil
+		}
+		if f < 1e3*(1<<31) && f >= 1e3*(-1<<31) {
+			// The timestamp is in milliseconds.
+			return int64(f * 1e6), nil
+		}
+		if f < 1e6*(1<<31) && f >= 1e6*(-1<<31) {
+			// The timestamp is in microseconds.
+			return int64(f * 1e3), nil
+		}
+		// The timestamp is in nanoseconds
+		if f > math.MaxInt64 {
+			return 0, fmt.Errorf("too big timestamp in nanoseconds: %v; mustn't exceed %v", f, int64(math.MaxInt64))
+		}
+		if f < math.MinInt64 {
+			return 0, fmt.Errorf("too small timestamp in nanoseconds: %v; must be bigger or equal to %v", f, int64(math.MinInt64))
+		}
+		return int64(f), nil
+	}
+
+	// Parse timestamp as integer
 	n, err := strconv.ParseInt(s, 10, 64)
 	if err != nil {
 		return 0, fmt.Errorf("cannot parse unix timestamp from %q: %w", s, err)

app/vlinsert/insertutil/timestamp_test.go

@@ -1,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"testing"
@@ -6,6 +6,63 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
+func TestParseUnixTimestamp_Success(t *testing.T) {
+	f := func(s string, timestampExpected int64) {
+		t.Helper()
+
+		timestamp, err := ParseUnixTimestamp(s)
+		if err != nil {
+			t.Fatalf("unexpected error in ParseUnixTimestamp(%q): %s", s, err)
+		}
+		if timestamp != timestampExpected {
+			t.Fatalf("unexpected timestamp returned from ParseUnixTimestamp(%q); got %d; want %d", s, timestamp, timestampExpected)
+		}
+	}
+
+	f("0", 0)
+
+	// nanoseconds
+	f("-1234567890123456789", -1234567890123456789)
+	f("1234567890123456789", 1234567890123456789)
+
+	// microseconds
+	f("-1234567890123456", -1234567890123456000)
+	f("1234567890123456", 1234567890123456000)
+	f("1234567890123456.789", 1234567890123456768)
+
+	// milliseconds
+	f("-1234567890123", -1234567890123000000)
+	f("1234567890123", 1234567890123000000)
+	f("1234567890123.456", 1234567890123456000)
+
+	// seconds
+	f("-1234567890", -1234567890000000000)
+	f("1234567890", 1234567890000000000)
+	f("-1234567890.123456", -1234567890123456000)
+}
+
+func TestParseUnixTimestamp_Failure(t *testing.T) {
+	f := func(s string) {
+		t.Helper()
+
+		_, err := ParseUnixTimestamp(s)
+		if err == nil {
+			t.Fatalf("expecting non-nil error in ParseUnixTimestamp(%q)", s)
+		}
+	}
+
+	// non-numeric timestamp
+	f("")
+	f("foobar")
+	f("foo.bar")
+
+	// too big timestamp
+	f("12345678901234567890")
+	f("-12345678901234567890")
+	f("12345678901234567890.235424")
+	f("-12345678901234567890.235424")
+}
+
 func TestExtractTimestampFromFields_Success(t *testing.T) {
 	f := func(timeField string, fields []logstorage.Field, nsecsExpected int64) {
 		t.Helper()

app/vlinsert/journald/journald.go

@@ -5,7 +5,6 @@ import (
 	"encoding/binary"
 	"flag"
 	"fmt"
-	"io"
 	"net/http"
 	"regexp"
 	"slices"
@@ -13,39 +12,37 @@ import (
 	"strings"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
 )
 
-const (
-	journaldEntryMaxNameLen = 64
-)
+// See https://github.com/systemd/systemd/blob/main/src/libsystemd/sd-journal/journal-file.c#L1703
+const journaldEntryMaxNameLen = 64
+
+var allowedJournaldEntryNameChars = regexp.MustCompile(`^[A-Z_][A-Z0-9_]*`)
 
 var (
-	bodyBufferPool                bytesutil.ByteBufferPool
-	allowedJournaldEntryNameChars = regexp.MustCompile(`^[A-Z_][A-Z0-9_]*`)
-)
-
-var (
-	journaldStreamFields = flagutil.NewArrayString("journald.streamFields", "Journal fields to be used as stream fields. "+
-		"See the list of allowed fields at https://www.freedesktop.org/software/systemd/man/latest/systemd.journal-fields.html.")
-	journaldIgnoreFields = flagutil.NewArrayString("journald.ignoreFields", "Journal fields to ignore. "+
-		"See the list of allowed fields at https://www.freedesktop.org/software/systemd/man/latest/systemd.journal-fields.html.")
-	journaldTimeField = flag.String("journald.timeField", "__REALTIME_TIMESTAMP", "Journal field to be used as time field. "+
-		"See the list of allowed fields at https://www.freedesktop.org/software/systemd/man/latest/systemd.journal-fields.html.")
-	journaldTenantID             = flag.String("journald.tenantID", "0:0", "TenantID for logs ingested via the Journald endpoint.")
+	journaldStreamFields = flagutil.NewArrayString("journald.streamFields", "Comma-separated list of fields to use as log stream fields for logs ingested over journald protocol. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#stream-fields")
+	journaldIgnoreFields = flagutil.NewArrayString("journald.ignoreFields", "Comma-separated list of fields to ignore for logs ingested over journald protocol. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#dropping-fields")
+	journaldTimeField = flag.String("journald.timeField", "__REALTIME_TIMESTAMP", "Field to use as a log timestamp for logs ingested via journald protocol. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#time-field")
+	journaldTenantID = flag.String("journald.tenantID", "0:0", "TenantID for logs ingested via the Journald endpoint. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#multitenancy")
 	journaldIncludeEntryMetadata = flag.Bool("journald.includeEntryMetadata", false, "Include journal entry fields, which with double underscores.")
+
+	maxRequestSize = flagutil.NewBytes("journald.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single journald request")
 )
 
-func getCommonParams(r *http.Request) (*insertutils.CommonParams, error) {
-	cp, err := insertutils.GetCommonParams(r)
+func getCommonParams(r *http.Request) (*insertutil.CommonParams, error) {
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		return nil, err
 	}
@@ -89,43 +86,29 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsJournaldTotal.Inc()
 
-	if err := vlstorage.CanWriteData(); err != nil {
-		httpserver.Errorf(w, r, "%s", err)
-		return
-	}
-
-	reader := r.Body
-	var err error
-
-	wcr := writeconcurrencylimiter.GetReader(reader)
-	data, err := io.ReadAll(wcr)
-	if err != nil {
-		httpserver.Errorf(w, r, "cannot read request body: %s", err)
-		return
-	}
-	writeconcurrencylimiter.PutReader(wcr)
-	bb := bodyBufferPool.Get()
-	defer bodyBufferPool.Put(bb)
-	if r.Header.Get("Content-Encoding") == "zstd" {
-		bb.B, err = zstd.Decompress(bb.B[:0], data)
-		if err != nil {
-			httpserver.Errorf(w, r, "cannot decompress zstd-encoded request with length %d: %s", len(data), err)
-			return
-		}
-		data = bb.B
-	}
 	cp, err := getCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return
 	}
 
-	lmp := cp.NewLogMessageProcessor("journald")
-	err = parseJournaldRequest(data, lmp, cp)
-	lmp.MustClose()
+	if err := vlstorage.CanWriteData(); err != nil {
+		errorsTotal.Inc()
+		httpserver.Errorf(w, r, "%s", err)
+		return
+	}
+
+	encoding := r.Header.Get("Content-Encoding")
+	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
+		lmp := cp.NewLogMessageProcessor("journald", false)
+		err := parseJournaldRequest(data, lmp, cp)
+		lmp.MustClose()
+		return err
+	})
 	if err != nil {
 		errorsTotal.Inc()
-		httpserver.Errorf(w, r, "cannot parse Journald protobuf request: %s", err)
+		httpserver.Errorf(w, r, "cannot read journald protocol data: %s", err)
 		return
 	}
 
@@ -148,7 +131,7 @@ var (
 )
 
 // 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 insertutil.LogMessageProcessor, cp *insertutil.CommonParams) error {
 	var fields []logstorage.Field
 	var ts int64
 	var size uint64
@@ -198,7 +181,7 @@ func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *
 			if err != nil {
 				return fmt.Errorf("failed to extract binary field %q value size: %w", name, err)
 			}
-			// skip binary data sise
+			// skip binary data size
 			data = data[idx:]
 			if size == 0 {
 				return fmt.Errorf("unexpected zero binary data size decoded %d", size)
@@ -218,7 +201,6 @@ func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *
 			}
 			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)
 		}

app/vlinsert/journald/journald_test.go

@@ -3,14 +3,14 @@ package journald
 import (
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func TestPushJournaldOk(t *testing.T) {
 	f := func(src string, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
-		tlp := &insertutils.TestLogMessageProcessor{}
-		cp := &insertutils.CommonParams{
+		tlp := &insertutil.TestLogMessageProcessor{}
+		cp := &insertutil.CommonParams{
 			TimeField: "__REALTIME_TIMESTAMP",
 			MsgFields: []string{"MESSAGE"},
 		}
@@ -44,8 +44,8 @@ func TestPushJournaldOk(t *testing.T) {
 func TestPushJournald_Failure(t *testing.T) {
 	f := func(data string) {
 		t.Helper()
-		tlp := &insertutils.TestLogMessageProcessor{}
-		cp := &insertutils.CommonParams{
+		tlp := &insertutil.TestLogMessageProcessor{}
+		cp := &insertutil.CommonParams{
 			TimeField: "__REALTIME_TIMESTAMP",
 			MsgFields: []string{"MESSAGE"},
 		}

app/vlinsert/jsonline/jsonline.go

@@ -6,12 +6,12 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -28,7 +28,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 
 	requestsTotal.Inc()
 
-	cp, err := insertutils.GetCommonParams(r)
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -38,18 +38,15 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	reader := r.Body
-	if r.Header.Get("Content-Encoding") == "gzip" {
-		zr, err := common.GetGzipReader(reader)
-		if err != nil {
-			logger.Errorf("cannot read gzipped jsonline request: %s", err)
-			return
-		}
-		defer common.PutGzipReader(zr)
-		reader = zr
+	encoding := r.Header.Get("Content-Encoding")
+	reader, err := protoparserutil.GetUncompressedReader(r.Body, encoding)
+	if err != nil {
+		logger.Errorf("cannot decode jsonline request: %s", err)
+		return
 	}
+	defer protoparserutil.PutUncompressedReader(reader)
 
-	lmp := cp.NewLogMessageProcessor("jsonline")
+	lmp := cp.NewLogMessageProcessor("jsonline", true)
 	streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
 	processStreamInternal(streamName, reader, cp.TimeField, cp.MsgFields, lmp)
 	lmp.MustClose()
@@ -57,11 +54,11 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 	requestDuration.UpdateDuration(startTime)
 }
 
-func processStreamInternal(streamName string, r io.Reader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) {
+func processStreamInternal(streamName string, r io.Reader, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) {
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
-	lr := insertutils.NewLineReader(streamName, wcr)
+	lr := insertutil.NewLineReader(streamName, wcr)
 
 	n := 0
 	for {
@@ -78,7 +75,7 @@ func processStreamInternal(streamName string, r io.Reader, timeField string, msg
 	}
 }
 
-func readLine(lr *insertutils.LineReader, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (bool, error) {
+func readLine(lr *insertutil.LineReader, timeField string, msgFields []string, lmp insertutil.LogMessageProcessor) (bool, error) {
 	var line []byte
 	for len(line) == 0 {
 		if !lr.NextLine() {
@@ -94,7 +91,7 @@ func readLine(lr *insertutils.LineReader, timeField string, msgFields []string,
 	if err := p.ParseLogMessage(line); err != nil {
 		return true, fmt.Errorf("cannot parse json-encoded line: %w; line contents: %q", err, line)
 	}
-	ts, err := insertutils.ExtractTimestampFromFields(timeField, p.Fields)
+	ts, err := insertutil.ExtractTimestampFromFields(timeField, p.Fields)
 	if err != nil {
 		return true, fmt.Errorf("cannot get timestamp from json-encoded line: %w; line contents: %q", err, line)
 	}

app/vlinsert/jsonline/jsonline_test.go

@@ -4,7 +4,7 @@ import (
 	"bytes"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func TestProcessStreamInternal(t *testing.T) {
@@ -12,7 +12,7 @@ func TestProcessStreamInternal(t *testing.T) {
 		t.Helper()
 
 		msgFields := []string{msgField}
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
 		processStreamInternal("test", r, timeField, msgFields, tlp)
 

app/vlinsert/loki/loki.go

@@ -1,12 +1,18 @@
 package loki
 
 import (
+	"flag"
+	"fmt"
 	"net/http"
+	"strconv"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
+var disableMessageParsing = flag.Bool("loki.disableMessageParsing", false, "Whether to disable automatic parsing of JSON-encoded log fields inside Loki log message into distinct log fields")
+
 // RequestHandler processes Loki insert requests
 func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 	switch path {
@@ -35,8 +41,17 @@ func handleInsert(r *http.Request, w http.ResponseWriter) {
 	}
 }
 
-func getCommonParams(r *http.Request) (*insertutils.CommonParams, error) {
-	cp, err := insertutils.GetCommonParams(r)
+type commonParams struct {
+	cp *insertutil.CommonParams
+
+	// Whether to parse JSON inside plaintext log message.
+	//
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8486
+	parseMessage bool
+}
+
+func getCommonParams(r *http.Request) (*commonParams, error) {
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		return nil, err
 	}
@@ -55,5 +70,17 @@ func getCommonParams(r *http.Request) (*insertutils.CommonParams, error) {
 
 	}
 
-	return cp, nil
+	parseMessage := !*disableMessageParsing
+	if rv := httputil.GetRequestValue(r, "disable_message_parsing", "VL-Loki-Disable-Message-Parsing"); rv != "" {
+		bv, err := strconv.ParseBool(rv)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse dusable_message_parsing=%q: %s", rv, err)
+		}
+		parseMessage = !bv
+	}
+
+	return &commonParams{
+		cp:           cp,
+		parseMessage: parseMessage,
+	}, nil
 }

app/vlinsert/loki/loki_json.go

@@ -2,47 +2,28 @@ package loki
 
 import (
 	"fmt"
-	"io"
-	"math"
 	"net/http"
-	"strconv"
 	"time"
 
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/valyala/fastjson"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"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/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 )
 
+var maxRequestSize = flagutil.NewBytes("loki.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single Loki request")
+
 var parserPool fastjson.ParserPool
 
 func handleJSON(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsJSONTotal.Inc()
-	reader := r.Body
-	if r.Header.Get("Content-Encoding") == "gzip" {
-		zr, err := common.GetGzipReader(reader)
-		if err != nil {
-			httpserver.Errorf(w, r, "cannot initialize gzip reader: %s", err)
-			return
-		}
-		defer common.PutGzipReader(zr)
-		reader = zr
-	}
-
-	wcr := writeconcurrencylimiter.GetReader(reader)
-	data, err := io.ReadAll(wcr)
-	writeconcurrencylimiter.PutReader(wcr)
-	if err != nil {
-		httpserver.Errorf(w, r, "cannot read request body: %s", err)
-		return
-	}
 
 	cp, err := getCommonParams(r)
 	if err != nil {
@@ -53,12 +34,17 @@ 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.MustClose()
+
+	encoding := r.Header.Get("Content-Encoding")
+	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
+		lmp := cp.cp.NewLogMessageProcessor("loki_json", false)
+		useDefaultStreamFields := len(cp.cp.StreamFields) == 0
+		err := parseJSONRequest(data, lmp, cp.cp.MsgFields, useDefaultStreamFields, cp.parseMessage)
+		lmp.MustClose()
+		return err
+	})
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot parse Loki json request: %s; data=%s", err, data)
+		httpserver.Errorf(w, r, "cannot read Loki json data: %s", err)
 		return
 	}
 
@@ -66,6 +52,9 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 	// There is no need in updating requestJSONDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
 	requestJSONDuration.UpdateDuration(startTime)
+
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8505
+	w.WriteHeader(http.StatusNoContent)
 }
 
 var (
@@ -73,9 +62,10 @@ var (
 	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 insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) 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)
@@ -90,11 +80,20 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 		return fmt.Errorf("`streams` item in the parsed JSON must contain an array; got %q", streamsV)
 	}
 
+	fields := getFields()
+	defer putFields(fields)
+
+	var msgParser *logstorage.JSONParser
+	if parseMessage {
+		msgParser = logstorage.GetJSONParser()
+		defer logstorage.PutJSONParser(msgParser)
+	}
+
 	currentTimestamp := time.Now().UnixNano()
-	var commonFields []logstorage.Field
+
 	for _, stream := range streams {
 		// populate common labels from `stream` dict
-		commonFields = commonFields[:0]
+		fields.fields = fields.fields[:0]
 		labelsV := stream.Get("stream")
 		var labels *fastjson.Object
 		if labelsV != nil {
@@ -110,7 +109,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 				err = fmt.Errorf("unexpected label value type for %q:%q; want string", k, v)
 				return
 			}
-			commonFields = append(commonFields, logstorage.Field{
+			fields.fields = append(fields.fields, logstorage.Field{
 				Name:  bytesutil.ToUnsafeString(k),
 				Value: bytesutil.ToUnsafeString(vStr),
 			})
@@ -129,8 +128,10 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 			return fmt.Errorf("`values` item in the parsed JSON must contain an array; got %q", linesV)
 		}
 
-		fields := commonFields
+		commonFieldsLen := len(fields.fields)
 		for _, line := range lines {
+			fields.fields = fields.fields[:commonFieldsLen]
+
 			lineA, err := line.Array()
 			if err != nil {
 				return fmt.Errorf("unexpected contents of `values` item; want array; got %q", line)
@@ -152,17 +153,6 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 				ts = currentTimestamp
 			}
 
-			// parse log message
-			msg, err := lineA[1].StringBytes()
-			if err != nil {
-				return fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
-			}
-
-			fields = append(fields[:len(commonFields)], logstorage.Field{
-				Name:  "_msg",
-				Value: bytesutil.ToUnsafeString(msg),
-			})
-
 			// parse structured metadata - see https://grafana.com/docs/loki/latest/reference/loki-http-api/#ingest-logs
 			if len(lineA) > 2 {
 				structuredMetadata, err := lineA[2].Object()
@@ -177,7 +167,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 						return
 					}
 
-					fields = append(fields, logstorage.Field{
+					fields.fields = append(fields.fields, logstorage.Field{
 						Name:  bytesutil.ToUnsafeString(k),
 						Value: bytesutil.ToUnsafeString(vStr),
 					})
@@ -186,39 +176,51 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, useDefau
 					return fmt.Errorf("error when parsing `structuredMetadata` object: %w", err)
 				}
 			}
+
+			// parse log message
+			msg, err := lineA[1].StringBytes()
+			if err != nil {
+				return fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
+			}
+			allowMsgRenaming := false
+			fields.fields, allowMsgRenaming = addMsgField(fields.fields, msgParser, bytesutil.ToUnsafeString(msg))
+
 			var streamFields []logstorage.Field
 			if useDefaultStreamFields {
-				streamFields = commonFields
+				streamFields = fields.fields[:commonFieldsLen]
 			}
-			lmp.AddRow(ts, fields, streamFields)
+			if allowMsgRenaming {
+				logstorage.RenameField(fields.fields[commonFieldsLen:], msgFields, "_msg")
+			}
+			lmp.AddRow(ts, fields.fields, streamFields)
 		}
 	}
 
 	return nil
 }
 
+func addMsgField(dst []logstorage.Field, msgParser *logstorage.JSONParser, msg string) ([]logstorage.Field, bool) {
+	if msgParser == nil || len(msg) < 2 || msg[0] != '{' || msg[len(msg)-1] != '}' {
+		return append(dst, logstorage.Field{
+			Name:  "_msg",
+			Value: msg,
+		}), false
+	}
+	if msgParser != nil && len(msg) >= 2 && msg[0] == '{' && msg[len(msg)-1] == '}' {
+		if err := msgParser.ParseLogMessage(bytesutil.ToUnsafeBytes(msg)); err == nil {
+			return append(dst, msgParser.Fields...), true
+		}
+	}
+	return append(dst, logstorage.Field{
+		Name:  "_msg",
+		Value: msg,
+	}), false
+}
+
 func parseLokiTimestamp(s string) (int64, error) {
 	if s == "" {
 		// Special case - an empty timestamp must be substituted with the current time by the caller.
 		return 0, nil
 	}
-	n, err := strconv.ParseInt(s, 10, 64)
-	if err != nil {
-		// Fall back to parsing floating-point value
-		f, err := strconv.ParseFloat(s, 64)
-		if err != nil {
-			return 0, err
-		}
-		if f > math.MaxInt64 {
-			return 0, fmt.Errorf("too big timestamp in nanoseconds: %v; mustn't exceed %v", f, int64(math.MaxInt64))
-		}
-		if f < math.MinInt64 {
-			return 0, fmt.Errorf("too small timestamp in nanoseconds: %v; must be bigger or equal to %v", f, int64(math.MinInt64))
-		}
-		n = int64(f)
-	}
-	if n < 0 {
-		return 0, fmt.Errorf("too small timestamp in nanoseconds: %d; must be bigger than 0", n)
-	}
-	return n, nil
+	return insertutil.ParseUnixTimestamp(s)
 }

app/vlinsert/loki/loki_json_test.go

@@ -3,15 +3,15 @@ package loki
 import (
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func TestParseJSONRequest_Failure(t *testing.T) {
 	f := func(s string) {
 		t.Helper()
 
-		tlp := &insertutils.TestLogMessageProcessor{}
-		if err := parseJSONRequest([]byte(s), tlp, false); err == nil {
+		tlp := &insertutil.TestLogMessageProcessor{}
+		if err := parseJSONRequest([]byte(s), tlp, nil, false, false); err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
 		if err := tlp.Verify(nil, ""); err != nil {
@@ -63,9 +63,9 @@ func TestParseJSONRequest_Success(t *testing.T) {
 	f := func(s string, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 
-		if err := parseJSONRequest([]byte(s), tlp, false); err != nil {
+		if err := parseJSONRequest([]byte(s), tlp, nil, false, false); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
@@ -89,9 +89,9 @@ func TestParseJSONRequest_Success(t *testing.T) {
 	"label2": "value2"
 },"values":[
 	["1577836800000000001", "foo bar"],
-	["1477836900005000002", "abc"],
+	["1686026123.62", "abc"],
 	["147.78369e9", "foobar"]
-]}]}`, []int64{1577836800000000001, 1477836900005000002, 147783690000}, `{"label1":"value1","label2":"value2","_msg":"foo bar"}
+]}]}`, []int64{1577836800000000001, 1686026123620000000, 147783690000000000}, `{"label1":"value1","label2":"value2","_msg":"foo bar"}
 {"label1":"value1","label2":"value2","_msg":"abc"}
 {"label1":"value1","label2":"value2","_msg":"foobar"}`)
 
@@ -122,6 +122,48 @@ func TestParseJSONRequest_Success(t *testing.T) {
 {"x":"y","_msg":"yx"}`)
 
 	// values with metadata
-	f(`{"streams":[{"values":[["1577836800000000001", "foo bar", {"metadata_1": "md_value"}]]}]}`, []int64{1577836800000000001}, `{"_msg":"foo bar","metadata_1":"md_value"}`)
+	f(`{"streams":[{"values":[["1577836800000000001", "foo bar", {"metadata_1": "md_value"}]]}]}`, []int64{1577836800000000001}, `{"metadata_1":"md_value","_msg":"foo bar"}`)
 	f(`{"streams":[{"values":[["1577836800000000001", "foo bar", {}]]}]}`, []int64{1577836800000000001}, `{"_msg":"foo bar"}`)
 }
+
+func TestParseJSONRequest_ParseMessage(t *testing.T) {
+	f := func(s string, msgFields []string, timestampsExpected []int64, resultExpected string) {
+		t.Helper()
+
+		tlp := &insertutil.TestLogMessageProcessor{}
+
+		if err := parseJSONRequest([]byte(s), tlp, msgFields, false, true); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	f(`{
+	"streams": [
+		{
+			"stream": {
+				"foo": "bar",
+				"a": "b"
+			},
+			"values": [
+				["1577836800000000001", "{\"user_id\":\"123\"}"],
+				["1577836900005000002", "abc", {"trace_id":"pqw"}],
+				["1577836900005000003", "{def}"]
+			]
+		},
+		{
+			"stream": {
+				"x": "y"
+			},
+			"values": [
+				["1877836900005000004", "{\"trace_id\":\"111\",\"parent_id\":\"abc\"}"]
+			]
+		}
+	]
+}`, []string{"a", "trace_id"}, []int64{1577836800000000001, 1577836900005000002, 1577836900005000003, 1877836900005000004}, `{"foo":"bar","a":"b","user_id":"123"}
+{"foo":"bar","a":"b","trace_id":"pqw","_msg":"abc"}
+{"foo":"bar","a":"b","_msg":"{def}"}
+{"x":"y","_msg":"111","parent_id":"abc"}`)
+}

app/vlinsert/loki/loki_json_timing_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func BenchmarkParseJSONRequest(b *testing.B) {
@@ -22,13 +22,13 @@ func BenchmarkParseJSONRequest(b *testing.B) {
 }
 
 func benchmarkParseJSONRequest(b *testing.B, streams, rows, labels int) {
-	blp := &insertutils.BenchmarkLogMessageProcessor{}
+	blp := &insertutil.BenchmarkLogMessageProcessor{}
 	b.ReportAllocs()
 	b.SetBytes(int64(streams * rows))
 	b.RunParallel(func(pb *testing.PB) {
 		data := getJSONBody(streams, rows, labels)
 		for pb.Next() {
-			if err := parseJSONRequest(data, blp, false); err != nil {
+			if err := parseJSONRequest(data, blp, nil, false, true); err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}
 		}

app/vlinsert/loki/loki_protobuf.go

@@ -2,38 +2,27 @@ package loki
 
 import (
 	"fmt"
-	"io"
 	"net/http"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
-	"github.com/golang/snappy"
 )
 
 var (
-	bytesBufPool bytesutil.ByteBufferPool
 	pushReqsPool sync.Pool
 )
 
 func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsProtobufTotal.Inc()
-	wcr := writeconcurrencylimiter.GetReader(r.Body)
-	data, err := io.ReadAll(wcr)
-	writeconcurrencylimiter.PutReader(wcr)
-	if err != nil {
-		httpserver.Errorf(w, r, "cannot read request body: %s", err)
-		return
-	}
 
 	cp, err := getCommonParams(r)
 	if err != nil {
@@ -44,12 +33,22 @@ 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.MustClose()
+
+	encoding := r.Header.Get("Content-Encoding")
+	if encoding == "" {
+		// Loki protocol uses snappy compression by default.
+		// See https://grafana.com/docs/loki/latest/reference/loki-http-api/#ingest-logs
+		encoding = "snappy"
+	}
+	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
+		lmp := cp.cp.NewLogMessageProcessor("loki_protobuf", false)
+		useDefaultStreamFields := len(cp.cp.StreamFields) == 0
+		err := parseProtobufRequest(data, lmp, cp.cp.MsgFields, useDefaultStreamFields, cp.parseMessage)
+		lmp.MustClose()
+		return err
+	})
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot parse Loki protobuf request: %s", err)
+		httpserver.Errorf(w, r, "cannot read Loki protobuf data: %s", err)
 		return
 	}
 
@@ -57,6 +56,9 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 	// There is no need in updating requestProtobufDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
 	requestProtobufDuration.UpdateDuration(startTime)
+
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8505
+	w.WriteHeader(http.StatusNoContent)
 }
 
 var (
@@ -64,20 +66,11 @@ var (
 	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 {
-	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)
-	}
-	bb.B = buf
-
+func parseProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {
 	req := getPushRequest()
 	defer putPushRequest(req)
 
-	err = req.UnmarshalProtobuf(bb.B)
+	err := req.UnmarshalProtobuf(data)
 	if err != nil {
 		return fmt.Errorf("cannot parse request body: %w", err)
 	}
@@ -85,8 +78,15 @@ func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useD
 	fields := getFields()
 	defer putFields(fields)
 
+	var msgParser *logstorage.JSONParser
+	if parseMessage {
+		msgParser = logstorage.GetJSONParser()
+		defer logstorage.PutJSONParser(msgParser)
+	}
+
 	streams := req.Streams
 	currentTimestamp := time.Now().UnixNano()
+
 	for i := range streams {
 		stream := &streams[i]
 		// st.Labels contains labels for the stream.
@@ -109,10 +109,8 @@ func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useD
 				})
 			}
 
-			fields.fields = append(fields.fields, logstorage.Field{
-				Name:  "_msg",
-				Value: e.Line,
-			})
+			allowMsgRenaming := false
+			fields.fields, allowMsgRenaming = addMsgField(fields.fields, msgParser, e.Line)
 
 			ts := e.Timestamp.UnixNano()
 			if ts == 0 {
@@ -123,6 +121,9 @@ func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useD
 			if useDefaultStreamFields {
 				streamFields = fields.fields[:commonFieldsLen]
 			}
+			if allowMsgRenaming {
+				logstorage.RenameField(fields.fields[commonFieldsLen:], msgFields, "_msg")
+			}
 			lmp.AddRow(ts, fields.fields, streamFields)
 		}
 	}

app/vlinsert/loki/loki_protobuf_test.go

@@ -6,9 +6,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/golang/snappy"
 )
 
 type testLogMessageProcessor struct {
@@ -53,7 +52,7 @@ func TestParseProtobufRequest_Success(t *testing.T) {
 		t.Helper()
 
 		tlp := &testLogMessageProcessor{}
-		if err := parseJSONRequest([]byte(s), tlp, false); err != nil {
+		if err := parseJSONRequest([]byte(s), tlp, nil, false, false); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		if len(tlp.pr.Streams) != len(timestampsExpected) {
@@ -61,10 +60,9 @@ func TestParseProtobufRequest_Success(t *testing.T) {
 		}
 
 		data := tlp.pr.MarshalProtobuf(nil)
-		encodedData := snappy.Encode(nil, data)
 
-		tlp2 := &insertutils.TestLogMessageProcessor{}
-		if err := parseProtobufRequest(encodedData, tlp2, false); err != nil {
+		tlp2 := &insertutil.TestLogMessageProcessor{}
+		if err := parseProtobufRequest(data, tlp2, nil, false, false); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
 		if err := tlp2.Verify(timestampsExpected, resultExpected); err != nil {
@@ -90,7 +88,7 @@ func TestParseProtobufRequest_Success(t *testing.T) {
 	["1577836800000000001", "foo bar"],
 	["1477836900005000002", "abc"],
 	["147.78369e9", "foobar"]
-]}]}`, []int64{1577836800000000001, 1477836900005000002, 147783690000}, `{"label1":"value1","label2":"value2","_msg":"foo bar"}
+]}]}`, []int64{1577836800000000001, 1477836900005000002, 147783690000000000}, `{"label1":"value1","label2":"value2","_msg":"foo bar"}
 {"label1":"value1","label2":"value2","_msg":"abc"}
 {"label1":"value1","label2":"value2","_msg":"foobar"}`)
 
@@ -121,6 +119,57 @@ func TestParseProtobufRequest_Success(t *testing.T) {
 {"x":"y","_msg":"yx"}`)
 }
 
+func TestParseProtobufRequest_ParseMessage(t *testing.T) {
+	f := func(s string, msgFields []string, timestampsExpected []int64, resultExpected string) {
+		t.Helper()
+
+		tlp := &testLogMessageProcessor{}
+		if err := parseJSONRequest([]byte(s), tlp, nil, false, false); 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))
+		}
+
+		data := tlp.pr.MarshalProtobuf(nil)
+
+		tlp2 := &insertutil.TestLogMessageProcessor{}
+		if err := parseProtobufRequest(data, tlp2, msgFields, false, true); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		if err := tlp2.Verify(timestampsExpected, resultExpected); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	f(`{
+	"streams": [
+		{
+			"stream": {
+				"foo": "bar",
+				"a": "b"
+			},
+			"values": [
+				["1577836800000000001", "{\"user_id\":\"123\"}"],
+				["1577836900005000002", "abc", {"trace_id":"pqw"}],
+				["1577836900005000003", "{def}"]
+			]
+		},
+		{
+			"stream": {
+				"x": "y"
+			},
+			"values": [
+				["1877836900005000004", "{\"trace_id\":\"432\",\"parent_id\":\"qwerty\"}"]
+			]
+		}
+	]
+}`, []string{"a", "trace_id"}, []int64{1577836800000000001, 1577836900005000002, 1577836900005000003, 1877836900005000004}, `{"foo":"bar","a":"b","user_id":"123"}
+{"foo":"bar","a":"b","trace_id":"pqw","_msg":"abc"}
+{"foo":"bar","a":"b","_msg":"{def}"}
+{"x":"y","_msg":"432","parent_id":"qwerty"}`)
+}
+
 func TestParsePromLabels_Success(t *testing.T) {
 	f := func(s string) {
 		t.Helper()

app/vlinsert/loki/loki_protobuf_timing_test.go

@@ -6,9 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/golang/snappy"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 )
 
@@ -25,13 +23,13 @@ func BenchmarkParseProtobufRequest(b *testing.B) {
 }
 
 func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
-	blp := &insertutils.BenchmarkLogMessageProcessor{}
+	blp := &insertutil.BenchmarkLogMessageProcessor{}
 	b.ReportAllocs()
 	b.SetBytes(int64(streams * rows))
 	b.RunParallel(func(pb *testing.PB) {
 		body := getProtobufBody(streams, rows, labels)
 		for pb.Next() {
-			if err := parseProtobufRequest(body, blp, false); err != nil {
+			if err := parseProtobufRequest(body, blp, nil, false, true); err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}
 		}
@@ -78,8 +76,5 @@ func getProtobufBody(streamsCount, rowsCount, labelsCount int) []byte {
 		Streams: streams,
 	}
 
-	body := pr.MarshalProtobuf(nil)
-	encodedBody := snappy.Encode(nil, body)
-
-	return encodedBody
+	return pr.MarshalProtobuf(nil)
 }

app/vlinsert/opentelemetry/opentelemetry.go

@@ -2,21 +2,22 @@ package opentelemetry
 
 import (
 	"fmt"
-	"io"
 	"net/http"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/pb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
+var maxRequestSize = flagutil.NewBytes("opentelemetry.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single OpenTelemetry request")
+
 // RequestHandler processes Opentelemetry insert requests
 func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 	switch path {
@@ -37,26 +38,8 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsProtobufTotal.Inc()
-	reader := r.Body
-	if r.Header.Get("Content-Encoding") == "gzip" {
-		zr, err := common.GetGzipReader(reader)
-		if err != nil {
-			httpserver.Errorf(w, r, "cannot initialize gzip reader: %s", err)
-			return
-		}
-		defer common.PutGzipReader(zr)
-		reader = zr
-	}
 
-	wcr := writeconcurrencylimiter.GetReader(reader)
-	data, err := io.ReadAll(wcr)
-	writeconcurrencylimiter.PutReader(wcr)
-	if err != nil {
-		httpserver.Errorf(w, r, "cannot read request body: %s", err)
-		return
-	}
-
-	cp, err := insertutils.GetCommonParams(r)
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return
@@ -66,12 +49,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.MustClose()
+	encoding := r.Header.Get("Content-Encoding")
+	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
+		lmp := cp.NewLogMessageProcessor("opentelelemtry_protobuf", false)
+		useDefaultStreamFields := len(cp.StreamFields) == 0
+		err := pushProtobufRequest(data, lmp, useDefaultStreamFields)
+		lmp.MustClose()
+		return err
+	})
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot parse OpenTelemetry protobuf request: %s", err)
+		httpserver.Errorf(w, r, "cannot read OpenTelemetry protocol data: %s", err)
 		return
 	}
 
@@ -88,7 +75,7 @@ var (
 	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 insertutil.LogMessageProcessor, useDefaultStreamFields bool) error {
 	var req pb.ExportLogsServiceRequest
 	if err := req.UnmarshalProtobuf(data); err != nil {
 		errorsTotal.Inc()
@@ -101,7 +88,7 @@ func pushProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useDe
 		commonFields = slicesutil.SetLength(commonFields, len(attributes))
 		for i, attr := range attributes {
 			commonFields[i].Name = attr.Key
-			commonFields[i].Value = attr.Value.FormatString()
+			commonFields[i].Value = attr.Value.FormatString(true)
 		}
 		commonFieldsLen := len(commonFields)
 		for _, sc := range rl.ScopeLogs {
@@ -112,18 +99,18 @@ func pushProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, useDe
 	return 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 insertutil.LogMessageProcessor, useDefaultStreamFields bool) []logstorage.Field {
 	fields := commonFields
 	for _, lr := range sc.LogRecords {
 		fields = fields[:len(commonFields)]
 		fields = append(fields, logstorage.Field{
 			Name:  "_msg",
-			Value: lr.Body.FormatString(),
+			Value: lr.Body.FormatString(true),
 		})
 		for _, attr := range lr.Attributes {
 			fields = append(fields, logstorage.Field{
 				Name:  attr.Key,
-				Value: attr.Value.FormatString(),
+				Value: attr.Value.FormatString(true),
 			})
 		}
 		if len(lr.TraceID) > 0 {

app/vlinsert/opentelemetry/opentelemetry_test.go

@@ -3,7 +3,7 @@ package opentelemetry
 import (
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/pb"
 )
 
@@ -15,7 +15,7 @@ func TestPushProtoOk(t *testing.T) {
 		}
 
 		pData := lr.MarshalProtobuf(nil)
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 		if err := pushProtobufRequest(pData, tlp, false); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -24,6 +24,7 @@ func TestPushProtoOk(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
+
 	// single line without resource attributes
 	f([]pb.ResourceLogs{
 		{
@@ -39,6 +40,7 @@ func TestPushProtoOk(t *testing.T) {
 		[]int64{1234},
 		`{"_msg":"log-line-message","severity":"Trace"}`,
 	)
+
 	// multi-line with resource attributes
 	f([]pb.ResourceLogs{
 		{
@@ -66,9 +68,9 @@ func TestPushProtoOk(t *testing.T) {
 		},
 	},
 		[]int64{1234, 1235, 1236},
-		`{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message","severity":"Trace"}
-{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message-msg-2","severity":"Unspecified"}
-{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message-msg-2","severity":"Unspecified"}`,
+		`{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message","severity":"Trace"}
+{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Unspecified"}
+{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Unspecified"}`,
 	)
 
 	// multi-scope with resource attributes and multi-line
@@ -106,19 +108,21 @@ func TestPushProtoOk(t *testing.T) {
 				{
 					LogRecords: []pb.LogRecord{
 						{TimeUnixNano: 2347, SeverityNumber: 12, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-0")}},
-						{ObservedTimeUnixNano: 2348, SeverityNumber: 12, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-1")}},
+						{TraceID: "1234", SpanID: "45", ObservedTimeUnixNano: 2348, SeverityNumber: 12, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-1")}},
+						{TraceID: "4bf92f3577b34da6a3ce929d0e0e4736", SpanID: "00f067aa0ba902b7", ObservedTimeUnixNano: 3333, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-2")}},
 					},
 				},
 			},
 		},
 	},
-		[]int64{1234, 1235, 2345, 2346, 2347, 2348},
-		`{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message","severity":"Trace"}
-{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message-msg-2","severity":"Debug"}
+		[]int64{1234, 1235, 2345, 2346, 2347, 2348, 3333},
+		`{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message","severity":"Trace"}
+{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Debug"}
 {"_msg":"log-line-resource-scope-1-0-0","severity":"Info2"}
 {"_msg":"log-line-resource-scope-1-0-1","severity":"Info2"}
 {"_msg":"log-line-resource-scope-1-1-0","severity":"Info4"}
-{"_msg":"log-line-resource-scope-1-1-1","severity":"Info4"}`,
+{"_msg":"log-line-resource-scope-1-1-1","trace_id":"1234","span_id":"45","severity":"Info4"}
+{"_msg":"log-line-resource-scope-1-1-2","trace_id":"4bf92f3577b34da6a3ce929d0e0e4736","span_id":"00f067aa0ba902b7","severity":"Unspecified"}`,
 	)
 }
 

app/vlinsert/opentelemetry/opentelemetry_timing_test.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/pb"
 )
 
@@ -21,7 +21,7 @@ func BenchmarkParseProtobufRequest(b *testing.B) {
 }
 
 func benchmarkParseProtobufRequest(b *testing.B, streams, rows, labels int) {
-	blp := &insertutils.BenchmarkLogMessageProcessor{}
+	blp := &insertutil.BenchmarkLogMessageProcessor{}
 	b.ReportAllocs()
 	b.SetBytes(int64(streams * rows))
 	b.RunParallel(func(pb *testing.PB) {

app/vlinsert/syslog/syslog.go

@@ -16,9 +16,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/klauspost/compress/gzip"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
@@ -27,7 +25,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
@@ -274,14 +272,14 @@ func runTCPListener(addr string, argIdx int) {
 
 func checkCompressMethod(compressMethod, addr, protocol string) {
 	switch compressMethod {
-	case "", "none", "gzip", "deflate":
+	case "", "none", "zstd", "gzip", "deflate":
 		return
 	default:
-		logger.Fatalf("unsupported -syslog.compressMethod.%s=%q for -syslog.listenAddr.%s=%q; supported values: 'none', 'gzip', 'deflate'", protocol, compressMethod, protocol, addr)
+		logger.Fatalf("unsupported -syslog.compressMethod.%s=%q for -syslog.listenAddr.%s=%q; supported values: 'none', 'zstd', 'gzip', 'deflate'", protocol, compressMethod, protocol, addr)
 	}
 }
 
-func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
+func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
 	gomaxprocs := cgroup.AvailableCPUs()
 	var wg sync.WaitGroup
 	localAddr := ln.LocalAddr()
@@ -289,7 +287,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod st
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			cp := insertutils.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
+			cp := insertutil.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
 			var bb bytesutil.ByteBuffer
 			bb.B = bytesutil.ResizeNoCopyNoOverallocate(bb.B, 64*1024)
 			for {
@@ -314,7 +312,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("udp", bb.NewReader(), encoding, useLocalTimestamp, cp); err != nil {
 					logger.Errorf("syslog: cannot process UDP data from %s at %s: %s", remoteAddr, localAddr, err)
 				}
 			}
@@ -323,7 +321,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod st
 	wg.Wait()
 }
 
-func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
+func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
 	var cm ingestserver.ConnsMap
 	cm.Init("syslog")
 
@@ -353,8 +351,8 @@ 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 {
+			cp := insertutil.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
+			if err := processStream("tcp", c, encoding, useLocalTimestamp, cp); err != nil {
 				logger.Errorf("syslog: cannot process TCP data at %q: %s", addr, err)
 			}
 
@@ -369,52 +367,29 @@ 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(protocol string, r io.Reader, encoding string, useLocalTimestamp bool, cp *insertutil.CommonParams) error {
 	if err := vlstorage.CanWriteData(); err != nil {
 		return err
 	}
 
-	lmp := cp.NewLogMessageProcessor("syslog_" + protocol)
-	err := processStreamInternal(r, compressMethod, useLocalTimestamp, lmp)
+	lmp := cp.NewLogMessageProcessor("syslog_"+protocol, true)
+	err := processStreamInternal(r, encoding, useLocalTimestamp, lmp)
 	lmp.MustClose()
 
 	return err
 }
 
-func processStreamInternal(r io.Reader, compressMethod string, useLocalTimestamp bool, lmp insertutils.LogMessageProcessor) error {
-	switch compressMethod {
-	case "", "none":
-	case "gzip":
-		zr, err := common.GetGzipReader(r)
-		if err != nil {
-			return fmt.Errorf("cannot read gzipped data: %w", err)
-		}
-		r = zr
-	case "deflate":
-		zr, err := common.GetZlibReader(r)
-		if err != nil {
-			return fmt.Errorf("cannot read deflated data: %w", err)
-		}
-		r = zr
-	default:
-		logger.Panicf("BUG: unsupported compressMethod=%q; supported values: none, gzip, deflate", compressMethod)
+func processStreamInternal(r io.Reader, encoding string, useLocalTimestamp bool, lmp insertutil.LogMessageProcessor) error {
+	reader, err := protoparserutil.GetUncompressedReader(r, encoding)
+	if err != nil {
+		return fmt.Errorf("cannot decode syslog data: %w", err)
 	}
+	defer protoparserutil.PutUncompressedReader(reader)
 
-	err := processUncompressedStream(r, useLocalTimestamp, lmp)
-
-	switch compressMethod {
-	case "gzip":
-		zr := r.(*gzip.Reader)
-		common.PutGzipReader(zr)
-	case "deflate":
-		zr := r.(io.ReadCloser)
-		common.PutZlibReader(zr)
-	}
-
-	return err
+	return processUncompressedStream(reader, useLocalTimestamp, lmp)
 }
 
-func processUncompressedStream(r io.Reader, useLocalTimestamp bool, lmp insertutils.LogMessageProcessor) error {
+func processUncompressedStream(r io.Reader, useLocalTimestamp bool, lmp insertutil.LogMessageProcessor) error {
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
@@ -499,7 +474,7 @@ again:
 			slr.err = fmt.Errorf("cannot parse message length from %q: %w", msgLenStr, err)
 			return false
 		}
-		if maxMsgLen := insertutils.MaxLineSizeBytes.IntN(); msgLen > uint64(maxMsgLen) {
+		if maxMsgLen := insertutil.MaxLineSizeBytes.IntN(); msgLen > uint64(maxMsgLen) {
 			slr.err = fmt.Errorf("cannot read message longer than %d bytes; msgLen=%d", maxMsgLen, msgLen)
 			return false
 		}
@@ -551,7 +526,7 @@ func putSyslogLineReader(slr *syslogLineReader) {
 
 var syslogLineReaderPool sync.Pool
 
-func processLine(line []byte, currentYear int, timezone *time.Location, useLocalTimestamp bool, lmp insertutils.LogMessageProcessor) error {
+func processLine(line []byte, currentYear int, timezone *time.Location, useLocalTimestamp bool, lmp insertutil.LogMessageProcessor) error {
 	p := logstorage.GetSyslogParser(currentYear, timezone)
 	lineStr := bytesutil.ToUnsafeString(line)
 	p.Parse(lineStr)
@@ -560,7 +535,7 @@ func processLine(line []byte, currentYear int, timezone *time.Location, useLocal
 	if useLocalTimestamp {
 		ts = time.Now().UnixNano()
 	} else {
-		nsecs, err := insertutils.ExtractTimestampFromFields("timestamp", p.Fields)
+		nsecs, err := insertutil.ExtractTimestampFromFields("timestamp", p.Fields)
 		if err != nil {
 			return fmt.Errorf("cannot get timestamp from syslog line %q: %w", line, err)
 		}

app/vlinsert/syslog/syslog_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 )
 
 func TestSyslogLineReader_Success(t *testing.T) {
@@ -84,7 +84,7 @@ func TestProcessStreamInternal_Success(t *testing.T) {
 		globalTimezone = time.UTC
 		globalCurrentYear.Store(int64(currentYear))
 
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
 		if err := processStreamInternal(r, "", false, tlp); err != nil {
 			t.Fatalf("unexpected error: %s", err)
@@ -113,7 +113,7 @@ func TestProcessStreamInternal_Failure(t *testing.T) {
 		MustInit()
 		defer MustStop()
 
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
 		if err := processStreamInternal(r, "", false, tlp); err == nil {
 			t.Fatalf("expecting non-nil error")

app/vlogscli/json_prettifier.go

@@ -154,7 +154,7 @@ func readNextJSONObject(d *json.Decoder) ([]logstorage.Field, error) {
 		}
 		value, ok := t.(string)
 		if !ok {
-			return nil, fmt.Errorf("unexpected token read for oject value: %v; want string", t)
+			return nil, fmt.Errorf("unexpected token read for object value: %v; want string", t)
 		}
 
 		fields = append(fields, logstorage.Field{

app/vlogsgenerator/main.go

@@ -31,7 +31,7 @@ var (
 	totalStreams  = flag.Int("totalStreams", 0, "The number of total log streams; if -totalStreams > -activeStreams, then some active streams are substituted with new streams "+
 		"during data generation")
 	logsPerStream     = flag.Int64("logsPerStream", 1_000, "The number of log entries to generate per each log stream. Log entries are evenly distributed between -start and -end")
-	constFieldsPerLog = flag.Int("constFieldsPerLog", 3, "The number of fields with constaint values to generate per each log entry; "+
+	constFieldsPerLog = flag.Int("constFieldsPerLog", 3, "The number of fields with constant values to generate per each log entry; "+
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
 	varFieldsPerLog = flag.Int("varFieldsPerLog", 1, "The number of fields with variable values to generate per each log entry; "+
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
@@ -177,7 +177,10 @@ func generateAndPushLogs(cfg *workerConfig, workerID int) {
 	sw := &statWriter{
 		w: pw,
 	}
-	bw := bufio.NewWriter(sw)
+
+	// The 1MB write buffer increases data ingestion performance by reducing the number of send() syscalls
+	bw := bufio.NewWriterSize(sw, 1024*1024)
+
 	doneCh := make(chan struct{})
 	go func() {
 		generateLogs(bw, workerID, cfg.activeStreams, cfg.totalStreams)

app/vlselect/logsql/logsql.go

@@ -19,7 +19,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
@@ -35,22 +35,22 @@ func ProcessFacetsRequest(ctx context.Context, w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	limit, err := httputils.GetInt(r, "limit")
+	limit, err := httputil.GetInt(r, "limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
-	maxValuesPerField, err := httputils.GetInt(r, "max_values_per_field")
+	maxValuesPerField, err := httputil.GetInt(r, "max_values_per_field")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
-	maxValueLen, err := httputils.GetInt(r, "max_value_len")
+	maxValueLen, err := httputil.GetInt(r, "max_value_len")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
-	keepConstFields := httputils.GetBool(r, "keep_const_fields")
+	keepConstFields := httputil.GetBool(r, "keep_const_fields")
 
 	q.DropAllPipes()
 	q.AddFacetsPipe(limit, maxValuesPerField, maxValueLen, keepConstFields)
@@ -141,7 +141,7 @@ func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	fields := r.Form["field"]
 
 	// Obtain limit on the number of top fields entries.
-	fieldsLimit, err := httputils.GetInt(r, "fields_limit")
+	fieldsLimit, err := httputil.GetInt(r, "fields_limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -310,7 +310,7 @@ func ProcessFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *ht
 	}
 
 	// Parse limit query arg
-	limit, err := httputils.GetInt(r, "limit")
+	limit, err := httputil.GetInt(r, "limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -370,7 +370,7 @@ func ProcessStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter,
 	}
 
 	// Parse limit query arg
-	limit, err := httputils.GetInt(r, "limit")
+	limit, err := httputil.GetInt(r, "limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -401,7 +401,7 @@ func ProcessStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http
 	}
 
 	// Parse limit query arg
-	limit, err := httputils.GetInt(r, "limit")
+	limit, err := httputil.GetInt(r, "limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -432,7 +432,7 @@ func ProcessStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.R
 	}
 
 	// Parse limit query arg
-	limit, err := httputils.GetInt(r, "limit")
+	limit, err := httputil.GetInt(r, "limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -470,21 +470,21 @@ func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.
 		return
 	}
 
-	refreshIntervalMsecs, err := httputils.GetDuration(r, "refresh_interval", 1000)
+	refreshIntervalMsecs, err := httputil.GetDuration(r, "refresh_interval", 1000)
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	refreshInterval := time.Millisecond * time.Duration(refreshIntervalMsecs)
 
-	startOffsetMsecs, err := httputils.GetDuration(r, "start_offset", 5*1000)
+	startOffsetMsecs, err := httputil.GetDuration(r, "start_offset", 5*1000)
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	startOffset := startOffsetMsecs * 1e6
 
-	offsetMsecs, err := httputils.GetDuration(r, "offset", 1000)
+	offsetMsecs, err := httputil.GetDuration(r, "offset", 1000)
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
@@ -863,7 +863,7 @@ func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Req
 	}
 
 	// Parse limit query arg
-	limit, err := httputils.GetInt(r, "limit")
+	limit, err := httputil.GetInt(r, "limit")
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return

app/vlselect/main.go

@@ -12,7 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlselect/logsql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -71,6 +71,7 @@ var vmuiFileServer = http.FileServer(http.FS(vmuiFiles))
 // RequestHandler handles select requests for VictoriaLogs
 func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	path := r.URL.Path
+
 	if !strings.HasPrefix(path, "/select/") {
 		// Skip requests, which do not start with /select/, since these aren't our requests.
 		return false
@@ -98,25 +99,65 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 
+	ctx := r.Context()
+	if path == "/select/logsql/tail" {
+		logsqlTailRequests.Inc()
+		// Process live tailing request without timeout, since it is OK to run live tailing requests for very long time.
+		// Also do not apply concurrency limit to tail requests, since these limits are intended for non-tail requests.
+		logsql.ProcessLiveTailRequest(ctx, w, r)
+		return true
+	}
+
 	// Limit the number of concurrent queries, which can consume big amounts of CPU time.
 	startTime := time.Now()
-	ctx := r.Context()
 	d := getMaxQueryDuration(r)
 	ctxWithTimeout, cancel := context.WithTimeout(ctx, d)
 	defer cancel()
 
-	stopCh := ctxWithTimeout.Done()
+	if !incRequestConcurrency(ctxWithTimeout, w, r) {
+		return true
+	}
+	defer decRequestConcurrency()
+
+	ok := processSelectRequest(ctxWithTimeout, w, r, path)
+	if !ok {
+		return false
+	}
+
+	err := ctxWithTimeout.Err()
+	switch err {
+	case nil:
+		// nothing to do
+	case context.Canceled:
+		// do not log canceled requests, since they are expected and legal.
+	case context.DeadlineExceeded:
+		err = &httpserver.ErrorWithStatusCode{
+			Err: fmt.Errorf("the request couldn't be executed in %.3f seconds; possible solutions: "+
+				"to increase -search.maxQueryDuration=%s; to pass bigger value to 'timeout' query arg", time.Since(startTime).Seconds(), maxQueryDuration),
+			StatusCode: http.StatusServiceUnavailable,
+		}
+		httpserver.Errorf(w, r, "%s", err)
+	default:
+		httpserver.Errorf(w, r, "unexpected error: %s", err)
+	}
+
+	return true
+}
+
+func incRequestConcurrency(ctx context.Context, w http.ResponseWriter, r *http.Request) bool {
+	startTime := time.Now()
+	stopCh := ctx.Done()
 	select {
 	case concurrencyLimitCh <- struct{}{}:
-		defer func() { <-concurrencyLimitCh }()
+		return true
 	default:
 		// Sleep for a while until giving up. This should resolve short bursts in requests.
 		concurrencyLimitReached.Inc()
 		select {
 		case concurrencyLimitCh <- struct{}{}:
-			defer func() { <-concurrencyLimitCh }()
+			return true
 		case <-stopCh:
-			switch ctxWithTimeout.Err() {
+			switch ctx.Err() {
 			case context.Canceled:
 				remoteAddr := httpserver.GetQuotedRemoteAddr(r)
 				requestURI := httpserver.GetRequestURI(r)
@@ -129,49 +170,18 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 						"are executed. Possible solutions: to reduce query load; to add more compute resources to the server; "+
 						"to increase -search.maxQueueDuration=%s; to increase -search.maxQueryDuration=%s; to increase -search.maxConcurrentRequests; "+
 						"to pass bigger value to 'timeout' query arg",
-						d.Seconds(), *maxConcurrentRequests, maxQueueDuration, maxQueryDuration),
+						time.Since(startTime).Seconds(), *maxConcurrentRequests, maxQueueDuration, maxQueryDuration),
 					StatusCode: http.StatusServiceUnavailable,
 				}
 				httpserver.Errorf(w, r, "%s", err)
 			}
-			return true
+			return false
 		}
 	}
+}
 
-	if path == "/select/logsql/tail" {
-		logsqlTailRequests.Inc()
-		// Process live tailing request without timeout (e.g. use ctx instead of ctxWithTimeout),
-		// since it is OK to run live tailing requests for very long time.
-		logsql.ProcessLiveTailRequest(ctx, w, r)
-		return true
-	}
-
-	ok := processSelectRequest(ctxWithTimeout, w, r, path)
-	if !ok {
-		return false
-	}
-
-	err := ctxWithTimeout.Err()
-	switch err {
-	case nil:
-		// nothing to do
-	case context.Canceled:
-		remoteAddr := httpserver.GetQuotedRemoteAddr(r)
-		requestURI := httpserver.GetRequestURI(r)
-		logger.Infof("client has canceled the request after %.3f seconds: remoteAddr=%s, requestURI: %q",
-			time.Since(startTime).Seconds(), remoteAddr, requestURI)
-	case context.DeadlineExceeded:
-		err = &httpserver.ErrorWithStatusCode{
-			Err: fmt.Errorf("the request couldn't be executed in %.3f seconds; possible solutions: "+
-				"to increase -search.maxQueryDuration=%s; to pass bigger value to 'timeout' query arg", d.Seconds(), maxQueryDuration),
-			StatusCode: http.StatusServiceUnavailable,
-		}
-		httpserver.Errorf(w, r, "%s", err)
-	default:
-		httpserver.Errorf(w, r, "unexpected error: %s", err)
-	}
-
-	return true
+func decRequestConcurrency() {
+	<-concurrencyLimitCh
 }
 
 func processSelectRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, path string) bool {
@@ -240,7 +250,7 @@ func processSelectRequest(ctx context.Context, w http.ResponseWriter, r *http.Re
 
 // getMaxQueryDuration returns the maximum duration for query from r.
 func getMaxQueryDuration(r *http.Request) time.Duration {
-	dms, err := httputils.GetDuration(r, "timeout", 0)
+	dms, err := httputil.GetDuration(r, "timeout", 0)
 	if err != nil {
 		dms = 0
 	}

app/vlselect/vmui/index.html

@@ -35,10 +35,10 @@
   <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 type="module" crossorigin src="./assets/index-DuTUAk-m.js"></script>
+  <script type="module" crossorigin src="./assets/index-BgdvCSTM.js"></script>
   <link rel="modulepreload" crossorigin href="./assets/vendor-DojlIpLz.js">
   <link rel="stylesheet" crossorigin href="./assets/vendor-D1GxaB_c.css">
-  <link rel="stylesheet" crossorigin href="./assets/index-CEiptoJw.css">
+  <link rel="stylesheet" crossorigin href="./assets/index-u4IOGr0E.css">
 </head>
 <body>
 <noscript>You need to enable JavaScript to run this app.</noscript>

app/vmagent/csvimport/request_handler.go

@@ -7,9 +7,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/csvimport/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -22,16 +22,16 @@ var (
 
 // InsertHandler processes csv data from req.
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	return stream.Parse(req, func(rows []parser.Row) error {
+	return stream.Parse(req, func(rows []csvimport.Row) error {
 		return insertRows(at, rows, extraLabels)
 	})
 }
 
-func insertRows(at *auth.Token, rows []parser.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []csvimport.Row, extraLabels []prompbmarshal.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 

app/vmagent/datadogsketches/request_handler.go

@@ -7,10 +7,10 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogsketches"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogsketches/stream"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -23,7 +23,7 @@ var (
 
 // InsertHandlerForHTTP processes remote write for DataDog POST /api/beta/sketches request.
 func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
@@ -56,7 +56,7 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 				})
 			}
 			for _, tag := range sketch.Tags {
-				name, value := datadogutils.SplitTag(tag)
+				name, value := datadogutil.SplitTag(tag)
 				if name == "host" {
 					name = "exported_host"
 				}

app/vmagent/datadogv1/request_handler.go

@@ -7,10 +7,10 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv1"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv1/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -23,7 +23,7 @@ var (
 
 // InsertHandlerForHTTP processes remote write for DataDog POST /api/v1/series request.
 func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
@@ -62,7 +62,7 @@ func insertRows(at *auth.Token, series []datadogv1.Series, extraLabels []prompbm
 			})
 		}
 		for _, tag := range ss.Tags {
-			name, value := datadogutils.SplitTag(tag)
+			name, value := datadogutil.SplitTag(tag)
 			if name == "host" {
 				name = "exported_host"
 			}

app/vmagent/datadogv2/request_handler.go

@@ -7,10 +7,10 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv2"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/datadogv2/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -25,7 +25,7 @@ var (
 //
 // See https://docs.datadoghq.com/api/latest/metrics/#submit-metrics
 func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
@@ -65,7 +65,7 @@ func insertRows(at *auth.Token, series []datadogv2.Series, extraLabels []prompbm
 			})
 		}
 		for _, tag := range ss.Tags {
-			name, value := datadogutils.SplitTag(tag)
+			name, value := datadogutil.SplitTag(tag)
 			if name == "host" {
 				name = "exported_host"
 			}

app/vmagent/graphite/request_handler.go

@@ -21,7 +21,7 @@ var (
 //
 // See https://graphite.readthedocs.io/en/latest/feeding-carbon.html#the-plaintext-protocol
 func InsertHandler(r io.Reader) error {
-	return stream.Parse(r, false, func(rows []parser.Row) error {
+	return stream.Parse(r, "", func(rows []parser.Row) error {
 		return insertRows(nil, rows)
 	})
 }

app/vmagent/influx/request_handler.go

@@ -12,9 +12,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/influx/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -35,8 +35,8 @@ var (
 // InsertHandlerForReader processes remote write for influx line protocol.
 //
 // See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
-func InsertHandlerForReader(at *auth.Token, r io.Reader, isGzipped bool) error {
-	return stream.Parse(r, true, isGzipped, "", "", func(db string, rows []parser.Row) error {
+func InsertHandlerForReader(at *auth.Token, r io.Reader, encoding string) error {
+	return stream.Parse(r, encoding, true, "", "", func(db string, rows []influx.Row) error {
 		return insertRows(at, db, rows, nil)
 	})
 }
@@ -45,22 +45,22 @@ func InsertHandlerForReader(at *auth.Token, r io.Reader, isGzipped bool) error {
 //
 // See https://github.com/influxdata/influxdb/blob/4cbdc197b8117fee648d62e2e5be75c6575352f0/tsdb/README.md
 func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-	isStreamMode := req.Header.Get("Stream-Mode") == "1"
 	q := req.URL.Query()
 	precision := q.Get("precision")
 	// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
 	db := q.Get("db")
-	return stream.Parse(req.Body, isStreamMode, isGzipped, precision, db, func(db string, rows []parser.Row) error {
+	encoding := req.Header.Get("Content-Encoding")
+	isStreamMode := req.Header.Get("Stream-Mode") == "1"
+	return stream.Parse(req.Body, encoding, isStreamMode, precision, db, func(db string, rows []influx.Row) error {
 		return insertRows(at, db, rows, extraLabels)
 	})
 }
 
-func insertRows(at *auth.Token, db string, rows []parser.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, db string, rows []influx.Row, extraLabels []prompbmarshal.Label) error {
 	ctx := getPushCtx()
 	defer putPushCtx(ctx)
 

app/vmagent/main.go

@@ -34,7 +34,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/influxutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/influxutil"
 	graphiteserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/graphite"
 	influxserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/influx"
 	opentsdbserver "github.com/VictoriaMetrics/VictoriaMetrics/lib/ingestserver/opentsdb"
@@ -42,8 +42,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"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/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/stringsutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
@@ -105,7 +105,7 @@ func main() {
 	// 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)
+	cgroup.SetGOGC(50)
 
 	// Write flags and help message to stdout, since it is easier to grep or pipe.
 	flag.CommandLine.SetOutput(os.Stdout)
@@ -145,10 +145,10 @@ func main() {
 	startTime := time.Now()
 	remotewrite.StartIngestionRateLimiter()
 	remotewrite.Init()
-	common.StartUnmarshalWorkers()
+	protoparserutil.StartUnmarshalWorkers()
 	if len(*influxListenAddr) > 0 {
 		influxServer = influxserver.MustStart(*influxListenAddr, *influxUseProxyProtocol, func(r io.Reader) error {
-			return influx.InsertHandlerForReader(nil, r, false)
+			return influx.InsertHandlerForReader(nil, r, "")
 		})
 	}
 	if len(*graphiteListenAddr) > 0 {
@@ -195,7 +195,7 @@ func main() {
 	if len(*opentsdbHTTPListenAddr) > 0 {
 		opentsdbhttpServer.MustStop()
 	}
-	common.StopUnmarshalWorkers()
+	protoparserutil.StopUnmarshalWorkers()
 	remotewrite.Stop()
 
 	logger.Infof("successfully stopped vmagent in %.3f seconds", time.Since(startTime).Seconds())
@@ -282,7 +282,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 	}
 	switch path {
 	case "/prometheus/api/v1/write", "/api/v1/write", "/api/v1/push", "/prometheus/api/v1/push":
-		if common.HandleVMProtoServerHandshake(w, r) {
+		if protoparserutil.HandleVMProtoServerHandshake(w, r) {
 			return true
 		}
 		prometheusWriteRequests.Inc()
@@ -331,11 +331,11 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	case "/influx/query", "/query":
 		influxQueryRequests.Inc()
-		influxutils.WriteDatabaseNames(w)
+		influxutil.WriteDatabaseNames(w)
 		return true
 	case "/influx/health":
 		influxHealthRequests.Inc()
-		influxutils.WriteHealthCheckResponse(w)
+		influxutil.WriteHealthCheckResponse(w)
 		return true
 	case "/opentelemetry/api/v1/push", "/opentelemetry/v1/metrics":
 		opentelemetryPushRequests.Inc()
@@ -443,8 +443,10 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 	case "/prometheus/api/v1/targets", "/api/v1/targets":
 		promscrapeAPIV1TargetsRequests.Inc()
 		w.Header().Set("Content-Type", "application/json")
+		// https://prometheus.io/docs/prometheus/latest/querying/api/#targets
 		state := r.FormValue("state")
-		promscrape.WriteAPIV1Targets(w, state)
+		scrapePool := r.FormValue("scrapePool")
+		promscrape.WriteAPIV1Targets(w, state, scrapePool)
 		return true
 	case "/prometheus/target_response", "/target_response":
 		promscrapeTargetResponseRequests.Inc()
@@ -587,11 +589,11 @@ func processMultitenantRequest(w http.ResponseWriter, r *http.Request, path stri
 		return true
 	case "influx/query":
 		influxQueryRequests.Inc()
-		influxutils.WriteDatabaseNames(w)
+		influxutil.WriteDatabaseNames(w)
 		return true
 	case "influx/health":
 		influxHealthRequests.Inc()
-		influxutils.WriteHealthCheckResponse(w)
+		influxutil.WriteHealthCheckResponse(w)
 		return true
 	case "opentelemetry/api/v1/push", "opentelemetry/v1/metrics":
 		opentelemetryPushRequests.Inc()

app/vmagent/native/request_handler.go

@@ -9,8 +9,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/native/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -25,12 +25,12 @@ var (
 //
 // See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	isGzip := req.Header.Get("Content-Encoding") == "gzip"
-	return stream.Parse(req.Body, isGzip, func(block *stream.Block) error {
+	encoding := req.Header.Get("Content-Encoding")
+	return stream.Parse(req.Body, encoding, func(block *stream.Block) error {
 		return insertRows(at, block, extraLabels)
 	})
 }

app/vmagent/newrelic/request_handler.go

@@ -10,9 +10,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/newrelic"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/newrelic/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 )
 
@@ -24,13 +24,12 @@ var (
 
 // InsertHandlerForHTTP processes remote write for NewRelic POST /infra/v2/metrics/events/bulk request.
 func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	ce := req.Header.Get("Content-Encoding")
-	isGzip := ce == "gzip"
-	return stream.Parse(req.Body, isGzip, func(rows []newrelic.Row) error {
+	encoding := req.Header.Get("Content-Encoding")
+	return stream.Parse(req.Body, encoding, func(rows []newrelic.Row) error {
 		return insertRows(at, rows, extraLabels)
 	})
 }

app/vmagent/opentelemetry/request_handler.go

@@ -8,9 +8,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/firehose"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -23,11 +23,11 @@ var (
 
 // InsertHandler processes opentelemetry metrics.
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	encoding := req.Header.Get("Content-Encoding")
 	var processBody func([]byte) ([]byte, error)
 	if req.Header.Get("Content-Type") == "application/json" {
 		if req.Header.Get("X-Amz-Firehose-Protocol-Version") != "" {
@@ -36,7 +36,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 			return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
 		}
 	}
-	return stream.ParseStream(req.Body, isGzipped, processBody, func(tss []prompbmarshal.TimeSeries) error {
+	return stream.ParseStream(req.Body, encoding, processBody, func(tss []prompbmarshal.TimeSeries) error {
 		return insertRows(at, tss, extraLabels)
 	})
 }

app/vmagent/opentsdbhttp/request_handler.go

@@ -7,9 +7,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentsdbhttp/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -21,16 +21,16 @@ var (
 // InsertHandler processes HTTP OpenTSDB put requests.
 // See http://opentsdb.net/docs/build/html/api_http/put.html
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	return stream.Parse(req, func(rows []parser.Row) error {
+	return stream.Parse(req, func(rows []opentsdbhttp.Row) error {
 		return insertRows(at, rows, extraLabels)
 	})
 }
 
-func insertRows(at *auth.Token, rows []parser.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []opentsdbhttp.Row, extraLabels []prompbmarshal.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 

app/vmagent/prometheusimport/request_handler.go

@@ -8,9 +8,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -23,23 +23,23 @@ var (
 
 // InsertHandler processes `/api/v1/import/prometheus` request.
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	defaultTimestamp, err := parserCommon.GetTimestamp(req)
+	defaultTimestamp, err := protoparserutil.GetTimestamp(req)
 	if err != nil {
 		return err
 	}
-	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-	return stream.Parse(req.Body, defaultTimestamp, isGzipped, true, func(rows []parser.Row) error {
+	encoding := req.Header.Get("Content-Encoding")
+	return stream.Parse(req.Body, defaultTimestamp, encoding, true, func(rows []prometheus.Row) error {
 		return insertRows(at, rows, extraLabels)
 	}, func(s string) {
 		httpserver.LogError(req, s)
 	})
 }
 
-func insertRows(at *auth.Token, rows []parser.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []prometheus.Row, extraLabels []prompbmarshal.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 

app/vmagent/prometheusimport/request_handler_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmagent/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 )
 
 var (
@@ -26,7 +26,7 @@ func TestInsertHandler(t *testing.T) {
 	req := httptest.NewRequest(http.MethodPost, "/insert/0/api/v1/import/prometheus", bytes.NewBufferString(`{"foo":"bar"}
 go_memstats_alloc_bytes_total 1`))
 	if err := InsertHandler(nil, req); err != nil {
-		t.Fatalf("unxepected error %s", err)
+		t.Fatalf("unexpected error %s", err)
 	}
 	expectedMsg := "cannot unmarshal Prometheus line"
 	if !strings.Contains(testOutput.String(), expectedMsg) {
@@ -44,14 +44,14 @@ func setUp() {
 		log.Fatalf("unable to set %q with value %q, err: %v", remoteWriteFlag, srv.URL, err)
 	}
 	logger.Init()
-	common.StartUnmarshalWorkers()
+	protoparserutil.StartUnmarshalWorkers()
 	remotewrite.Init()
 	testOutput = &bytes.Buffer{}
 	logger.SetOutputForTests(testOutput)
 }
 
 func tearDown() {
-	common.StopUnmarshalWorkers()
+	protoparserutil.StopUnmarshalWorkers()
 	srv.Close()
 	logger.ResetOutputForTest()
 	tmpDataDir := flag.Lookup("remoteWrite.tmpDataPath").Value.String()

app/vmagent/promremotewrite/request_handler.go

@@ -8,8 +8,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/promremotewrite/stream"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -22,7 +22,7 @@ var (
 
 // InsertHandler processes remote write for prometheus.
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}

app/vmagent/remotewrite/client.go

@@ -14,11 +14,11 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/awsapi"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
@@ -124,14 +124,14 @@ func newHTTPClient(argIdx int, remoteWriteURL, sanitizedURL string, fq *persiste
 	if err != nil {
 		logger.Fatalf("cannot initialize AWS Config for -remoteWrite.url=%q: %s", remoteWriteURL, err)
 	}
-	tr := &http.Transport{
-		DialContext:         netutil.NewStatDialFunc("vmagent_remotewrite"),
-		TLSHandshakeTimeout: tlsHandshakeTimeout.GetOptionalArg(argIdx),
-		MaxConnsPerHost:     2 * concurrency,
-		MaxIdleConnsPerHost: 2 * concurrency,
-		IdleConnTimeout:     time.Minute,
-		WriteBufferSize:     64 * 1024,
-	}
+
+	tr := httputil.NewTransport(false, "vmagent_remotewrite")
+	tr.TLSHandshakeTimeout = tlsHandshakeTimeout.GetOptionalArg(argIdx)
+	tr.MaxConnsPerHost = 2 * concurrency
+	tr.MaxIdleConnsPerHost = 2 * concurrency
+	tr.IdleConnTimeout = time.Minute
+	tr.WriteBufferSize = 64 * 1024
+
 	pURL := proxyURL.GetOptionalArg(argIdx)
 	if len(pURL) > 0 {
 		if !strings.Contains(pURL, "://") {
@@ -170,7 +170,7 @@ func newHTTPClient(argIdx int, remoteWriteURL, sanitizedURL string, fq *persiste
 		doRequest := func(url string) (*http.Response, error) {
 			return c.doRequest(url, nil)
 		}
-		useVMProto = common.HandleVMProtoClientHandshake(c.remoteWriteURL, doRequest)
+		useVMProto = protoparserutil.HandleVMProtoClientHandshake(c.remoteWriteURL, doRequest)
 		if !useVMProto {
 			logger.Infof("the remote storage at %q doesn't support VictoriaMetrics remote write protocol. Switching to Prometheus remote write protocol. "+
 				"See https://docs.victoriametrics.com/vmagent/#victoriametrics-remote-write-protocol", sanitizedURL)

app/vmagent/remotewrite/pendingseries.go

@@ -16,6 +16,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/persistentqueue"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/slicesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/golang/snappy"
@@ -118,7 +119,7 @@ type writeRequest struct {
 }
 
 func (wr *writeRequest) reset() {
-	// Do not reset lastFlushTime, fq, isVMRemoteWrite, significantFigures and roundDigits, since they are re-used.
+	// Do not reset lastFlushTime, fq, isVMRemoteWrite, significantFigures and roundDigits, since they are reused.
 
 	wr.wr.Timeseries = nil
 
@@ -197,28 +198,43 @@ func (wr *writeRequest) tryPush(src []prompbmarshal.TimeSeries) bool {
 }
 
 func (wr *writeRequest) copyTimeSeries(dst, src *prompbmarshal.TimeSeries) {
-	labelsDst := wr.labels
+	labelsSrc := src.Labels
+
+	// Pre-allocate memory for labels.
 	labelsLen := len(wr.labels)
-	samplesDst := wr.samples
-	buf := wr.buf
-	for i := range src.Labels {
-		labelsDst = append(labelsDst, prompbmarshal.Label{})
-		dstLabel := &labelsDst[len(labelsDst)-1]
-		srcLabel := &src.Labels[i]
+	wr.labels = slicesutil.SetLength(wr.labels, labelsLen+len(labelsSrc))
+	labelsDst := wr.labels[labelsLen:]
 
-		buf = append(buf, srcLabel.Name...)
-		dstLabel.Name = bytesutil.ToUnsafeString(buf[len(buf)-len(srcLabel.Name):])
-		buf = append(buf, srcLabel.Value...)
-		dstLabel.Value = bytesutil.ToUnsafeString(buf[len(buf)-len(srcLabel.Value):])
+	// Pre-allocate memory for byte slice needed for storing label names and values.
+	neededBufLen := 0
+	for i := range labelsSrc {
+		label := &labelsSrc[i]
+		neededBufLen += len(label.Name) + len(label.Value)
 	}
-	dst.Labels = labelsDst[labelsLen:]
+	bufLen := len(wr.buf)
+	wr.buf = slicesutil.SetLength(wr.buf, bufLen+neededBufLen)
+	buf := wr.buf[:bufLen]
 
-	samplesDst = append(samplesDst, src.Samples...)
-	dst.Samples = samplesDst[len(samplesDst)-len(src.Samples):]
+	// Copy labels
+	for i := range labelsSrc {
+		dstLabel := &labelsDst[i]
+		srcLabel := &labelsSrc[i]
 
-	wr.samples = samplesDst
-	wr.labels = labelsDst
+		bufLen := len(buf)
+		buf = append(buf, srcLabel.Name...)
+		dstLabel.Name = bytesutil.ToUnsafeString(buf[bufLen:])
+
+		bufLen = len(buf)
+		buf = append(buf, srcLabel.Value...)
+		dstLabel.Value = bytesutil.ToUnsafeString(buf[bufLen:])
+	}
 	wr.buf = buf
+	dst.Labels = labelsDst
+
+	// Copy samples
+	samplesLen := len(wr.samples)
+	wr.samples = append(wr.samples, src.Samples...)
+	dst.Samples = wr.samples[samplesLen:]
 }
 
 // marshalConcurrency limits the maximum number of concurrent workers, which marshal and compress WriteRequest.

app/vmagent/remotewrite/relabel_test.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 func TestApplyRelabeling(t *testing.T) {
@@ -63,7 +63,7 @@ func TestAppendExtraLabels(t *testing.T) {
 func parseSeries(data string) []prompbmarshal.TimeSeries {
 	var tss []prompbmarshal.TimeSeries
 	tss = append(tss, prompbmarshal.TimeSeries{
-		Labels: promutils.MustNewLabelsFromString(data).GetLabels(),
+		Labels: promutil.MustNewLabelsFromString(data).GetLabels(),
 	})
 	return tss
 }

app/vmagent/remotewrite/remotewrite.go

@@ -25,7 +25,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ratelimiter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeserieslimits"
@@ -578,7 +578,7 @@ func tryShardingBlockAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []pr
 	defer putTSSShards(x)
 
 	shards := x.shards
-	tmpLabels := promutils.GetLabels()
+	tmpLabels := promutil.GetLabels()
 	for _, ts := range tssBlock {
 		hashLabels := ts.Labels
 		if len(shardByURLLabelsMap) > 0 {
@@ -613,7 +613,7 @@ func tryShardingBlockAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []pr
 			}
 		}
 	}
-	promutils.PutLabels(tmpLabels)
+	promutil.PutLabels(tmpLabels)
 
 	// Push sharded samples to remote storage systems in parallel in order to reduce
 	// the time needed for sending the data to multiple remote storage systems.

app/vmagent/remotewrite/streamaggr.go

@@ -56,7 +56,7 @@ var (
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignoring-old-samples")
 	streamAggrIgnoreFirstIntervals = flagutil.NewArrayInt("remoteWrite.streamAggr.ignoreFirstIntervals", 0, "Number of aggregation intervals to skip after the start "+
 		"for the corresponding -remoteWrite.streamAggr.config at the corresponding -remoteWrite.url. Increase this value if "+
-		"you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving bufferred delayed data from clients pushing data into the vmagent. "+
+		"you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving buffered delayed data from clients pushing data into the vmagent. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
 	streamAggrDropInputLabels = flagutil.NewArrayString("remoteWrite.streamAggr.dropInputLabels", "An optional list of labels to drop from samples "+
 		"before stream de-duplication and aggregation with -remoteWrite.streamAggr.config and -remoteWrite.streamAggr.dedupInterval at the corresponding -remoteWrite.url. "+
@@ -131,7 +131,7 @@ func reloadStreamAggrConfigGlobal() {
 func initStreamAggrConfigGlobal() {
 	sas, err := newStreamAggrConfigGlobal()
 	if err != nil {
-		logger.Fatalf("cannot initialize gloabl stream aggregators: %s", err)
+		logger.Fatalf("cannot initialize global stream aggregators: %s", err)
 	}
 	if sas != nil {
 		filePath := sas.FilePath()

app/vmagent/vmimport/request_handler.go

@@ -9,8 +9,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	parserCommon "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
-	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/vmimport/stream"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/tenantmetrics"
 	"github.com/VictoriaMetrics/metrics"
@@ -26,17 +26,17 @@ var (
 //
 // See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6
 func InsertHandler(at *auth.Token, req *http.Request) error {
-	extraLabels, err := parserCommon.GetExtraLabels(req)
+	extraLabels, err := protoparserutil.GetExtraLabels(req)
 	if err != nil {
 		return err
 	}
-	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
-	return stream.Parse(req.Body, isGzipped, func(rows []parser.Row) error {
+	encoding := req.Header.Get("Content-Encoding")
+	return stream.Parse(req.Body, encoding, func(rows []vmimport.Row) error {
 		return insertRows(at, rows, extraLabels)
 	})
 }
 
-func insertRows(at *auth.Token, rows []parser.Row, extraLabels []prompbmarshal.Label) error {
+func insertRows(at *auth.Token, rows []vmimport.Row, extraLabels []prompbmarshal.Label) error {
 	ctx := common.GetPushCtx()
 	defer common.PutPushCtx(ctx)
 

app/vmalert-tool/unittest/alerting.go

@@ -1,15 +1,15 @@
 package unittest
 
 import (
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 // alertTestCase holds alert_rule_test cases defined in test file
 type alertTestCase struct {
-	EvalTime  *promutils.Duration `yaml:"eval_time"`
-	GroupName string              `yaml:"groupname"`
-	Alertname string              `yaml:"alertname"`
-	ExpAlerts []expAlert          `yaml:"exp_alerts"`
+	EvalTime  *promutil.Duration `yaml:"eval_time"`
+	GroupName string             `yaml:"groupname"`
+	Alertname string             `yaml:"alertname"`
+	ExpAlerts []expAlert         `yaml:"exp_alerts"`
 }
 
 // expAlert holds exp_alerts defined in test file

app/vmalert-tool/unittest/input.go

@@ -14,7 +14,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"github.com/VictoriaMetrics/metricsql"
 )
 
@@ -41,7 +41,7 @@ func httpWrite(address string, r io.Reader) {
 }
 
 // writeInputSeries send input series to vmstorage and flush them
-func writeInputSeries(input []series, interval *promutils.Duration, startStamp time.Time, dst string) error {
+func writeInputSeries(input []series, interval *promutil.Duration, startStamp time.Time, dst string) error {
 	r := testutil.WriteRequest{}
 	var err error
 	r.Timeseries, err = parseInputSeries(input, interval, startStamp)
@@ -56,7 +56,7 @@ func writeInputSeries(input []series, interval *promutils.Duration, startStamp t
 	return nil
 }
 
-func parseInputSeries(input []series, interval *promutils.Duration, startStamp time.Time) ([]testutil.TimeSeries, error) {
+func parseInputSeries(input []series, interval *promutil.Duration, startStamp time.Time) ([]testutil.TimeSeries, error) {
 	var res []testutil.TimeSeries
 	for _, data := range input {
 		expr, err := metricsql.Parse(data.Series)

app/vmalert-tool/unittest/input_test.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 func TestParseInputValue_Failure(t *testing.T) {
@@ -70,7 +70,7 @@ func TestParseInputValue_Success(t *testing.T) {
 func TestParseInputSeries_Success(t *testing.T) {
 	f := func(input []series) {
 		t.Helper()
-		var interval promutils.Duration
+		var interval promutil.Duration
 		_, err := parseInputSeries(input, &interval, time.Now())
 		if err != nil {
 			t.Fatalf("expect to see no error: %v", err)
@@ -86,7 +86,7 @@ func TestParseInputSeries_Success(t *testing.T) {
 func TestParseInputSeries_Fail(t *testing.T) {
 	f := func(input []series) {
 		t.Helper()
-		var interval promutils.Duration
+		var interval promutil.Duration
 		_, err := parseInputSeries(input, &interval, time.Now())
 		if err == nil {
 			t.Fatalf("expect to see error: %v", err)

app/vmalert-tool/unittest/recording.go

@@ -10,15 +10,15 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"github.com/VictoriaMetrics/metricsql"
 )
 
 // metricsqlTestCase holds metricsql_expr_test cases defined in test file
 type metricsqlTestCase struct {
-	Expr       string              `yaml:"expr"`
-	EvalTime   *promutils.Duration `yaml:"eval_time"`
-	ExpSamples []expSample         `yaml:"exp_samples"`
+	Expr       string             `yaml:"expr"`
+	EvalTime   *promutil.Duration `yaml:"eval_time"`
+	ExpSamples []expSample        `yaml:"exp_samples"`
 }
 
 type expSample struct {
@@ -95,7 +95,7 @@ Outer:
 	return
 }
 
-func durationToTime(pd *promutils.Duration) time.Time {
+func durationToTime(pd *promutil.Duration) time.Time {
 	if pd == nil {
 		return time.Time{}
 	}

app/vmalert-tool/unittest/unittest.go

@@ -36,7 +36,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -182,7 +182,7 @@ func ruleUnitTest(filename string, content []byte, externalLabels map[string]str
 
 	if unitTestInp.EvaluationInterval.Duration() == 0 {
 		fmt.Println("evaluation_interval set to 1m by default")
-		unitTestInp.EvaluationInterval = &promutils.Duration{D: 1 * time.Minute}
+		unitTestInp.EvaluationInterval = &promutil.Duration{D: 1 * time.Minute}
 	}
 
 	groupOrderMap := make(map[string]int)
@@ -312,7 +312,7 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 	defer tearDown()
 
 	if tg.Interval == nil {
-		tg.Interval = promutils.NewDuration(evalInterval)
+		tg.Interval = promutil.NewDuration(evalInterval)
 	}
 	err := writeInputSeries(tg.InputSeries, tg.Interval, testStartTime, fmt.Sprintf("http://127.0.0.1:%s/api/v1/write", httpListenAddr))
 	if err != nil {
@@ -472,15 +472,15 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 
 // unitTestFile holds the contents of a single unit test file
 type unitTestFile struct {
-	RuleFiles          []string            `yaml:"rule_files"`
-	EvaluationInterval *promutils.Duration `yaml:"evaluation_interval"`
-	GroupEvalOrder     []string            `yaml:"group_eval_order"`
-	Tests              []testGroup         `yaml:"tests"`
+	RuleFiles          []string           `yaml:"rule_files"`
+	EvaluationInterval *promutil.Duration `yaml:"evaluation_interval"`
+	GroupEvalOrder     []string           `yaml:"group_eval_order"`
+	Tests              []testGroup        `yaml:"tests"`
 }
 
 // testGroup is a group of input series and test cases associated with it
 type testGroup struct {
-	Interval           *promutils.Duration `yaml:"interval"`
+	Interval           *promutil.Duration  `yaml:"interval"`
 	InputSeries        []series            `yaml:"input_series"`
 	AlertRuleTests     []alertTestCase     `yaml:"alert_rule_test"`
 	MetricsqlExprTests []metricsqlTestCase `yaml:"metricsql_expr_test"`

app/vmalert/config/config.go

@@ -12,9 +12,9 @@ import (
 	"strings"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config/log"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envtemplate"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"gopkg.in/yaml.v2"
 )
 
@@ -27,15 +27,15 @@ var (
 type Group struct {
 	Type       Type `yaml:"type,omitempty"`
 	File       string
-	Name       string              `yaml:"name"`
-	Interval   *promutils.Duration `yaml:"interval,omitempty"`
-	EvalOffset *promutils.Duration `yaml:"eval_offset,omitempty"`
+	Name       string             `yaml:"name"`
+	Interval   *promutil.Duration `yaml:"interval,omitempty"`
+	EvalOffset *promutil.Duration `yaml:"eval_offset,omitempty"`
 	// EvalDelay will adjust the `time` parameter of rule evaluation requests to compensate intentional query delay from datasource.
 	// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5155
-	EvalDelay   *promutils.Duration `yaml:"eval_delay,omitempty"`
-	Limit       int                 `yaml:"limit,omitempty"`
-	Rules       []Rule              `yaml:"rules"`
-	Concurrency int                 `yaml:"concurrency"`
+	EvalDelay   *promutil.Duration `yaml:"eval_delay,omitempty"`
+	Limit       int                `yaml:"limit,omitempty"`
+	Rules       []Rule             `yaml:"rules"`
+	Concurrency int                `yaml:"concurrency"`
 	// Labels is a set of label value pairs, that will be added to every rule.
 	// It has priority over the external labels.
 	Labels map[string]string `yaml:"labels"`
@@ -88,6 +88,9 @@ func (g *Group) Validate(validateTplFn ValidateTplFn, validateExpressions bool)
 	if g.EvalOffset.Duration() > g.Interval.Duration() {
 		return fmt.Errorf("eval_offset should be smaller than interval; now eval_offset: %v, interval: %v", g.EvalOffset.Duration(), g.Interval.Duration())
 	}
+	if g.EvalOffset != nil && g.EvalDelay != nil {
+		return fmt.Errorf("eval_offset cannot be used with eval_delay")
+	}
 	if g.Limit < 0 {
 		return fmt.Errorf("invalid limit %d, shouldn't be less than 0", g.Limit)
 	}
@@ -132,15 +135,15 @@ func (g *Group) Validate(validateTplFn ValidateTplFn, validateExpressions bool)
 // recording rule or alerting rule.
 type Rule struct {
 	ID     uint64
-	Record string              `yaml:"record,omitempty"`
-	Alert  string              `yaml:"alert,omitempty"`
-	Expr   string              `yaml:"expr"`
-	For    *promutils.Duration `yaml:"for,omitempty"`
+	Record string             `yaml:"record,omitempty"`
+	Alert  string             `yaml:"alert,omitempty"`
+	Expr   string             `yaml:"expr"`
+	For    *promutil.Duration `yaml:"for,omitempty"`
 	// Alert will continue firing for this long even when the alerting expression no longer has results.
-	KeepFiringFor *promutils.Duration `yaml:"keep_firing_for,omitempty"`
-	Labels        map[string]string   `yaml:"labels,omitempty"`
-	Annotations   map[string]string   `yaml:"annotations,omitempty"`
-	Debug         bool                `yaml:"debug,omitempty"`
+	KeepFiringFor *promutil.Duration `yaml:"keep_firing_for,omitempty"`
+	Labels        map[string]string  `yaml:"labels,omitempty"`
+	Annotations   map[string]string  `yaml:"annotations,omitempty"`
+	Debug         bool               `yaml:"debug,omitempty"`
 	// UpdateEntriesLimit defines max number of rule's state updates stored in memory.
 	// Overrides `-rule.updateEntriesLimit`.
 	UpdateEntriesLimit *int `yaml:"update_entries_limit,omitempty"`
@@ -262,7 +265,7 @@ func Parse(pathPatterns []string, validateTplFn ValidateTplFn, validateExpressio
 }
 
 func parse(files map[string][]byte, validateTplFn ValidateTplFn, validateExpressions bool) ([]Group, error) {
-	errGroup := new(utils.ErrGroup)
+	errGroup := new(vmalertutil.ErrGroup)
 	var groups []Group
 	for file, data := range files {
 		uniqueGroups := map[string]struct{}{}

app/vmalert/config/config_test.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 	"gopkg.in/yaml.v2"
 )
 
@@ -162,7 +162,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 		Rules: []Rule{
 			{
 				Expr: "sum(up == 0 ) by (host)",
-				For:  promutils.NewDuration(10 * time.Millisecond),
+				For:  promutil.NewDuration(10 * time.Millisecond),
 			},
 			{
 				Expr: "sumSeries(time('foo.bar',10))",
@@ -172,13 +172,13 @@ func TestGroupValidate_Failure(t *testing.T) {
 
 	f(&Group{
 		Name:     "negative interval",
-		Interval: promutils.NewDuration(-1),
+		Interval: promutil.NewDuration(-1),
 	}, false, "interval shouldn't be lower than 0")
 
 	f(&Group{
 		Name:       "wrong eval_offset",
-		Interval:   promutils.NewDuration(time.Minute),
-		EvalOffset: promutils.NewDuration(2 * time.Minute),
+		Interval:   promutil.NewDuration(time.Minute),
+		EvalOffset: promutil.NewDuration(2 * time.Minute),
 	}, false, "eval_offset should be smaller than interval")
 
 	f(&Group{
@@ -309,7 +309,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 				Record: "r1",
 				ID:     1,
 				Expr:   "sumSeries(time('foo.bar',10))",
-				For:    promutils.NewDuration(10 * time.Millisecond),
+				For:    promutil.NewDuration(10 * time.Millisecond),
 			},
 			{
 				Record: "r2",
@@ -326,7 +326,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 			{
 				Record: "r1",
 				Expr:   "sum(up == 0 ) by (host)",
-				For:    promutils.NewDuration(10 * time.Millisecond),
+				For:    promutil.NewDuration(10 * time.Millisecond),
 			},
 		},
 	}, true, "bad LogsQL expr")
@@ -338,7 +338,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 			{
 				Record: "r1",
 				Expr:   "* | stats by (path) count()",
-				For:    promutils.NewDuration(10 * time.Millisecond),
+				For:    promutil.NewDuration(10 * time.Millisecond),
 			},
 		},
 	}, true, "bad prometheus expr")
@@ -488,7 +488,7 @@ func TestHashRule_Equal(t *testing.T) {
 	f(Rule{Alert: "record", Expr: "up == 1"}, Rule{Alert: "record", Expr: "up == 1"})
 
 	f(Rule{
-		Alert: "alert", Expr: "up == 1", For: promutils.NewDuration(time.Minute), KeepFiringFor: promutils.NewDuration(time.Minute),
+		Alert: "alert", Expr: "up == 1", For: promutil.NewDuration(time.Minute), KeepFiringFor: promutil.NewDuration(time.Minute),
 	}, Rule{Alert: "alert", Expr: "up == 1"})
 }
 

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

@@ -9,7 +9,7 @@ groups:
         annotations:
           summary: "{{ }}"
           description: "{{$labels}}"
-      - alert: UnkownAnnotationsFunction
+      - alert: UnknownAnnotationsFunction
         for: 5m
         expr: vm_rows > 0
         labels:

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

@@ -1,7 +1,7 @@
 groups:
   - name: group
     rules:
-      - alert: UnkownLabelFunction
+      - alert: UnknownLabelFunction
         for: 5m
         expr: vm_rows > 0
         labels:

app/vmalert/datasource/client_prom.go

@@ -35,6 +35,8 @@ type promResponse struct {
 	Stats struct {
 		SeriesFetched *string `json:"seriesFetched,omitempty"`
 	} `json:"stats,omitempty"`
+	// IsPartial supported by VictoriaMetrics
+	IsPartial *bool `json:"isPartial,omitempty"`
 }
 
 // see https://prometheus.io/docs/prometheus/latest/querying/api/#instant-queries
@@ -209,7 +211,7 @@ func parsePrometheusResponse(req *http.Request, resp *http.Response) (res Result
 	if err != nil {
 		return res, err
 	}
-	res = Result{Data: ms}
+	res = Result{Data: ms, IsPartial: r.IsPartial}
 	if r.Stats.SeriesFetched != nil {
 		intV, err := strconv.Atoi(*r.Stats.SeriesFetched)
 		if err != nil {

app/vmalert/datasource/client_test.go

@@ -12,7 +12,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
@@ -72,12 +72,14 @@ func TestVMInstantQuery(t *testing.T) {
 			w.Write([]byte(`{"status":"success","data":{"resultType":"scalar","result":[1583786142, "1"]}}`))
 		case 7:
 			w.Write([]byte(`{"status":"success","data":{"resultType":"scalar","result":[1583786142, "1"]},"stats":{"seriesFetched": "42"}}`))
+		case 8:
+			w.Write([]byte(`{"status":"success", "isPartial":true, "data":{"resultType":"scalar","result":[1583786142, "1"]}}`))
 		}
 	})
 	mux.HandleFunc("/render", func(w http.ResponseWriter, _ *http.Request) {
 		c++
 		switch c {
-		case 8:
+		case 9:
 			w.Write([]byte(`[{"target":"constantLine(10)","tags":{"name":"constantLine(10)"},"datapoints":[[10,1611758343],[10,1611758373],[10,1611758403]]}]`))
 		}
 	})
@@ -100,9 +102,9 @@ func TestVMInstantQuery(t *testing.T) {
 			t.Fatalf("failed to parse 'time' query param %q: %s", timeParam, err)
 		}
 		switch c {
-		case 9:
-			w.Write([]byte("[]"))
 		case 10:
+			w.Write([]byte("[]"))
+		case 11:
 			w.Write([]byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"__name__":"total","foo":"bar"},"value":[1583786142,"13763"]},{"metric":{"__name__":"total","foo":"baz"},"value":[1583786140,"2000"]}]}}`))
 		}
 	})
@@ -203,10 +205,18 @@ func TestVMInstantQuery(t *testing.T) {
 			*res.SeriesFetched)
 	}
 
+	res, _, err = pq.Query(ctx, vmQuery, ts) // 8
+	if err != nil {
+		t.Fatalf("unexpected %s", err)
+	}
+	if res.IsPartial != nil && !*res.IsPartial {
+		t.Fatalf("unexpected metric isPartial want %+v", true)
+	}
+
 	// test graphite
 	gq := s.BuildWithParams(QuerierParams{DataSourceType: string(datasourceGraphite)})
 
-	res, _, err = gq.Query(ctx, queryRender, ts) // 8 - graphite
+	res, _, err = gq.Query(ctx, queryRender, ts) // 9 - graphite
 	if err != nil {
 		t.Fatalf("unexpected %s", err)
 	}
@@ -226,9 +236,9 @@ func TestVMInstantQuery(t *testing.T) {
 	vlogs := datasourceVLogs
 	pq = s.BuildWithParams(QuerierParams{DataSourceType: string(vlogs), EvaluationInterval: 15 * time.Second})
 
-	expErr(vlogsQuery, "error parsing response") // 9
+	expErr(vlogsQuery, "error parsing response") // 10
 
-	res, _, err = pq.Query(ctx, vlogsQuery, ts) // 10
+	res, _, err = pq.Query(ctx, vlogsQuery, ts) // 11
 	if err != nil {
 		t.Fatalf("unexpected %s", err)
 	}
@@ -753,7 +763,7 @@ func TestHeaders(t *testing.T) {
 
 	// basic auth
 	f(func() *Client {
-		cfg, err := utils.AuthConfig(utils.WithBasicAuth("foo", "bar", ""))
+		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBasicAuth("foo", "bar", ""))
 		if err != nil {
 			t.Fatalf("Error get auth config: %s", err)
 		}
@@ -766,7 +776,7 @@ func TestHeaders(t *testing.T) {
 
 	// bearer auth
 	f(func() *Client {
-		cfg, err := utils.AuthConfig(utils.WithBearer("foo", ""))
+		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBearer("foo", ""))
 		if err != nil {
 			t.Fatalf("Error get auth config: %s", err)
 		}
@@ -798,7 +808,7 @@ func TestHeaders(t *testing.T) {
 
 	// custom header overrides basic auth
 	f(func() *Client {
-		cfg, err := utils.AuthConfig(utils.WithBasicAuth("foo", "bar", ""))
+		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBasicAuth("foo", "bar", ""))
 		if err != nil {
 			t.Fatalf("Error get auth config: %s", err)
 		}

app/vmalert/datasource/datasource.go

@@ -34,6 +34,9 @@ type Result struct {
 	// If nil, then this feature is not supported by the datasource.
 	// SeriesFetched is supported by VictoriaMetrics since v1.90.
 	SeriesFetched *int
+	// IsPartial is used by VictoriaMetrics to indicate
+	// whether response data is partial.
+	IsPartial *bool
 }
 
 // QuerierBuilder builds Querier with given params.

app/vmalert/datasource/faker.go

@@ -10,8 +10,9 @@ import (
 // FakeQuerier is a mock querier that return predefined results and error message
 type FakeQuerier struct {
 	sync.Mutex
-	metrics []Metric
-	err     error
+	metrics   []Metric
+	err       error
+	isPartial *bool
 }
 
 // SetErr sets query error message
@@ -21,11 +22,19 @@ func (fq *FakeQuerier) SetErr(err error) {
 	fq.Unlock()
 }
 
+// SetPartialResponse marks query response as partial
+func (fq *FakeQuerier) SetPartialResponse(partial bool) {
+	fq.Lock()
+	fq.isPartial = &partial
+	fq.Unlock()
+}
+
 // Reset reset querier's error message and results
 func (fq *FakeQuerier) Reset() {
 	fq.Lock()
 	fq.err = nil
 	fq.metrics = fq.metrics[:0]
+	fq.isPartial = nil
 	fq.Unlock()
 }
 
@@ -57,7 +66,7 @@ func (fq *FakeQuerier) Query(_ context.Context, _ string, _ time.Time) (Result,
 	cp := make([]Metric, len(fq.metrics))
 	copy(cp, fq.metrics)
 	req, _ := http.NewRequest(http.MethodPost, "foo.com", nil)
-	return Result{Data: cp}, req, nil
+	return Result{Data: cp, IsPartial: fq.isPartial}, req, nil
 }
 
 // FakeQuerierWithRegistry can store different results for different query expr

app/vmalert/datasource/init.go

@@ -8,10 +8,10 @@ import (
 	"strings"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 )
 
 var (
@@ -79,15 +79,13 @@ type Param struct {
 // Provided extraParams will be added as GET params for
 // each request.
 func Init(extraParams url.Values) (QuerierBuilder, error) {
-	if *addr == "" {
-		return nil, fmt.Errorf("datasource.url is empty")
+	if err := httputil.CheckURL(*addr); err != nil {
+		return nil, fmt.Errorf("invalid -datasource.url: %w", err)
 	}
-
-	tr, err := httputils.Transport(*addr, *tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify)
+	tr, err := promauth.NewTLSTransport(*tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify, "vmalert_datasource")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport for -datasource.url=%q: %w", *addr, err)
 	}
-	tr.DialContext = netutil.NewStatDialFunc("vmalert_datasource")
 	tr.DisableKeepAlives = *disableKeepAlive
 	tr.MaxIdleConnsPerHost = *maxIdleConnections
 	if tr.MaxIdleConns != 0 && tr.MaxIdleConns < tr.MaxIdleConnsPerHost {
@@ -106,11 +104,11 @@ func Init(extraParams url.Values) (QuerierBuilder, error) {
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse JSON for -datasource.oauth2.endpointParams=%s: %w", *oauth2EndpointParams, err)
 	}
-	authCfg, err := utils.AuthConfig(
-		utils.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
-		utils.WithBearer(*bearerToken, *bearerTokenFile),
-		utils.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
-		utils.WithHeaders(*headers))
+	authCfg, err := vmalertutil.AuthConfig(
+		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
+		vmalertutil.WithBearer(*bearerToken, *bearerTokenFile),
+		vmalertutil.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
+		vmalertutil.WithHeaders(*headers))
 	if err != nil {
 		return nil, fmt.Errorf("failed to configure auth: %w", err)
 	}

app/vmalert/main_test.go

@@ -169,7 +169,6 @@ groups:
 	checkCfg(nil)
 	groupsLen = lenLocked(m)
 	if groupsLen != 2 {
-		fmt.Println(m.groups)
 		t.Fatalf("expected to have exactly 2 groups loaded; got %d", groupsLen)
 	}
 

app/vmalert/manager.go

@@ -83,7 +83,8 @@ func (m *manager) close() {
 
 func (m *manager) startGroup(ctx context.Context, g *rule.Group, restore bool) error {
 	m.wg.Add(1)
-	id := g.ID()
+	id := g.GetID()
+	g.Init()
 	go func() {
 		defer m.wg.Done()
 		if restore {
@@ -112,7 +113,7 @@ func (m *manager) update(ctx context.Context, groupsCfg []config.Group, restore
 			}
 		}
 		ng := rule.NewGroup(cfg, m.querierBuilder, *evaluationInterval, m.labels)
-		groupsRegistry[ng.ID()] = ng
+		groupsRegistry[ng.GetID()] = ng
 	}
 
 	if rrPresent && m.rw == nil {
@@ -130,17 +131,17 @@ func (m *manager) update(ctx context.Context, groupsCfg []config.Group, restore
 
 	m.groupsMu.Lock()
 	for _, og := range m.groups {
-		ng, ok := groupsRegistry[og.ID()]
+		ng, ok := groupsRegistry[og.GetID()]
 		if !ok {
 			// old group is not present in new list,
 			// so must be stopped and deleted
 			og.Close()
-			delete(m.groups, og.ID())
+			delete(m.groups, og.GetID())
 			og = nil
 			continue
 		}
-		delete(groupsRegistry, ng.ID())
-		if og.Checksum != ng.Checksum {
+		delete(groupsRegistry, ng.GetID())
+		if og.GetCheckSum() != ng.GetCheckSum() {
 			toUpdate = append(toUpdate, updateItem{old: og, new: ng})
 		}
 	}

app/vmalert/manager_test.go

@@ -144,7 +144,7 @@ func TestManagerUpdate_Success(t *testing.T) {
 		}
 
 		for _, wantG := range groupsExpected {
-			gotG, ok := m.groups[wantG.ID()]
+			gotG, ok := m.groups[wantG.CreateID()]
 			if !ok {
 				t.Fatalf("expected to have group %q", wantG.Name)
 			}
@@ -258,7 +258,7 @@ func compareGroups(t *testing.T, a, b *rule.Group) {
 	}
 	for i, r := range a.Rules {
 		got, want := r, b.Rules[i]
-		if a.ID() != b.ID() {
+		if a.CreateID() != b.CreateID() {
 			t.Fatalf("expected to have rule %q; got %q", want.ID(), got.ID())
 		}
 		if err := rule.CompareRules(t, want, got); err != nil {

app/vmalert/notifier/alert.go

@@ -9,7 +9,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
@@ -141,7 +141,7 @@ func ValidateTemplates(annotations map[string]string) error {
 func templateAnnotations(annotations map[string]string, data AlertTplData, tmpl *textTpl.Template, execute bool) (map[string]string, error) {
 	var builder strings.Builder
 	var buf bytes.Buffer
-	eg := new(utils.ErrGroup)
+	eg := new(vmalertutil.ErrGroup)
 	r := make(map[string]string, len(annotations))
 	tData := tplData{data, externalLabels, externalURL}
 	header := strings.Join(tplHeaders, "")

app/vmalert/notifier/alertmanager.go

@@ -12,8 +12,8 @@ import (
 
 	"github.com/VictoriaMetrics/metrics"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
@@ -37,8 +37,9 @@ type AlertManager struct {
 type notifierMetrics struct {
 	set *metrics.Set
 
-	alertsSent       *metrics.Counter
-	alertsSendErrors *metrics.Counter
+	alertsSent         *metrics.Counter
+	alertsSendErrors   *metrics.Counter
+	alertsSendDuration *metrics.Histogram
 }
 
 func newNotifierMetrics(addr string) *notifierMetrics {
@@ -46,9 +47,10 @@ func newNotifierMetrics(addr string) *notifierMetrics {
 	metrics.RegisterSet(set)
 
 	return &notifierMetrics{
-		set:              set,
-		alertsSent:       set.GetOrCreateCounter(fmt.Sprintf("vmalert_alerts_sent_total{addr=%q}", addr)),
-		alertsSendErrors: set.GetOrCreateCounter(fmt.Sprintf("vmalert_alerts_send_errors_total{addr=%q}", addr)),
+		set:                set,
+		alertsSent:         set.NewCounter(fmt.Sprintf("vmalert_alerts_sent_total{addr=%q}", addr)),
+		alertsSendErrors:   set.NewCounter(fmt.Sprintf("vmalert_alerts_send_errors_total{addr=%q}", addr)),
+		alertsSendDuration: set.NewHistogram(fmt.Sprintf("vmalert_alerts_send_duration_seconds{addr=%q}", addr)),
 	}
 }
 
@@ -72,7 +74,9 @@ func (am AlertManager) Addr() string {
 // Send an alert or resolve message
 func (am *AlertManager) Send(ctx context.Context, alerts []Alert, headers map[string]string) error {
 	am.metrics.alertsSent.Add(len(alerts))
+	startTime := time.Now()
 	err := am.send(ctx, alerts, headers)
+	am.metrics.alertsSendDuration.UpdateDuration(startTime)
 	if err != nil {
 		am.metrics.alertsSendErrors.Add(len(alerts))
 	}
@@ -148,11 +152,16 @@ const alertManagerPath = "/api/v2/alerts"
 func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, authCfg promauth.HTTPClientConfig,
 	relabelCfg *promrelabel.ParsedConfigs, timeout time.Duration,
 ) (*AlertManager, error) {
+
+	if err := httputil.CheckURL(alertManagerURL); err != nil {
+		return nil, fmt.Errorf("invalid alertmanager URL: %w", err)
+	}
+
 	tls := &promauth.TLSConfig{}
 	if authCfg.TLSConfig != nil {
 		tls = authCfg.TLSConfig
 	}
-	tr, err := httputils.Transport(alertManagerURL, tls.CertFile, tls.KeyFile, tls.CAFile, tls.ServerName, tls.InsecureSkipVerify)
+	tr, err := promauth.NewTLSTransport(tls.CertFile, tls.KeyFile, tls.CAFile, tls.ServerName, tls.InsecureSkipVerify, "vmalert_notifier")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport for alertmanager URL=%q: %w", alertManagerURL, err)
 
@@ -167,11 +176,11 @@ func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, authCfg proma
 		oauth = authCfg.OAuth2
 	}
 
-	aCfg, err := utils.AuthConfig(
-		utils.WithBasicAuth(ba.Username, ba.Password.String(), ba.PasswordFile),
-		utils.WithBearer(authCfg.BearerToken.String(), authCfg.BearerTokenFile),
-		utils.WithOAuth(oauth.ClientID, oauth.ClientSecret.String(), oauth.ClientSecretFile, oauth.TokenURL, strings.Join(oauth.Scopes, ";"), oauth.EndpointParams),
-		utils.WithHeaders(strings.Join(authCfg.Headers, "^^")),
+	aCfg, err := vmalertutil.AuthConfig(
+		vmalertutil.WithBasicAuth(ba.Username, ba.Password.String(), ba.PasswordFile),
+		vmalertutil.WithBearer(authCfg.BearerToken.String(), authCfg.BearerTokenFile),
+		vmalertutil.WithOAuth(oauth.ClientID, oauth.ClientSecret.String(), oauth.ClientSecretFile, oauth.TokenURL, strings.Join(oauth.Scopes, ";"), oauth.EndpointParams),
+		vmalertutil.WithHeaders(strings.Join(authCfg.Headers, "^^")),
 	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to configure auth: %w", err)
@@ -189,8 +198,10 @@ func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, authCfg proma
 		argFunc:        fn,
 		authCfg:        aCfg,
 		relabelConfigs: relabelCfg,
-		client:         &http.Client{Transport: tr},
-		timeout:        timeout,
-		metrics:        newNotifierMetrics(alertManagerURL),
+		client: &http.Client{
+			Transport: tr,
+		},
+		timeout: timeout,
+		metrics: newNotifierMetrics(alertManagerURL),
 	}, nil
 }

app/vmalert/notifier/config.go

@@ -16,7 +16,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/consul"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/dns"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 // Config contains list of supported configuration settings
@@ -44,7 +44,7 @@ type Config struct {
 	// AlertRelabelConfigs contains list of relabeling rules alert labels
 	AlertRelabelConfigs []promrelabel.RelabelConfig `yaml:"alert_relabel_configs,omitempty"`
 	// The timeout used when sending alerts.
-	Timeout *promutils.Duration `yaml:"timeout,omitempty"`
+	Timeout *promutil.Duration `yaml:"timeout,omitempty"`
 
 	// Checksum stores the hash of yaml definition for the config.
 	// May be used to detect any changes to the config file.
@@ -82,7 +82,7 @@ func (cfg *Config) UnmarshalYAML(unmarshal func(any) error) error {
 		cfg.Scheme = "http"
 	}
 	if cfg.Timeout.Duration() == 0 {
-		cfg.Timeout = promutils.NewDuration(time.Second * 10)
+		cfg.Timeout = promutil.NewDuration(time.Second * 10)
 	}
 	rCfg, err := promrelabel.ParseRelabelConfigs(cfg.RelabelConfigs)
 	if err != nil {
@@ -130,7 +130,7 @@ func parseConfig(path string) (*Config, error) {
 	return cfg, nil
 }
 
-func parseLabels(target string, metaLabels *promutils.Labels, cfg *Config) (string, *promutils.Labels, error) {
+func parseLabels(target string, metaLabels *promutil.Labels, cfg *Config) (string, *promutil.Labels, error) {
 	labels := mergeLabels(target, metaLabels, cfg)
 	labels.Labels = cfg.parsedRelabelConfigs.Apply(labels.Labels, 0)
 	labels.RemoveMetaLabels()
@@ -176,9 +176,9 @@ func addMissingPort(scheme, target string) string {
 	return target
 }
 
-func mergeLabels(target string, metaLabels *promutils.Labels, cfg *Config) *promutils.Labels {
+func mergeLabels(target string, metaLabels *promutil.Labels, cfg *Config) *promutil.Labels {
 	// See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config
-	m := promutils.NewLabels(3 + metaLabels.Len())
+	m := promutil.NewLabels(3 + metaLabels.Len())
 	address := target
 	scheme := cfg.Scheme
 	alertsPath := path.Join("/", cfg.PathPrefix, alertManagerPath)

app/vmalert/notifier/config_watcher.go

@@ -9,7 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/consul"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/dns"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 // configWatcher supports dynamic reload of Notifier objects
@@ -157,7 +157,7 @@ func targetsFromLabels(labelsFn getLabels, cfg *Config, genFn AlertURLGenerator)
 	return targets, errors
 }
 
-type getLabels func() ([]*promutils.Labels, error)
+type getLabels func() ([]*promutil.Labels, error)
 
 func (cw *configWatcher) start() error {
 	if len(cw.cfg.StaticConfigs) > 0 {
@@ -183,8 +183,8 @@ func (cw *configWatcher) start() error {
 	}
 
 	if len(cw.cfg.ConsulSDConfigs) > 0 {
-		err := cw.add(TargetConsul, *consul.SDCheckInterval, func() ([]*promutils.Labels, error) {
-			var labels []*promutils.Labels
+		err := cw.add(TargetConsul, *consul.SDCheckInterval, func() ([]*promutil.Labels, error) {
+			var labels []*promutil.Labels
 			for i := range cw.cfg.ConsulSDConfigs {
 				sdc := &cw.cfg.ConsulSDConfigs[i]
 				targetLabels, err := sdc.GetLabels(cw.cfg.baseDir)
@@ -201,8 +201,8 @@ func (cw *configWatcher) start() error {
 	}
 
 	if len(cw.cfg.DNSSDConfigs) > 0 {
-		err := cw.add(TargetDNS, *dns.SDCheckInterval, func() ([]*promutils.Labels, error) {
-			var labels []*promutils.Labels
+		err := cw.add(TargetDNS, *dns.SDCheckInterval, func() ([]*promutil.Labels, error) {
+			var labels []*promutil.Labels
 			for i := range cw.cfg.DNSSDConfigs {
 				sdc := &cw.cfg.DNSSDConfigs[i]
 				targetLabels, err := sdc.GetLabels(cw.cfg.baseDir)

app/vmalert/notifier/init.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 var (
@@ -102,9 +102,9 @@ func Init(gen AlertURLGenerator, extLabels map[string]string, extURL string) (fu
 		if len(*addrs) > 0 || *configPath != "" {
 			return nil, fmt.Errorf("only one of -notifier.blackhole, -notifier.url and -notifier.config flags must be specified")
 		}
-
+		notifier := newBlackHoleNotifier()
 		staticNotifiersFn = func() []Notifier {
-			return []Notifier{newBlackHoleNotifier()}
+			return []Notifier{notifier}
 		}
 		return staticNotifiersFn, nil
 	}
@@ -189,7 +189,7 @@ func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
 // list of labels added during discovery.
 type Target struct {
 	Notifier
-	Labels *promutils.Labels
+	Labels *promutil.Labels
 }
 
 // TargetType defines how the Target was discovered

app/vmalert/remoteread/init.go

@@ -7,10 +7,10 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 )
 
 var (
@@ -66,22 +66,24 @@ func Init() (datasource.QuerierBuilder, error) {
 	if *addr == "" {
 		return nil, nil
 	}
-	tr, err := httputils.Transport(*addr, *tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify)
+	if err := httputil.CheckURL(*addr); err != nil {
+		return nil, fmt.Errorf("invalid -remoteRead.url: %w", err)
+	}
+	tr, err := promauth.NewTLSTransport(*tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify, "vmalert_remoteread")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport for -remoteRead.url=%q: %w", *addr, err)
 	}
 	tr.IdleConnTimeout = *idleConnectionTimeout
-	tr.DialContext = netutil.NewStatDialFunc("vmalert_remoteread")
 
 	endpointParams, err := flagutil.ParseJSONMap(*oauth2EndpointParams)
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse JSON for -remoteRead.oauth2.endpointParams=%s: %w", *oauth2EndpointParams, err)
 	}
-	authCfg, err := utils.AuthConfig(
-		utils.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
-		utils.WithBearer(*bearerToken, *bearerTokenFile),
-		utils.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
-		utils.WithHeaders(*headers))
+	authCfg, err := vmalertutil.AuthConfig(
+		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
+		vmalertutil.WithBearer(*bearerToken, *bearerTokenFile),
+		vmalertutil.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
+		vmalertutil.WithHeaders(*headers))
 	if err != nil {
 		return nil, fmt.Errorf("failed to configure auth: %w", err)
 	}

app/vmalert/remotewrite/client.go

@@ -16,6 +16,7 @@ import (
 	"github.com/golang/snappy"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
@@ -92,7 +93,7 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 		cfg.FlushInterval = defaultFlushInterval
 	}
 	if cfg.Transport == nil {
-		cfg.Transport = http.DefaultTransport.(*http.Transport).Clone()
+		cfg.Transport = httputil.NewTransport(false, "vmalert_remotewrite")
 	}
 	cc := defaultConcurrency
 	if cfg.Concurrency > 0 {

app/vmalert/remotewrite/debug_client.go

@@ -11,7 +11,8 @@ import (
 
 	"github.com/golang/snappy"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
@@ -29,15 +30,17 @@ func NewDebugClient() (*DebugClient, error) {
 	if *addr == "" {
 		return nil, nil
 	}
-
-	t, err := httputils.Transport(*addr, *tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify)
+	if err := httputil.CheckURL(*addr); err != nil {
+		return nil, fmt.Errorf("invalid -remoteWrite.url: %w", err)
+	}
+	tr, err := promauth.NewTLSTransport(*tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify, "vmalert_remotewrite_debug")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport for -remoteWrite.url=%q: %w", *addr, err)
 	}
 	c := &DebugClient{
 		c: &http.Client{
 			Timeout:   *sendTimeout,
-			Transport: t,
+			Transport: tr,
 		},
 		addr: strings.TrimSuffix(*addr, "/"),
 	}

app/vmalert/remotewrite/init.go

@@ -6,10 +6,10 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 )
 
 var (
@@ -69,23 +69,24 @@ func Init(ctx context.Context) (*Client, error) {
 	if *addr == "" {
 		return nil, nil
 	}
-
-	t, err := httputils.Transport(*addr, *tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify)
+	if err := httputil.CheckURL(*addr); err != nil {
+		return nil, fmt.Errorf("invalid -remoteWrite.url: %w", err)
+	}
+	tr, err := promauth.NewTLSTransport(*tlsCertFile, *tlsKeyFile, *tlsCAFile, *tlsServerName, *tlsInsecureSkipVerify, "vmalert_remotewrite")
 	if err != nil {
 		return nil, fmt.Errorf("failed to create transport for -remoteWrite.url=%q: %w", *addr, err)
 	}
-	t.IdleConnTimeout = *idleConnectionTimeout
-	t.DialContext = netutil.NewStatDialFunc("vmalert_remotewrite")
+	tr.IdleConnTimeout = *idleConnectionTimeout
 
 	endpointParams, err := flagutil.ParseJSONMap(*oauth2EndpointParams)
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse JSON for -remoteWrite.oauth2.endpointParams=%s: %w", *oauth2EndpointParams, err)
 	}
-	authCfg, err := utils.AuthConfig(
-		utils.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
-		utils.WithBearer(*bearerToken, *bearerTokenFile),
-		utils.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
-		utils.WithHeaders(*headers))
+	authCfg, err := vmalertutil.AuthConfig(
+		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
+		vmalertutil.WithBearer(*bearerToken, *bearerTokenFile),
+		vmalertutil.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
+		vmalertutil.WithHeaders(*headers))
 	if err != nil {
 		return nil, fmt.Errorf("failed to configure auth: %w", err)
 	}
@@ -97,6 +98,6 @@ func Init(ctx context.Context) (*Client, error) {
 		MaxQueueSize:  *maxQueueSize,
 		MaxBatchSize:  *maxBatchSize,
 		FlushInterval: *flushInterval,
-		Transport:     t,
+		Transport:     tr,
 	})
 }

app/vmalert/replay_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 type fakeReplayQuerier struct {
@@ -88,7 +88,7 @@ func TestReplay(t *testing.T) {
 
 	// datapoints per step
 	f("2021-01-01T12:00:00.000Z", "2021-01-01T15:02:30.000Z", 60, []config.Group{
-		{Interval: promutils.NewDuration(time.Minute), Rules: []config.Rule{{Record: "foo", Expr: "sum(up)"}}},
+		{Interval: promutil.NewDuration(time.Minute), Rules: []config.Rule{{Record: "foo", Expr: "sum(up)"}}},
 	}, &fakeReplayQuerier{
 		registry: map[string]map[string]struct{}{
 			"sum(up)": {

app/vmalert/rule/alerting.go

@@ -15,7 +15,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
@@ -52,18 +52,18 @@ type AlertingRule struct {
 }
 
 type alertingRuleMetrics struct {
-	errors        *utils.Counter
-	pending       *utils.Gauge
-	active        *utils.Gauge
-	samples       *utils.Gauge
-	seriesFetched *utils.Gauge
+	errors        *vmalertutil.Counter
+	pending       *vmalertutil.Gauge
+	active        *vmalertutil.Gauge
+	samples       *vmalertutil.Gauge
+	seriesFetched *vmalertutil.Gauge
 }
 
 func newAlertingRuleMetrics(set *metrics.Set, ar *AlertingRule) *alertingRuleMetrics {
 	labels := fmt.Sprintf(`alertname=%q, group=%q, file=%q, id="%d"`, ar.Name, ar.GroupName, ar.File, ar.ID())
 	arm := &alertingRuleMetrics{}
 
-	arm.pending = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerts_pending{%s}`, labels),
+	arm.pending = vmalertutil.NewGauge(set, fmt.Sprintf(`vmalert_alerts_pending{%s}`, labels),
 		func() float64 {
 			ar.alertsMu.RLock()
 			defer ar.alertsMu.RUnlock()
@@ -75,7 +75,7 @@ func newAlertingRuleMetrics(set *metrics.Set, ar *AlertingRule) *alertingRuleMet
 			}
 			return float64(num)
 		})
-	arm.active = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerts_firing{%s}`, labels),
+	arm.active = vmalertutil.NewGauge(set, fmt.Sprintf(`vmalert_alerts_firing{%s}`, labels),
 		func() float64 {
 			ar.alertsMu.RLock()
 			defer ar.alertsMu.RUnlock()
@@ -87,13 +87,13 @@ func newAlertingRuleMetrics(set *metrics.Set, ar *AlertingRule) *alertingRuleMet
 			}
 			return float64(num)
 		})
-	arm.errors = utils.NewCounter(set, fmt.Sprintf(`vmalert_alerting_rules_errors_total{%s}`, labels))
-	arm.samples = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_samples{%s}`, labels),
+	arm.errors = vmalertutil.NewCounter(set, fmt.Sprintf(`vmalert_alerting_rules_errors_total{%s}`, labels))
+	arm.samples = vmalertutil.NewGauge(set, fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_samples{%s}`, labels),
 		func() float64 {
 			e := ar.state.getLast()
 			return float64(e.Samples)
 		})
-	arm.seriesFetched = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_series_fetched{%s}`, labels),
+	arm.seriesFetched = vmalertutil.NewGauge(set, fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_series_fetched{%s}`, labels),
 		func() float64 {
 			e := ar.state.getLast()
 			if e.SeriesFetched == nil {
@@ -133,7 +133,7 @@ func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 		KeepFiringFor: cfg.KeepFiringFor.Duration(),
 		Labels:        cfg.Labels,
 		Annotations:   cfg.Annotations,
-		GroupID:       group.ID(),
+		GroupID:       group.GetID(),
 		GroupName:     group.Name,
 		File:          group.File,
 		EvalInterval:  group.Interval,
@@ -159,12 +159,11 @@ func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 	ar.state = &ruleState{
 		entries: make([]StateEntry, entrySize),
 	}
-	ar.metrics = newAlertingRuleMetrics(group.metrics.set, ar)
 	return ar
 }
 
-func (ar *AlertingRule) registerMetrics(g *Group) {
-	ar.metrics = newAlertingRuleMetrics(g.metrics.set, ar)
+func (ar *AlertingRule) registerMetrics(set *metrics.Set) {
+	ar.metrics = newAlertingRuleMetrics(set, ar)
 }
 
 // close unregisters rule metrics
@@ -208,8 +207,8 @@ func (ar *AlertingRule) logDebugf(at time.Time, a *notifier.Alert, format string
 	if !ar.Debug {
 		return
 	}
-	prefix := fmt.Sprintf("DEBUG rule %q:%q (%d) at %v: ",
-		ar.GroupName, ar.Name, ar.RuleID, at.Format(time.RFC3339))
+	prefix := fmt.Sprintf("DEBUG alerting rule %q, %q:%q (%d) at %v: ",
+		ar.File, ar.GroupName, ar.Name, ar.RuleID, at.Format(time.RFC3339))
 
 	if a != nil {
 		labelKeys := make([]string, len(a.Labels))
@@ -409,8 +408,8 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 	if err != nil {
 		return nil, fmt.Errorf("failed to execute query %q: %w", ar.Expr, err)
 	}
-	ar.logDebugf(ts, nil, "query returned %d samples (elapsed: %s)", curState.Samples, curState.Duration)
 
+	ar.logDebugf(ts, nil, "query returned %d samples (elapsed: %s, isPartial: %t)", curState.Samples, curState.Duration, isPartialResponse(res))
 	qFn := func(query string) ([]datasource.Metric, error) {
 		res, _, err := ar.q.Query(ctx, query, ts)
 		return res.Data, err

app/vmalert/rule/alerting_test.go

@@ -16,12 +16,38 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
+func TestNewAlertingRule(t *testing.T) {
+	f := func(group *Group, rule config.Rule, expectRule *AlertingRule) {
+		t.Helper()
+
+		r := NewAlertingRule(&datasource.FakeQuerier{}, group, rule)
+		if err := CompareRules(t, expectRule, r); err != nil {
+			t.Fatalf("unexpected rule mismatch: %s", err)
+		}
+	}
+
+	f(&Group{Name: "foo"},
+		config.Rule{
+			Alert: "health",
+			Expr:  "up == 0",
+			Labels: map[string]string{
+				"foo": "bar",
+			},
+		}, &AlertingRule{
+			Name: "health",
+			Expr: "up == 0",
+			Labels: map[string]string{
+				"foo": "bar",
+			},
+		})
+}
+
 func TestAlertingRuleToTimeSeries(t *testing.T) {
 	timestamp := time.Now()
 
@@ -198,7 +224,7 @@ func TestAlertingRule_Exec(t *testing.T) {
 		fakeGroup := Group{
 			Name: "TestRule_Exec",
 		}
-		rule.GroupID = fakeGroup.ID()
+		rule.GroupID = fakeGroup.GetID()
 		for i, step := range steps {
 			fq.Reset()
 			fq.Add(step...)
@@ -556,7 +582,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 
 		fq := &datasource.FakeQuerier{}
 		rule.q = fq
-		rule.GroupID = fakeGroup.ID()
+		rule.GroupID = fakeGroup.GetID()
 		fq.Add(data...)
 		gotTS, err := rule.execRange(context.TODO(), time.Unix(1, 0), time.Unix(5, 0))
 		if err != nil {
@@ -625,7 +651,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{State: notifier.StatePending, ActiveAt: time.Unix(5, 0)},
 	}, map[uint64]*notifier.Alert{
 		hash(map[string]string{"alertname": "for-pending"}): {
-			GroupID:     fakeGroup.ID(),
+			GroupID:     fakeGroup.GetID(),
 			Name:        "for-pending",
 			Labels:      map[string]string{"alertname": "for-pending"},
 			Annotations: map[string]string{"activeAt": "5000"},
@@ -644,7 +670,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{State: notifier.StateFiring, ActiveAt: time.Unix(1, 0)},
 	}, map[uint64]*notifier.Alert{
 		hash(map[string]string{"alertname": "for-firing"}): {
-			GroupID:     fakeGroup.ID(),
+			GroupID:     fakeGroup.GetID(),
 			Name:        "for-firing",
 			Labels:      map[string]string{"alertname": "for-firing"},
 			Annotations: map[string]string{"activeAt": "1000"},
@@ -664,7 +690,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		{State: notifier.StatePending, ActiveAt: time.Unix(5, 0)},
 	}, map[uint64]*notifier.Alert{
 		hash(map[string]string{"alertname": "for-hold-pending"}): {
-			GroupID:     fakeGroup.ID(),
+			GroupID:     fakeGroup.GetID(),
 			Name:        "for-hold-pending",
 			Labels:      map[string]string{"alertname": "for-hold-pending"},
 			Annotations: map[string]string{"activeAt": "5000"},
@@ -719,7 +745,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 		},
 	}, map[uint64]*notifier.Alert{
 		hash(map[string]string{"alertname": "multi-series"}): {
-			GroupID:     fakeGroup.ID(),
+			GroupID:     fakeGroup.GetID(),
 			Name:        "multi-series",
 			Labels:      map[string]string{"alertname": "multi-series"},
 			Annotations: map[string]string{},
@@ -730,7 +756,7 @@ func TestAlertingRuleExecRange(t *testing.T) {
 			For:         3 * time.Second,
 		},
 		hash(map[string]string{"alertname": "multi-series", "foo": "bar"}): {
-			GroupID:     fakeGroup.ID(),
+			GroupID:     fakeGroup.GetID(),
 			Name:        "multi-series",
 			Labels:      map[string]string{"alertname": "multi-series", "foo": "bar"},
 			Annotations: map[string]string{},
@@ -789,6 +815,7 @@ func TestGroup_Restore(t *testing.T) {
 		}
 
 		fg := NewGroup(config.Group{Name: "TestRestore", Rules: rules}, fqr, time.Second, nil)
+		fg.Init()
 		wg := sync.WaitGroup{}
 		wg.Add(1)
 		go func() {
@@ -840,7 +867,7 @@ func TestGroup_Restore(t *testing.T) {
 
 	// one active alert, no previous state
 	fn(
-		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)}},
+		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
@@ -854,7 +881,7 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo"}[3600s])`,
 		stateMetric("foo", ts))
 	fn(
-		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)}},
+		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
@@ -868,8 +895,8 @@ func TestGroup_Restore(t *testing.T) {
 		stateMetric("bar", ts))
 	fn(
 		[]config.Rule{
-			{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)},
-			{Alert: "bar", Expr: "bar", For: promutils.NewDuration(time.Second)},
+			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
+			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Second)},
 		},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
@@ -890,8 +917,8 @@ func TestGroup_Restore(t *testing.T) {
 		stateMetric("bar", ts))
 	fn(
 		[]config.Rule{
-			{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)},
-			{Alert: "bar", Expr: "bar", For: promutils.NewDuration(time.Second)},
+			{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)},
+			{Alert: "bar", Expr: "bar", For: promutil.NewDuration(time.Second)},
 		},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
@@ -909,7 +936,7 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertname="bar",alertgroup="TestRestore"}[3600s])`,
 		stateMetric("wrong alert", ts))
 	fn(
-		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutils.NewDuration(time.Second)}},
+		[]config.Rule{{Alert: "foo", Expr: "foo", For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore"}): {
 				Name:     "foo",
@@ -922,7 +949,7 @@ func TestGroup_Restore(t *testing.T) {
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev"))
 	fn(
-		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutils.NewDuration(time.Second)}},
+		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "dev"}): {
 				Name:     "foo",
@@ -930,12 +957,12 @@ func TestGroup_Restore(t *testing.T) {
 			},
 		})
 
-	// one active alert with restore labels missmatch
+	// one active alert with restore labels mismatch
 	ts = time.Now().Truncate(time.Hour)
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev", "team", "foo"))
 	fn(
-		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutils.NewDuration(time.Second)}},
+		[]config.Rule{{Alert: "foo", Expr: "foo", Labels: map[string]string{"env": "dev"}, For: promutil.NewDuration(time.Second)}},
 		map[uint64]*notifier.Alert{
 			hash(map[string]string{alertNameLabel: "foo", alertGroupNameLabel: "TestRestore", "env": "dev"}): {
 				Name:     "foo",
@@ -975,7 +1002,7 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
 		t.Fatalf("expected to get err; got nil")
 	}
 	if !strings.Contains(err.Error(), expErr) {
-		t.Fatalf("expected to get err %q; got %q insterad", expErr, err)
+		t.Fatalf("expected to get err %q; got %q instead", expErr, err)
 	}
 }
 
@@ -1041,7 +1068,7 @@ func TestAlertingRule_Template(t *testing.T) {
 			Name: "TestRule_Exec",
 		}
 		fq := &datasource.FakeQuerier{}
-		rule.GroupID = fakeGroup.ID()
+		rule.GroupID = fakeGroup.GetID()
 		rule.q = fq
 		rule.state = &ruleState{
 			entries: make([]StateEntry, 10),
@@ -1257,7 +1284,7 @@ func newTestAlertingRule(name string, waitFor time.Duration) *AlertingRule {
 
 func getTestAlertingRuleMetrics(name string) *alertingRuleMetrics {
 	m := &alertingRuleMetrics{}
-	m.errors = utils.NewCounter(metrics.NewSet(), fmt.Sprintf(`vmalert_alerting_rules_errors_total{alertname=%q}`, name))
+	m.errors = vmalertutil.NewCounter(metrics.NewSet(), fmt.Sprintf(`vmalert_alerting_rules_errors_total{alertname=%q}`, name))
 	return m
 }
 
@@ -1321,3 +1348,23 @@ func TestAlertingRule_ToLabels(t *testing.T) {
 		t.Fatalf("processed labels mismatch, got: %v, want: %v", ls.processed, expectedProcessedLabels)
 	}
 }
+
+func TestAlertingRuleExec_Partial(t *testing.T) {
+	fq := &datasource.FakeQuerier{}
+	fq.Add(metricWithValueAndLabels(t, 10, "__name__", "bar"))
+	fq.SetPartialResponse(true)
+
+	ar := newTestAlertingRule("test", 0)
+	ar.Debug = true
+	ar.Labels = map[string]string{"job": "test"}
+	ar.q = fq
+	ar.For = time.Second
+
+	fq.Add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
+
+	ts := time.Now()
+	_, err := ar.exec(context.TODO(), ts, 0)
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+}

app/vmalert/rule/group.go

@@ -19,7 +19,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
@@ -27,19 +27,22 @@ import (
 var (
 	ruleUpdateEntriesLimit = flag.Int("rule.updateEntriesLimit", 20, "Defines the max number of rule's state updates stored in-memory. "+
 		"Rule's updates are available on rule's Details page and are used for debugging purposes. The number of stored updates can be overridden per rule via update_entries_limit param.")
-	resendDelay        = flag.Duration("rule.resendDelay", 0, "MiniMum amount of time to wait before resending an alert to notifier")
+	resendDelay        = flag.Duration("rule.resendDelay", 0, "MiniMum amount of time to wait before resending an alert to notifier.")
 	maxResolveDuration = flag.Duration("rule.maxResolveDuration", 0, "Limits the maxiMum duration for automatic alert expiration, "+
 		"which by default is 4 times evaluationInterval of the parent group")
-	evalDelay = flag.Duration("rule.evalDelay", 30*time.Second, "Adjustment of the `time` parameter for rule evaluation requests to compensate intentional data delay from the datasource."+
-		"Normally, should be equal to `-search.latencyOffset` (cmd-line flag configured for VictoriaMetrics single-node or vmselect).")
+	evalDelay = flag.Duration("rule.evalDelay", 30*time.Second, "Adjustment of the `time` parameter for rule evaluation requests to compensate intentional data delay from the datasource. "+
+		"Normally, should be equal to `-search.latencyOffset` (cmd-line flag configured for VictoriaMetrics single-node or vmselect). "+
+		"This doesn't apply to groups with eval_offset specified.")
 	disableAlertGroupLabel = flag.Bool("disableAlertgroupLabel", false, "Whether to disable adding group's Name as label to generated alerts and time series.")
-	remoteReadLookBack     = flag.Duration("remoteRead.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries."+
-		" For example, if lookback=1h then range from now() to now()-1h will be scanned.")
+	remoteReadLookBack     = flag.Duration("remoteRead.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries. "+
+		"For example, if lookback=1h then range from now() to now()-1h will be scanned.")
 )
 
 // Group is an entity for grouping rules
 type Group struct {
-	mu         sync.RWMutex
+	mu sync.RWMutex
+	// id stores the unique id for the group, and shouldn't change after the group create.
+	id         uint64
 	Name       string
 	File       string
 	Rules      []Rule
@@ -48,10 +51,11 @@ type Group struct {
 	EvalOffset *time.Duration
 	// EvalDelay will adjust timestamp for rule evaluation requests to compensate intentional query delay from datasource.
 	// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5155
-	EvalDelay      *time.Duration
-	Limit          int
-	Concurrency    int
-	Checksum       string
+	EvalDelay   *time.Duration
+	Limit       int
+	Concurrency int
+	// checksum stores the hash of yaml definition for this group.
+	checksum       string
 	LastEvaluation time.Time
 
 	Labels          map[string]string
@@ -67,7 +71,7 @@ type Group struct {
 	// evalCancel stores the cancel fn for interrupting
 	// rules evaluation. Used on groups update() and close().
 	evalCancel context.CancelFunc
-
+	// metrics contains metrics for group and its rules, will be created during Init()
 	metrics *groupMetrics
 	// evalAlignment will make the timestamp of group query
 	// requests be aligned with interval
@@ -83,31 +87,6 @@ type groupMetrics struct {
 	iterationInterval *metrics.Gauge
 }
 
-func newGroupMetrics(g *Group) *groupMetrics {
-	m := &groupMetrics{}
-	m.set = metrics.NewSet()
-
-	labels := fmt.Sprintf(`group=%q, file=%q`, g.Name, g.File)
-	m.iterationTotal = m.set.GetOrCreateCounter(fmt.Sprintf(`vmalert_iteration_total{%s}`, labels))
-	m.iterationDuration = m.set.GetOrCreateSummary(fmt.Sprintf(`vmalert_iteration_duration_seconds{%s}`, labels))
-	m.iterationMissed = m.set.GetOrCreateCounter(fmt.Sprintf(`vmalert_iteration_missed_total{%s}`, labels))
-	m.iterationInterval = m.set.GetOrCreateGauge(fmt.Sprintf(`vmalert_iteration_interval_seconds{%s}`, labels), func() float64 {
-		g.mu.RLock()
-		i := g.Interval.Seconds()
-		g.mu.RUnlock()
-		return i
-	})
-	return m
-}
-
-func (m *groupMetrics) start() {
-	metrics.RegisterSet(m.set)
-}
-
-func (m *groupMetrics) close() {
-	metrics.UnregisterSet(m.set, true)
-}
-
 // merges group rule labels into result map
 // set2 has priority over set1.
 func mergeLabels(groupName, ruleName string, set1, set2 map[string]string) map[string]string {
@@ -134,7 +113,7 @@ func NewGroup(cfg config.Group, qb datasource.QuerierBuilder, defaultInterval ti
 		Interval:        cfg.Interval.Duration(),
 		Limit:           cfg.Limit,
 		Concurrency:     cfg.Concurrency,
-		Checksum:        cfg.Checksum,
+		checksum:        cfg.Checksum,
 		Params:          cfg.Params,
 		Headers:         make(map[string]string),
 		NotifierHeaders: make(map[string]string),
@@ -157,13 +136,13 @@ func NewGroup(cfg config.Group, qb datasource.QuerierBuilder, defaultInterval ti
 	if cfg.EvalDelay != nil {
 		g.EvalDelay = &cfg.EvalDelay.D
 	}
+	g.id = g.CreateID()
 	for _, h := range cfg.Headers {
 		g.Headers[h.Key] = h.Value
 	}
 	for _, h := range cfg.NotifierHeaders {
 		g.NotifierHeaders[h.Key] = h.Value
 	}
-	g.metrics = newGroupMetrics(g)
 	rules := make([]Rule, len(cfg.Rules))
 	for i, r := range cfg.Rules {
 		var extraLabels map[string]string
@@ -193,12 +172,22 @@ func (g *Group) newRule(qb datasource.QuerierBuilder, r config.Rule) Rule {
 	return NewRecordingRule(qb, g, r)
 }
 
-// ID return unique group ID that consists of
-// rules file and group Name
-func (g *Group) ID() uint64 {
+// GetCheckSum returns group checksum
+func (g *Group) GetCheckSum() string {
 	g.mu.RLock()
 	defer g.mu.RUnlock()
+	return g.checksum
+}
 
+// GetID returns the unique group ID
+func (g *Group) GetID() uint64 {
+	return g.id
+}
+
+// CreateID returns the unique ID based on group basic fields.
+// Should only be called when creating new group,
+// and use GetID() afterward.
+func (g *Group) CreateID() uint64 {
 	hash := fnv.New64a()
 	hash.Write([]byte(g.File))
 	hash.Write([]byte("\xff"))
@@ -270,20 +259,17 @@ func (g *Group) updateWith(newGroup *Group) error {
 	}
 	// add the rest of rules from registry
 	for _, nr := range rulesRegistry {
-		nr.registerMetrics(g)
+		nr.registerMetrics(g.metrics.set)
 		newRules = append(newRules, nr)
 	}
-	// note that g.Interval is not updated here
-	// so the value can be compared later in
-	// group.Start function
-	g.Type = newGroup.Type
+
 	g.Concurrency = newGroup.Concurrency
 	g.Params = newGroup.Params
 	g.Headers = newGroup.Headers
 	g.NotifierHeaders = newGroup.NotifierHeaders
 	g.Labels = newGroup.Labels
 	g.Limit = newGroup.Limit
-	g.Checksum = newGroup.Checksum
+	g.checksum = newGroup.checksum
 	g.Rules = newRules
 	return nil
 }
@@ -309,23 +295,42 @@ func (g *Group) Close() {
 	g.InterruptEval()
 	<-g.finishedCh
 
-	g.metrics.close()
+	g.closeGroupMetrics()
+}
+
+func (g *Group) closeGroupMetrics() {
+	metrics.UnregisterSet(g.metrics.set, true)
 }
 
 // SkipRandSleepOnGroupStart will skip random sleep delay in group first evaluation
 var SkipRandSleepOnGroupStart bool
 
+// Init must be called before group Start()
+func (g *Group) Init() {
+	ns := metrics.NewSet()
+	g.metrics = &groupMetrics{set: ns}
+	labels := fmt.Sprintf(`group=%q, file=%q`, g.Name, g.File)
+	g.metrics.iterationTotal = g.metrics.set.NewCounter(fmt.Sprintf(`vmalert_iteration_total{%s}`, labels))
+	g.metrics.iterationDuration = g.metrics.set.NewSummary(fmt.Sprintf(`vmalert_iteration_duration_seconds{%s}`, labels))
+	g.metrics.iterationMissed = g.metrics.set.NewCounter(fmt.Sprintf(`vmalert_iteration_missed_total{%s}`, labels))
+	g.metrics.iterationInterval = g.metrics.set.NewGauge(fmt.Sprintf(`vmalert_iteration_interval_seconds{%s}`, labels), func() float64 {
+		i := g.Interval.Seconds()
+		return i
+	})
+	for i := range g.Rules {
+		g.Rules[i].registerMetrics(g.metrics.set)
+	}
+	metrics.RegisterSet(g.metrics.set)
+}
+
 // Start starts group's evaluation
 func (g *Group) Start(ctx context.Context, nts func() []notifier.Notifier, rw remotewrite.RWClient, rr datasource.QuerierBuilder) {
 	defer func() { close(g.finishedCh) }()
-
-	g.metrics.start()
-
 	evalTS := time.Now()
 	// sleep random duration to spread group rules evaluation
 	// over time in order to reduce load on datasource.
 	if !SkipRandSleepOnGroupStart {
-		sleepBeforeStart := delayBeforeStart(evalTS, g.ID(), g.Interval, g.EvalOffset)
+		sleepBeforeStart := delayBeforeStart(evalTS, g.GetID(), g.Interval, g.EvalOffset)
 		g.infof("will start in %v", sleepBeforeStart)
 
 		sleepTimer := time.NewTimer(sleepBeforeStart)
@@ -375,6 +380,7 @@ func (g *Group) Start(ctx context.Context, nts func() []notifier.Notifier, rw re
 		}
 
 		resolveDuration := getResolveDuration(g.Interval, *resendDelay, *maxResolveDuration)
+		// adjust request timestamp using evalDelay and evalAlignment if necessary
 		ts = g.adjustReqTimestamp(ts)
 		errs := e.execConcurrently(ctx, g.Rules, ts, g.Concurrency, resolveDuration, g.Limit)
 		for err := range errs {
@@ -468,10 +474,18 @@ func (g *Group) DeepCopy() *Group {
 	return &newG
 }
 
-// delayBeforeStart returns a duration on the interval between [ts..ts+interval].
-// delayBeforeStart accounts for `offset`, so returned duration should be always
-// bigger than the `offset`.
+// if offset is specified, delayBeforeStart returns a duration to help aligning timestamp with offset;
+// otherwise, it returns a random duration between [0..interval] based on group key.
 func delayBeforeStart(ts time.Time, key uint64, interval time.Duration, offset *time.Duration) time.Duration {
+	if offset != nil {
+		currentOffsetPoint := ts.Truncate(interval).Add(*offset)
+		if currentOffsetPoint.Before(ts) {
+			// wait until the next offset point
+			return currentOffsetPoint.Add(interval).Sub(ts)
+		}
+		return currentOffsetPoint.Sub(ts)
+	}
+
 	var randSleep time.Duration
 	randSleep = time.Duration(float64(interval) * (float64(key) / (1 << 64)))
 	sleepOffset := time.Duration(ts.UnixNano() % interval.Nanoseconds())
@@ -479,15 +493,6 @@ func delayBeforeStart(ts time.Time, key uint64, interval time.Duration, offset *
 		randSleep += interval
 	}
 	randSleep -= sleepOffset
-	// check if `ts` after randSleep is before `offset`,
-	// if it is, add extra eval_offset to randSleep.
-	// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3409.
-	if offset != nil {
-		tmpEvalTS := ts.Add(randSleep)
-		if tmpEvalTS.Before(tmpEvalTS.Truncate(interval).Add(*offset)) {
-			randSleep += *offset
-		}
-	}
 	return randSleep
 }
 
@@ -593,26 +598,14 @@ func getResolveDuration(groupInterval, delta, maxDuration time.Duration) time.Du
 }
 
 func (g *Group) adjustReqTimestamp(timestamp time.Time) time.Time {
+	// if `eval_offset` is specified, timestamp is already aligned with offset, do nothing
 	if g.EvalOffset != nil {
-		// calculate the min timestamp on the evaluationInterval
-		intervalStart := timestamp.Truncate(g.Interval)
-		ts := intervalStart.Add(*g.EvalOffset)
-		if timestamp.Before(ts) {
-			// if passed timestamp is before the expected evaluation offset,
-			// then we should adjust it to the previous evaluation round.
-			// E.g. request with evaluationInterval=1h and evaluationOffset=30m
-			// was evaluated at 11:20. Then the timestamp should be adjusted
-			// to 10:30, to the previous evaluationInterval.
-			return ts.Add(-g.Interval)
-		}
-		// when `eval_offset` is using, ts shouldn't be effect by `eval_alignment` and `eval_delay`
-		// since it should be always aligned.
-		return ts
+		return timestamp
 	}
 
 	timestamp = timestamp.Add(-g.getEvalDelay())
 
-	// always apply the alignment as a last step
+	// apply the alignment as the last step
 	if g.evalAlignment == nil || *g.evalAlignment {
 		// align query time with interval to get similar result with grafana when plotting time series.
 		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5049
@@ -714,7 +707,7 @@ func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDurati
 	}
 
 	wg := sync.WaitGroup{}
-	errGr := new(utils.ErrGroup)
+	errGr := new(vmalertutil.ErrGroup)
 	for _, nt := range e.Notifiers() {
 		wg.Add(1)
 		go func(nt notifier.Notifier) {

app/vmalert/rule/group_test.go

@@ -18,7 +18,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/templates"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
 
 func init() {
@@ -39,10 +39,11 @@ func TestUpdateWith(t *testing.T) {
 	f := func(currentRules, newRules []config.Rule) {
 		t.Helper()
 
+		ns := metrics.NewSet()
 		g := &Group{
-			Name: "test",
+			Name:    "test",
+			metrics: &groupMetrics{set: ns},
 		}
-		g.metrics = newGroupMetrics(g)
 		qb := &datasource.FakeQuerier{}
 		for _, r := range currentRules {
 			r.ID = config.HashRule(r)
@@ -52,7 +53,6 @@ func TestUpdateWith(t *testing.T) {
 		ng := &Group{
 			Name: "test",
 		}
-		ng.metrics = newGroupMetrics(ng)
 		for _, r := range newRules {
 			r.ID = config.HashRule(r)
 			ng.Rules = append(ng.Rules, ng.newRule(qb, r))
@@ -93,7 +93,7 @@ func TestUpdateWith(t *testing.T) {
 		{
 			Alert: "foo",
 			Expr:  "up > 0",
-			For:   promutils.NewDuration(time.Second),
+			For:   promutil.NewDuration(time.Second),
 			Labels: map[string]string{
 				"bar": "baz",
 			},
@@ -105,7 +105,7 @@ func TestUpdateWith(t *testing.T) {
 		{
 			Alert: "bar",
 			Expr:  "up > 0",
-			For:   promutils.NewDuration(time.Second),
+			For:   promutil.NewDuration(time.Second),
 			Labels: map[string]string{
 				"bar": "baz",
 			},
@@ -114,7 +114,7 @@ func TestUpdateWith(t *testing.T) {
 		{
 			Alert: "foo",
 			Expr:  "up > 10",
-			For:   promutils.NewDuration(time.Second),
+			For:   promutil.NewDuration(time.Second),
 			Labels: map[string]string{
 				"baz": "bar",
 			},
@@ -125,8 +125,8 @@ func TestUpdateWith(t *testing.T) {
 		{
 			Alert:         "bar",
 			Expr:          "up > 0",
-			For:           promutils.NewDuration(2 * time.Second),
-			KeepFiringFor: promutils.NewDuration(time.Minute),
+			For:           promutil.NewDuration(2 * time.Second),
+			KeepFiringFor: promutil.NewDuration(time.Minute),
 			Labels: map[string]string{
 				"bar": "baz",
 			},
@@ -143,6 +143,7 @@ func TestUpdateWith(t *testing.T) {
 	}}, []config.Rule{{
 		Record: "foo",
 		Expr:   "min(up)",
+		Debug:  true,
 		Labels: map[string]string{
 			"baz": "bar",
 		},
@@ -198,7 +199,7 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 		Interval: 100 * time.Hour,
 		updateCh: make(chan *Group),
 	}
-	g.metrics = newGroupMetrics(g)
+	g.Init()
 	go g.Start(context.Background(), nil, nil, nil)
 
 	rule1 := AlertingRule{
@@ -213,7 +214,6 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 			&rule1,
 		},
 	}
-	g1.metrics = newGroupMetrics(g1)
 	g.updateCh <- g1
 	time.Sleep(10 * time.Millisecond)
 	g.mu.RLock()
@@ -237,7 +237,6 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 			&rule2,
 		},
 	}
-	g2.metrics = newGroupMetrics(g2)
 	g.updateCh <- g2
 	time.Sleep(10 * time.Millisecond)
 	g.mu.RLock()
@@ -331,6 +330,7 @@ func TestGroupStart(t *testing.T) {
 	finished := make(chan struct{})
 	fs.Add(m1)
 	fs.Add(m2)
+	g.Init()
 	go func() {
 		g.Start(context.Background(), func() []notifier.Notifier { return []notifier.Notifier{fn} }, nil, fs)
 		close(finished)
@@ -487,6 +487,7 @@ func TestCloseWithEvalInterruption(t *testing.T) {
 
 	const evalInterval = time.Millisecond
 	g := NewGroup(groups[0], fq, evalInterval, nil)
+	g.Init()
 
 	go g.Start(context.Background(), nil, nil, nil)
 
@@ -533,34 +534,14 @@ func TestGroupStartDelay(t *testing.T) {
 	f("2023-01-01T00:00:29.000+00:00", "2023-01-01T00:00:30.000+00:00")
 	f("2023-01-01T00:00:31.000+00:00", "2023-01-01T00:05:30.000+00:00")
 
-	// test group with offset smaller than above fixed randSleep,
-	// this way randSleep will always be enough
-	offset := 20 * time.Second
+	// test group with offset
+	offset := 3 * time.Minute
 	g.EvalOffset = &offset
 
-	f("2023-01-01T00:00:00.000+00:00", "2023-01-01T00:00:30.000+00:00")
-	f("2023-01-01T00:00:29.000+00:00", "2023-01-01T00:00:30.000+00:00")
-	f("2023-01-01T00:00:31.000+00:00", "2023-01-01T00:05:30.000+00:00")
-
-	// test group with offset bigger than above fixed randSleep,
-	// this way offset will be added to delay
-	offset = 3 * time.Minute
-	g.EvalOffset = &offset
-
-	f("2023-01-01T00:00:00.000+00:00", "2023-01-01T00:03:30.000+00:00")
-	f("2023-01-01T00:00:29.000+00:00", "2023-01-01T00:03:30.000+00:00")
-	f("2023-01-01T00:01:00.000+00:00", "2023-01-01T00:08:30.000+00:00")
-	f("2023-01-01T00:03:30.000+00:00", "2023-01-01T00:08:30.000+00:00")
-	f("2023-01-01T00:07:30.000+00:00", "2023-01-01T00:13:30.000+00:00")
-
-	offset = 10 * time.Minute
-	g.EvalOffset = &offset
-	// interval of 1h and key generate a static delay of 6m
-	g.Interval = time.Hour
-
-	f("2023-01-01T00:00:00.000+00:00", "2023-01-01T00:16:00.000+00:00")
-	f("2023-01-01T00:05:00.000+00:00", "2023-01-01T00:16:00.000+00:00")
-	f("2023-01-01T00:30:00.000+00:00", "2023-01-01T01:16:00.000+00:00")
+	f("2023-01-01T00:00:15.000+00:00", "2023-01-01T00:03:00.000+00:00")
+	f("2023-01-01T00:01:00.000+00:00", "2023-01-01T00:03:00.000+00:00")
+	f("2023-01-01T00:03:30.000+00:00", "2023-01-01T00:08:00.000+00:00")
+	f("2023-01-01T00:08:00.000+00:00", "2023-01-01T00:08:00.000+00:00")
 }
 
 func TestGetPrometheusReqTimestamp(t *testing.T) {
@@ -590,17 +571,11 @@ func TestGetPrometheusReqTimestamp(t *testing.T) {
 		evalAlignment: &disableAlign,
 	}, "2023-08-28T11:11:00+00:00", "2023-08-28T11:10:30+00:00")
 
-	// with eval_offset, find previous offset point + default evalDelay
+	// with eval_offset
 	f(&Group{
 		EvalOffset: &offset,
 		Interval:   time.Hour,
-	}, "2023-08-28T11:11:00+00:00", "2023-08-28T10:30:00+00:00")
-
-	// with eval_offset + default evalDelay
-	f(&Group{
-		EvalOffset: &offset,
-		Interval:   time.Hour,
-	}, "2023-08-28T11:41:00+00:00", "2023-08-28T11:30:00+00:00")
+	}, "2023-08-28T11:30:00+00:00", "2023-08-28T11:30:00+00:00")
 
 	// 1h interval with eval_delay
 	f(&Group{

app/vmalert/rule/recording.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
@@ -30,6 +30,7 @@ type RecordingRule struct {
 	GroupID   uint64
 	GroupName string
 	File      string
+	Debug     bool
 
 	q datasource.Querier
 
@@ -43,16 +44,16 @@ type RecordingRule struct {
 }
 
 type recordingRuleMetrics struct {
-	errors  *utils.Counter
-	samples *utils.Gauge
+	errors  *vmalertutil.Counter
+	samples *vmalertutil.Gauge
 }
 
 func newRecordingRuleMetrics(set *metrics.Set, rr *RecordingRule) *recordingRuleMetrics {
 	rmr := &recordingRuleMetrics{}
 
 	labels := fmt.Sprintf(`recording=%q, group=%q, file=%q, id="%d"`, rr.Name, rr.GroupName, rr.File, rr.ID())
-	rmr.errors = utils.NewCounter(set, fmt.Sprintf(`vmalert_recording_rules_errors_total{%s}`, labels))
-	rmr.samples = utils.NewGauge(set, fmt.Sprintf(`vmalert_recording_rules_last_evaluation_samples{%s}`, labels),
+	rmr.errors = vmalertutil.NewCounter(set, fmt.Sprintf(`vmalert_recording_rules_errors_total{%s}`, labels))
+	rmr.samples = vmalertutil.NewGauge(set, fmt.Sprintf(`vmalert_recording_rules_last_evaluation_samples{%s}`, labels),
 		func() float64 {
 			e := rr.state.getLast()
 			return float64(e.Samples)
@@ -88,15 +89,17 @@ func NewRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rul
 		Name:      cfg.Record,
 		Expr:      cfg.Expr,
 		Labels:    cfg.Labels,
-		GroupID:   group.ID(),
+		GroupID:   group.GetID(),
 		GroupName: group.Name,
 		File:      group.File,
+		Debug:     cfg.Debug,
 		q: qb.BuildWithParams(datasource.QuerierParams{
 			DataSourceType:            group.Type.String(),
 			ApplyIntervalAsTimeFilter: setIntervalAsTimeFilter(group.Type.String(), cfg.Expr),
 			EvaluationInterval:        group.Interval,
 			QueryParams:               group.Params,
 			Headers:                   group.Headers,
+			Debug:                     cfg.Debug,
 		}),
 	}
 
@@ -110,13 +113,11 @@ func NewRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rul
 	rr.state = &ruleState{
 		entries: make([]StateEntry, entrySize),
 	}
-	rr.metrics = newRecordingRuleMetrics(group.metrics.set, rr)
-
 	return rr
 }
 
-func (rr *RecordingRule) registerMetrics(g *Group) {
-	rr.metrics = newRecordingRuleMetrics(g.metrics.set, rr)
+func (rr *RecordingRule) registerMetrics(set *metrics.Set) {
+	rr.metrics = newRecordingRuleMetrics(set, rr)
 }
 
 // close unregisters rule metrics
@@ -171,6 +172,8 @@ func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]p
 		return nil, curState.Err
 	}
 
+	rr.logDebugf(ts, "query returned %d samples (elapsed: %s, isPartial: %t)", curState.Samples, curState.Duration, isPartialResponse(res))
+
 	qMetrics := res.Data
 	numSeries := len(qMetrics)
 	if limit > 0 && numSeries > limit {
@@ -204,6 +207,17 @@ func (rr *RecordingRule) exec(ctx context.Context, ts time.Time, limit int) ([]p
 	return tss, nil
 }
 
+func (rr *RecordingRule) logDebugf(at time.Time, format string, args ...any) {
+	if !rr.Debug {
+		return
+	}
+	prefix := fmt.Sprintf("DEBUG recording rule %q, %q:%q (%d) at %v: ",
+		rr.File, rr.GroupName, rr.Name, rr.RuleID, at.Format(time.RFC3339))
+
+	msg := fmt.Sprintf(format, args...)
+	logger.Infof("%s", prefix+msg)
+}
+
 func stringToLabels(s string) []prompbmarshal.Label {
 	labels := strings.Split(s, ",")
 	rLabels := make([]prompbmarshal.Label, 0, len(labels))

app/vmalert/rule/recording_test.go

@@ -9,11 +9,34 @@ import (
 
 	"github.com/VictoriaMetrics/metrics"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
 
+func TestNewRecordingRule(t *testing.T) {
+	f := func(group *Group, rule config.Rule, expectRule *AlertingRule) {
+		t.Helper()
+
+		r := NewAlertingRule(&datasource.FakeQuerier{}, group, rule)
+		if err := CompareRules(t, expectRule, r); err != nil {
+			t.Fatalf("unexpected rule mismatch: %s", err)
+		}
+	}
+
+	f(&Group{Name: "foo"},
+		config.Rule{
+			Alert:  "health",
+			Expr:   "up == 0",
+			Labels: map[string]string{},
+		}, &AlertingRule{
+			Name:   "health",
+			Expr:   "up == 0",
+			Labels: map[string]string{},
+		})
+}
+
 func TestRecordingRule_Exec(t *testing.T) {
 	ts, _ := time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
 	const defaultStep = 5 * time.Millisecond
@@ -194,7 +217,7 @@ func TestRecordingRule_ExecRange(t *testing.T) {
 			t.Fatalf("unexpected RecordingRule.execRange error: %s", err)
 		}
 		if err := compareTimeSeries(t, tssExpected, tss); err != nil {
-			t.Fatalf("timeseries missmatch: %s", err)
+			t.Fatalf("timeseries mismatch: %s", err)
 		}
 	}
 
@@ -307,7 +330,8 @@ func TestRecordingRuleLimit_Failure(t *testing.T) {
 
 		fq := &datasource.FakeQuerier{}
 		fq.Add(testMetrics...)
-		rule := &RecordingRule{Name: "job:foo",
+		rule := &RecordingRule{
+			Name:  "job:foo",
 			state: &ruleState{entries: make([]StateEntry, 10)},
 			Labels: map[string]string{
 				"source": "test_limit",
@@ -342,7 +366,8 @@ func TestRecordingRuleLimit_Success(t *testing.T) {
 
 		fq := &datasource.FakeQuerier{}
 		fq.Add(testMetrics...)
-		rule := &RecordingRule{Name: "job:foo",
+		rule := &RecordingRule{
+			Name:  "job:foo",
 			state: &ruleState{entries: make([]StateEntry, 10)},
 			Labels: map[string]string{
 				"source": "test_limit",
@@ -384,7 +409,7 @@ func TestRecordingRuleExec_Negative(t *testing.T) {
 		t.Fatalf("expected to get err; got nil")
 	}
 	if !strings.Contains(err.Error(), expErr) {
-		t.Fatalf("expected to get err %q; got %q insterad", expErr, err)
+		t.Fatalf("expected to get err %q; got %q instead", expErr, err)
 	}
 
 	fq.Reset()
@@ -421,3 +446,36 @@ func TestSetIntervalAsTimeFilter(t *testing.T) {
 	f(`error AND _time:5m | count()`, "vlogs", false)
 	f(`* | error AND _time:5m | count()`, "vlogs", false)
 }
+
+func TestRecordingRuleExec_Partial(t *testing.T) {
+	ts, _ := time.Parse(time.RFC3339, "2024-10-29T00:00:00Z")
+	fq := &datasource.FakeQuerier{}
+
+	m := metricWithValueAndLabels(t, 10, "__name__", "bar")
+	fq.Add(m)
+	fq.SetPartialResponse(true)
+	rule := &RecordingRule{
+		GroupName: "Bar",
+		Name:      "foo",
+		state: &ruleState{
+			entries: make([]StateEntry, 10),
+		},
+	}
+	rule.Debug = true
+	rule.q = fq
+	got, err := rule.exec(context.TODO(), ts, 0)
+	want := []prompbmarshal.TimeSeries{
+		newTimeSeries([]float64{10}, []int64{ts.UnixNano()}, []prompbmarshal.Label{
+			{
+				Name:  "__name__",
+				Value: "foo",
+			},
+		}),
+	}
+	if err != nil {
+		t.Fatalf("fail to test rule %s: unexpected error: %s", rule.Name, err)
+	}
+	if err := compareTimeSeries(t, want, got); err != nil {
+		t.Fatalf("fail to test rule %s: time series mismatch: %s", rule.Name, err)
+	}
+}

app/vmalert/rule/rule.go

@@ -7,6 +7,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
@@ -30,7 +32,7 @@ type Rule interface {
 	// unregister Rule metrics
 	unregisterMetrics()
 	// register Rule metrics with the given group
-	registerMetrics(g *Group)
+	registerMetrics(set *metrics.Set)
 }
 
 var errDuplicate = errors.New("result contains metrics with the same labelset during evaluation. See https://docs.victoriametrics.com/vmalert/#series-with-the-same-labelset for details")

app/vmalert/rule/test_helpers.go

@@ -38,6 +38,9 @@ func compareRecordingRules(t *testing.T, a, b *RecordingRule) error {
 	if a.Expr != b.Expr {
 		return fmt.Errorf("expected to have expression %q; got %q", a.Expr, b.Expr)
 	}
+	if a.Debug != b.Debug {
+		return fmt.Errorf("expected to have debug=%t; got %t", a.Debug, b.Debug)
+	}
 	if !reflect.DeepEqual(a.Labels, b.Labels) {
 		return fmt.Errorf("expected to have labels %#v; got %#v", a.Labels, b.Labels)
 	}
@@ -64,6 +67,9 @@ func compareAlertingRules(t *testing.T, a, b *AlertingRule) error {
 	if a.Type.String() != b.Type.String() {
 		return fmt.Errorf("expected to have Type %#v; got %#v", a.Type.String(), b.Type.String())
 	}
+	if a.Debug != b.Debug {
+		return fmt.Errorf("expected to have debug=%t; got %t", a.Debug, b.Debug)
+	}
 	return nil
 }
 

app/vmalert/rule/utils.go

@@ -105,3 +105,10 @@ func isSecreteHeader(str string) bool {
 	}
 	return false
 }
+
+func isPartialResponse(res datasource.Result) bool {
+	if res.IsPartial != nil && *res.IsPartial {
+		return true
+	}
+	return false
+}

app/vmalert/tpl/footer.qtpl

@@ -1,12 +1,12 @@
 {% import (
     "net/http"
 
-    "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+    "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 ) %}
 
 
 {% func Footer(r *http.Request) %}
-    {%code prefix := utils.Prefix(r.URL.Path) %}
+    {%code prefix := vmalertutil.Prefix(r.URL.Path) %}
         </main>
         <script src="{%s prefix %}static/js/jquery-3.6.0.min.js" type="text/javascript"></script>
         <script src="{%s prefix %}static/js/bootstrap.bundle.min.js" type="text/javascript"></script>

app/vmalert/tpl/footer.qtpl.go

@@ -8,7 +8,7 @@ package tpl
 import (
 	"net/http"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 )
 
 //line app/vmalert/tpl/footer.qtpl:8
@@ -30,7 +30,7 @@ func StreamFooter(qw422016 *qt422016.Writer, r *http.Request) {
 	qw422016.N().S(`
     `)
 //line app/vmalert/tpl/footer.qtpl:9
-	prefix := utils.Prefix(r.URL.Path)
+	prefix := vmalertutil.Prefix(r.URL.Path)
 
 //line app/vmalert/tpl/footer.qtpl:9
 	qw422016.N().S(`

app/vmalert/tpl/header.qtpl

@@ -3,11 +3,11 @@
     "path"
     "net/url"
 
-    "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+    "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 ) %}
 
 {% func Header(r *http.Request, navItems []NavItem, title string, userErr error) %}
-{%code prefix := utils.Prefix(r.URL.Path) %}
+{%code prefix := vmalertutil.Prefix(r.URL.Path) %}
 <!DOCTYPE html>
 <html lang="en">
 <head>
@@ -85,7 +85,7 @@ type NavItem struct {
 
 {% func printNavItems(r *http.Request, current string, items []NavItem, userErr error) %}
 {%code
-    prefix := utils.Prefix(r.URL.Path)
+    prefix := vmalertutil.Prefix(r.URL.Path)
 %}
 <nav class="navbar navbar-expand-md navbar-dark fixed-top bg-dark">
   <div class="container-fluid">

app/vmalert/tpl/header.qtpl.go

@@ -10,7 +10,7 @@ import (
 	"net/url"
 	"path"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 )
 
 //line app/vmalert/tpl/header.qtpl:9
@@ -32,7 +32,7 @@ func StreamHeader(qw422016 *qt422016.Writer, r *http.Request, navItems []NavItem
 	qw422016.N().S(`
 `)
 //line app/vmalert/tpl/header.qtpl:10
-	prefix := utils.Prefix(r.URL.Path)
+	prefix := vmalertutil.Prefix(r.URL.Path)
 
 //line app/vmalert/tpl/header.qtpl:10
 	qw422016.N().S(`
@@ -165,7 +165,7 @@ func streamprintNavItems(qw422016 *qt422016.Writer, r *http.Request, current str
 	qw422016.N().S(`
 `)
 //line app/vmalert/tpl/header.qtpl:88
-	prefix := utils.Prefix(r.URL.Path)
+	prefix := vmalertutil.Prefix(r.URL.Path)
 
 //line app/vmalert/tpl/header.qtpl:89
 	qw422016.N().S(`