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

app/vlinsert/datadog/datadog.go

@@ -10,13 +10,13 @@ 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/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 )
 
 var (
@@ -61,7 +61,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 		ts = startTime.UnixNano()
 	}
 
-	cp, err := insertutils.GetCommonParams(r)
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return true
@@ -80,8 +80,8 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 	}
 
 	encoding := r.Header.Get("Content-Encoding")
-	err = common.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.NewLogMessageProcessor("datadog")
+	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
@@ -107,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
@@ -170,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,7 +99,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 			httpserver.Errorf(w, r, "%s", err)
 			return true
 		}
-		lmp := cp.NewLogMessageProcessor("elasticsearch_bulk")
+		lmp := cp.NewLogMessageProcessor("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, encoding, cp.TimeField, cp.MsgFields, lmp)
@@ -131,19 +131,19 @@ var (
 	bulkRequestDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/elasticsearch/_bulk"}`)
 )
 
-func readBulkRequest(streamName string, r io.Reader, encoding string, 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
 
-	reader, err := common.GetUncompressedReader(r, encoding)
+	reader, err := protoparserutil.GetUncompressedReader(r, encoding)
 	if err != nil {
 		return 0, fmt.Errorf("cannot decode Elasticsearch protocol data: %w", err)
 	}
-	defer common.PutUncompressedReader(reader)
+	defer protoparserutil.PutUncompressedReader(reader)
 
 	wcr := writeconcurrencylimiter.GetReader(reader)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
-	lr := insertutils.NewLineReader(streamName, wcr)
+	lr := insertutil.NewLineReader(streamName, wcr)
 
 	n := 0
 	for {
@@ -156,7 +156,7 @@ func readBulkRequest(streamName string, r io.Reader, encoding string, timeField
 	}
 }
 
-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"
@@ -228,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

@@ -10,14 +10,14 @@ import (
 	"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, "", "_time", []string{"_msg"}, tlp)
 		if err == nil {
@@ -41,7 +41,7 @@ func TestReadBulkRequest_Success(t *testing.T) {
 		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)
@@ -57,7 +57,7 @@ func TestReadBulkRequest_Success(t *testing.T) {
 		}
 
 		// Read the request with compression
-		tlp = &insertutils.TestLogMessageProcessor{}
+		tlp = &insertutil.TestLogMessageProcessor{}
 		if encoding != "" {
 			data = compressData(data, encoding)
 		}

app/vlinsert/elasticsearch/elasticsearch_timing_test.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 )
 
@@ -42,7 +42,7 @@ func benchmarkReadBulkRequest(b *testing.B, encoding string) {
 
 	timeField := "@timestamp"
 	msgFields := []string{"message"}
-	blp := &insertutils.BenchmarkLogMessageProcessor{}
+	blp := &insertutil.BenchmarkLogMessageProcessor{}
 
 	b.ReportAllocs()
 	b.SetBytes(int64(len(data)))

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,4 +1,4 @@
-package insertutils
+package insertutil
 
 import (
 	"fmt"

app/vlinsert/insertutil/timestamp_test.go

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

app/vlinsert/journald/journald.go

@@ -12,13 +12,13 @@ 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/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/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -41,8 +41,8 @@ var (
 	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
 	}
@@ -100,8 +100,8 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
 	}
 
 	encoding := r.Header.Get("Content-Encoding")
-	err = common.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.NewLogMessageProcessor("journald")
+	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
@@ -131,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
@@ -181,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)

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
@@ -39,14 +39,14 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	encoding := r.Header.Get("Content-Encoding")
-	reader, err := common.GetUncompressedReader(r.Body, encoding)
+	reader, err := protoparserutil.GetUncompressedReader(r.Body, encoding)
 	if err != nil {
 		logger.Errorf("cannot decode jsonline request: %s", err)
 		return
 	}
-	defer common.PutUncompressedReader(reader)
+	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()
@@ -54,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 {
@@ -75,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() {
@@ -91,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

@@ -6,8 +6,8 @@ import (
 	"net/http"
 	"strconv"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
@@ -42,7 +42,7 @@ func handleInsert(r *http.Request, w http.ResponseWriter) {
 }
 
 type commonParams struct {
-	cp *insertutils.CommonParams
+	cp *insertutil.CommonParams
 
 	// Whether to parse JSON inside plaintext log message.
 	//
@@ -51,7 +51,7 @@ type commonParams struct {
 }
 
 func getCommonParams(r *http.Request) (*commonParams, error) {
-	cp, err := insertutils.GetCommonParams(r)
+	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		return nil, err
 	}
@@ -71,7 +71,7 @@ func getCommonParams(r *http.Request) (*commonParams, error) {
 	}
 
 	parseMessage := !*disableMessageParsing
-	if rv := httputils.GetRequestValue(r, "disable_message_parsing", "VL-Loki-Disable-Message-Parsing"); rv != "" {
+	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)

app/vlinsert/loki/loki_json.go

@@ -8,13 +8,13 @@ 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/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"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")
@@ -36,8 +36,8 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 	}
 
 	encoding := r.Header.Get("Content-Encoding")
-	err = common.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.cp.NewLogMessageProcessor("loki_json")
+	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()
@@ -62,7 +62,7 @@ var (
 	requestJSONDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
 )
 
-func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {
+func parseJSONRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {
 	p := parserPool.Get()
 	defer parserPool.Put(p)
 
@@ -222,5 +222,5 @@ func parseLokiTimestamp(s string) (int64, error) {
 		// Special case - an empty timestamp must be substituted with the current time by the caller.
 		return 0, nil
 	}
-	return insertutils.ParseUnixTimestamp(s)
+	return insertutil.ParseUnixTimestamp(s)
 }

app/vlinsert/loki/loki_json_test.go

@@ -3,14 +3,14 @@ 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{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 		if err := parseJSONRequest([]byte(s), tlp, nil, false, false); err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -63,7 +63,7 @@ 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, nil, false, false); err != nil {
 			t.Fatalf("unexpected error: %s", err)
@@ -130,7 +130,7 @@ func TestParseJSONRequest_ParseMessage(t *testing.T) {
 	f := func(s string, msgFields []string, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
-		tlp := &insertutils.TestLogMessageProcessor{}
+		tlp := &insertutil.TestLogMessageProcessor{}
 
 		if err := parseJSONRequest([]byte(s), tlp, msgFields, false, true); err != nil {
 			t.Fatalf("unexpected error: %s", err)

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,7 +22,7 @@ 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) {

app/vlinsert/loki/loki_protobuf.go

@@ -8,11 +8,11 @@ import (
 	"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/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
 )
 
@@ -40,8 +40,8 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 		// See https://grafana.com/docs/loki/latest/reference/loki-http-api/#ingest-logs
 		encoding = "snappy"
 	}
-	err = common.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.cp.NewLogMessageProcessor("loki_protobuf")
+	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()
@@ -66,7 +66,7 @@ var (
 	requestProtobufDuration = metrics.NewHistogram(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
 )
 
-func parseProtobufRequest(data []byte, lmp insertutils.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {
+func parseProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields, parseMessage bool) error {
 	req := getPushRequest()
 	defer putPushRequest(req)
 

app/vlinsert/loki/loki_protobuf_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
@@ -61,7 +61,7 @@ func TestParseProtobufRequest_Success(t *testing.T) {
 
 		data := tlp.pr.MarshalProtobuf(nil)
 
-		tlp2 := &insertutils.TestLogMessageProcessor{}
+		tlp2 := &insertutil.TestLogMessageProcessor{}
 		if err := parseProtobufRequest(data, tlp2, nil, false, false); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -133,7 +133,7 @@ func TestParseProtobufRequest_ParseMessage(t *testing.T) {
 
 		data := tlp.pr.MarshalProtobuf(nil)
 
-		tlp2 := &insertutils.TestLogMessageProcessor{}
+		tlp2 := &insertutil.TestLogMessageProcessor{}
 		if err := parseProtobufRequest(data, tlp2, msgFields, false, true); err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}

app/vlinsert/loki/loki_protobuf_timing_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 )
 
@@ -23,7 +23,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/opentelemetry/opentelemetry.go

@@ -5,13 +5,13 @@ 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/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/metrics"
 )
@@ -39,7 +39,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsProtobufTotal.Inc()
 
-	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
@@ -50,8 +50,8 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 	}
 
 	encoding := r.Header.Get("Content-Encoding")
-	err = common.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.NewLogMessageProcessor("opentelelemtry_protobuf")
+	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()
@@ -75,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()
@@ -99,7 +99,7 @@ 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)]

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

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,7 +16,7 @@ import (
 	"sync/atomic"
 	"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/cgroup"
@@ -25,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"
@@ -287,7 +287,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string,
 		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 {
@@ -351,7 +351,7 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, us
 
 		wg.Add(1)
 		go func() {
-			cp := insertutils.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
+			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)
 			}
@@ -367,29 +367,29 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, us
 }
 
 // processStream parses a stream of syslog messages from r and ingests them into vlstorage.
-func processStream(protocol string, r io.Reader, encoding 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)
+	lmp := cp.NewLogMessageProcessor("syslog_"+protocol, true)
 	err := processStreamInternal(r, encoding, useLocalTimestamp, lmp)
 	lmp.MustClose()
 
 	return err
 }
 
-func processStreamInternal(r io.Reader, encoding string, useLocalTimestamp bool, lmp insertutils.LogMessageProcessor) error {
-	reader, err := common.GetUncompressedReader(r, encoding)
+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 common.PutUncompressedReader(reader)
+	defer protoparserutil.PutUncompressedReader(reader)
 
 	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)
 
@@ -474,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
 		}
@@ -526,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)
@@ -535,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")

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-C68hz-qY.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-B_R5bdPN.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/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"
 )
@@ -36,7 +36,7 @@ var (
 //
 // See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
 func InsertHandlerForReader(at *auth.Token, r io.Reader, encoding string) error {
-	return stream.Parse(r, encoding, true, "", "", func(db string, rows []parser.Row) error {
+	return stream.Parse(r, encoding, true, "", "", func(db string, rows []influx.Row) error {
 		return insertRows(at, db, rows, nil)
 	})
 }
@@ -45,7 +45,7 @@ func InsertHandlerForReader(at *auth.Token, r io.Reader, encoding string) 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
 	}
@@ -55,12 +55,12 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	db := q.Get("db")
 	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 []parser.Row) error {
+	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,7 +145,7 @@ 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, "")
@@ -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,7 +25,7 @@ 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
 	}

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,7 +24,7 @@ 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
 	}

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,7 +23,7 @@ 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
 	}

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
 	}
 	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, defaultTimestamp, encoding, true, func(rows []parser.Row) error {
+	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
 	}
 	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, encoding, func(rows []parser.Row) error {
+	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"`
@@ -135,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"`
@@ -265,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"
@@ -152,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)
 
@@ -171,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)
@@ -193,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"
 )
@@ -40,7 +40,9 @@ var (
 
 // 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
@@ -49,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
@@ -68,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
@@ -84,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.NewCounter(fmt.Sprintf(`vmalert_iteration_total{%s}`, labels))
-	m.iterationDuration = m.set.NewSummary(fmt.Sprintf(`vmalert_iteration_duration_seconds{%s}`, labels))
-	m.iterationMissed = m.set.NewCounter(fmt.Sprintf(`vmalert_iteration_missed_total{%s}`, labels))
-	m.iterationInterval = m.set.NewGauge(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 {
@@ -135,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),
@@ -158,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
@@ -194,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"))
@@ -271,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
 }
@@ -310,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)
@@ -703,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)
 

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(`

app/vmalert/vmalertutil/auth.go

@@ -1,4 +1,4 @@
-package utils
+package vmalertutil
 
 import (
 	"strings"

app/vmalert/vmalertutil/err_group.go

@@ -1,4 +1,4 @@
-package utils
+package vmalertutil
 
 import (
 	"fmt"