debug: [vmagent] log req header and resp header for scrape status code != 200

.github/workflows/main.yml

@@ -85,12 +85,12 @@ jobs:
           restore-keys: go-artifacts-${{ runner.os }}-${{ matrix.scenario }}-
 
       - name: Run tests
-        run: GOGC=10 make ${{ matrix.scenario}}
+        run: make ${{ matrix.scenario}}
 
       - name: Publish coverage
         uses: codecov/codecov-action@v5
         with:
-          files: ./coverage.txt
+          file: ./coverage.txt
 
   integration-test:
     name: integration-test

.github/workflows/sync-docs.yml

@@ -5,7 +5,6 @@ on:
       - 'master'
     paths:
       - 'docs/**'
-      - '.github/workflows/docs.yaml'
   workflow_dispatch: {}
 permissions:
   contents: read  # This is required for actions/checkout and to commit back image update
@@ -18,41 +17,35 @@ jobs:
       - name: Code checkout
         uses: actions/checkout@v4
         with:
-          path: __vm
-
+          path: main
       - name: Checkout private code
         uses: actions/checkout@v4
         with:
           repository: VictoriaMetrics/vmdocs
           token: ${{ secrets.VM_BOT_GH_TOKEN }}
-          path: __vm-docs
-
+          path: docs
       - name: Import GPG key
         uses: crazy-max/ghaction-import-gpg@v6
-        id: import-gpg
         with:
           gpg_private_key: ${{ secrets.VM_BOT_GPG_PRIVATE_KEY }}
           passphrase: ${{ secrets.VM_BOT_PASSPHRASE }}
           git_user_signingkey: true
           git_commit_gpgsign: true
-          git_config_global: true
-
-      - name: Copy docs
-        id: update
+          workdir: docs
+      - name: Set short git commit SHA
+        id: vars
         run: |
-          rsync -zarv \
-            --exclude="Makefile" \
-            docs/ ../__vm-docs/content/victoriametrics
-          echo "SHORT_SHA=$(git rev-parse --short $GITHUB_SHA)" >> $GITHUB_OUTPUT
-        working-directory: __vm
-
-      - name: Push to vmdocs
+          calculatedSha=$(git rev-parse --short ${{ github.sha }})
+          echo "short_sha=$calculatedSha" >> $GITHUB_OUTPUT
+        working-directory: main
+      - name: update code and commit
         run: |
+          rm -rf content
+          cp -r ../main/docs content
+          make clean-after-copy
           git config --global user.name "${{ steps.import-gpg.outputs.email }}"
           git config --global user.email "${{ steps.import-gpg.outputs.email }}"
-          if [[ -n $(git status --porcelain) ]]; then
-            git add .
-            git commit -S -m "sync docs with VictoriaMetrics/VictoriaMetrics commit: ${{ steps.update.outputs.SHORT_SHA }}"
-            git push
-          fi
-        working-directory: __vm-docs
+          git add .
+          git commit -S -m "sync docs with VictoriaMetrics/VictoriaMetrics commit: ${{ steps.vars.outputs.short_sha }}"
+          git push
+        working-directory: docs

LICENSE

@@ -175,7 +175,7 @@
 
    END OF TERMS AND CONDITIONS
 
-   Copyright 2019-2025 VictoriaMetrics, Inc.
+   Copyright 2019-2024 VictoriaMetrics, Inc.
 
    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.

Makefile

@@ -18,7 +18,7 @@ TAR_OWNERSHIP ?= --owner=1000 --group=1000
 .PHONY: $(MAKECMDGOALS)
 
 include app/*/Makefile
-include codespell/Makefile
+include cspell/Makefile
 include docs/Makefile
 include deployment/*/Makefile
 include dashboards/Makefile
@@ -513,19 +513,19 @@ check-all: fmt vet golangci-lint govulncheck
 clean-checkers: remove-golangci-lint remove-govulncheck
 
 test:
-	go test ./lib/... ./app/...
+	DISABLE_FSYNC_FOR_TESTING=1 go test ./lib/... ./app/...
 
 test-race:
-	go test -race ./lib/... ./app/...
+	DISABLE_FSYNC_FOR_TESTING=1 go test -race ./lib/... ./app/...
 
 test-pure:
-	CGO_ENABLED=0 go test ./lib/... ./app/...
+	DISABLE_FSYNC_FOR_TESTING=1 CGO_ENABLED=0 go test ./lib/... ./app/...
 
 test-full:
-	go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
+	DISABLE_FSYNC_FOR_TESTING=1 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
 test-full-386:
-	GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
+	DISABLE_FSYNC_FOR_TESTING=1 GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
 integration-test: victoria-metrics vmagent vmalert vmauth
 	go test ./apptest/... -skip="^TestCluster.*"
@@ -567,7 +567,7 @@ golangci-lint: install-golangci-lint
 	golangci-lint run
 
 install-golangci-lint:
-	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.64.7
+	which golangci-lint || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v1.60.3
 
 remove-golangci-lint:
 	rm -rf `which golangci-lint`

README.md

@@ -1,14 +1,12 @@
 # VictoriaMetrics
 
-![Latest Release](https://img.shields.io/github/v/release/VictoriaMetrics/VictoriaMetrics?sort=semver&label=&filter=!*-victorialogs&logo=github&labelColor=gray&color=gray&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Freleases%2Flatest)
-![Docker Pulls](https://img.shields.io/docker/pulls/victoriametrics/victoria-metrics?label=&logo=docker&logoColor=white&labelColor=2496ED&color=2496ED&link=https%3A%2F%2Fhub.docker.com%2Fr%2Fvictoriametrics%2Fvictoria-metrics)
-![Go Report](https://goreportcard.com/badge/github.com/VictoriaMetrics/VictoriaMetrics?link=https%3A%2F%2Fgoreportcard.com%2Freport%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics)
-![Build Status](https://github.com/VictoriaMetrics/VictoriaMetrics/actions/workflows/main.yml/badge.svg?branch=master&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Factions)
-![codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/branch/master/graph/badge.svg?link=https%3A%2F%2Fcodecov.io%2Fgh%2FVictoriaMetrics%2FVictoriaMetrics)
-![License](https://img.shields.io/github/license/VictoriaMetrics/VictoriaMetrics?labelColor=green&label=&link=https%3A%2F%2Fgithub.com%2FVictoriaMetrics%2FVictoriaMetrics%2Fblob%2Fmaster%2FLICENSE)
-![Slack](https://img.shields.io/badge/Join-4A154B?logo=slack&link=https%3A%2F%2Fslack.victoriametrics.com)
-![X](https://img.shields.io/twitter/follow/VictoriaMetrics?style=flat&label=Follow&color=black&logo=x&labelColor=black&link=https%3A%2F%2Fx.com%2FVictoriaMetrics)
-![Reddit](https://img.shields.io/reddit/subreddit-subscribers/VictoriaMetrics?style=flat&label=Join&labelColor=red&logoColor=white&logo=reddit&link=https%3A%2F%2Fwww.reddit.com%2Fr%2FVictoriaMetrics)
+[![Latest Release](https://img.shields.io/github/release/VictoriaMetrics/VictoriaMetrics.svg?style=flat-square)](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest)
+[![Docker Pulls](https://img.shields.io/docker/pulls/victoriametrics/victoria-metrics.svg?maxAge=604800)](https://hub.docker.com/r/victoriametrics/victoria-metrics)
+[![Slack](https://img.shields.io/badge/join%20slack-%23victoriametrics-brightgreen.svg)](https://slack.victoriametrics.com/)
+[![GitHub license](https://img.shields.io/github/license/VictoriaMetrics/VictoriaMetrics.svg)](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/LICENSE)
+[![Go Report](https://goreportcard.com/badge/github.com/VictoriaMetrics/VictoriaMetrics)](https://goreportcard.com/report/github.com/VictoriaMetrics/VictoriaMetrics)
+[![Build Status](https://github.com/VictoriaMetrics/VictoriaMetrics/workflows/main/badge.svg)](https://github.com/VictoriaMetrics/VictoriaMetrics/actions)
+[![codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/branch/master/graph/badge.svg)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics)
 
 <picture>
   <source srcset="docs/logo_white.webp" media="(prefers-color-scheme: dark)">
@@ -22,7 +20,7 @@ Here are some resources and information about VictoriaMetrics:
 
 - Documentation: [docs.victoriametrics.com](https://docs.victoriametrics.com)
 - Case studies: [Grammarly, Roblox, Wix,...](https://docs.victoriametrics.com/casestudies/).
-- Available: [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest), docker images [Docker Hub](https://hub.docker.com/r/victoriametrics/victoria-metrics/) and [Quay](https://quay.io/repository/victoriametrics/victoria-metrics), [Source code](https://github.com/VictoriaMetrics/VictoriaMetrics)
+- Available: [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest), [Docker images](https://hub.docker.com/r/victoriametrics/victoria-metrics/), [Source code](https://github.com/VictoriaMetrics/VictoriaMetrics)
 - Deployment types: [Single-node version](https://docs.victoriametrics.com/), [Cluster version](https://docs.victoriametrics.com/cluster-victoriametrics/), and [Enterprise version](https://docs.victoriametrics.com/enterprise/)
 - Changelog: [CHANGELOG](https://docs.victoriametrics.com/changelog/), and [How to upgrade](https://docs.victoriametrics.com/#how-to-upgrade-victoriametrics)
 - Community: [Slack](https://slack.victoriametrics.com/), [X (Twitter)](https://x.com/VictoriaMetrics), [LinkedIn](https://www.linkedin.com/company/victoriametrics/), [YouTube](https://www.youtube.com/@VictoriaMetrics)

SECURITY.md

@@ -8,7 +8,7 @@ The following versions of VictoriaMetrics receive regular security fixes:
 |---------|--------------------|
 | [latest release](https://docs.victoriametrics.com/changelog/) | :white_check_mark: |
 | v1.102.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
-| v1.110.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
+| v1.97.x [LTS line](https://docs.victoriametrics.com/lts-releases/) | :white_check_mark: |
 | other releases  | :x:                |
 
 See [this page](https://victoriametrics.com/security/) for more details.

app/victoria-metrics/main.go

@@ -42,10 +42,6 @@ var (
 		"Smaller intervals increase disk IO load. Minimum supported value is 1s")
 	maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
 		"By default there are no limits on samples ingestion rate.")
-	finalDedupScheduleInterval = flag.Duration("storage.finalDedupScheduleCheckInterval", time.Hour, "The interval for checking when final deduplication process should be started."+
-		"Storage unconditionally adds 25% jitter to the interval value on each check evaluation."+
-		" Changing the interval to the bigger values may delay downsampling, deduplication for historical data."+
-		" See also https://docs.victoriametrics.com/#deduplication")
 )
 
 func main() {
@@ -90,10 +86,6 @@ func main() {
 	startTime := time.Now()
 	storage.SetDedupInterval(*minScrapeInterval)
 	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
-	if *finalDedupScheduleInterval < time.Hour {
-		logger.Fatalf("-dedup.finalDedupScheduleCheckInterval cannot be smaller than 1 hour; got %s", *finalDedupScheduleInterval)
-	}
-	storage.SetFinalDedupScheduleInterval(*finalDedupScheduleInterval)
 	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
 	vmselect.Init()
 	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)

app/vlinsert/datadog/datadog.go

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

app/vlinsert/elasticsearch/elasticsearch.go

@@ -60,7 +60,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 	switch path {
-	case "/", "":
+	case "/":
 		switch r.Method {
 		case http.MethodGet:
 			// Return fake response for Elasticsearch ping request.
@@ -100,9 +100,9 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 			return true
 		}
 		lmp := cp.NewLogMessageProcessor("elasticsearch_bulk")
-		encoding := r.Header.Get("Content-Encoding")
+		isGzip := r.Header.Get("Content-Encoding") == "gzip"
 		streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
-		n, err := readBulkRequest(streamName, r.Body, encoding, cp.TimeField, cp.MsgFields, lmp)
+		n, err := readBulkRequest(streamName, r.Body, isGzip, cp.TimeField, cp.MsgFields, lmp)
 		lmp.MustClose()
 		if err != nil {
 			logger.Warnf("cannot decode log message #%d in /_bulk request: %s, stream fields: %s", n, err, cp.StreamFields)
@@ -131,16 +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, isGzip bool, timeField string, msgFields []string, lmp insertutils.LogMessageProcessor) (int, error) {
 	// See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
 
-	reader, err := common.GetUncompressedReader(r, encoding)
-	if err != nil {
-		return 0, fmt.Errorf("cannot decode Elasticsearch protocol data: %w", err)
+	if isGzip {
+		zr, err := common.GetGzipReader(r)
+		if err != nil {
+			return 0, fmt.Errorf("cannot read gzipped _bulk request: %w", err)
+		}
+		defer common.PutGzipReader(zr)
+		r = zr
 	}
-	defer common.PutUncompressedReader(reader)
 
-	wcr := writeconcurrencylimiter.GetReader(reader)
+	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
 	lr := insertutils.NewLineReader(streamName, wcr)

app/vlinsert/elasticsearch/elasticsearch_test.go

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

app/vlinsert/elasticsearch/elasticsearch_timing_test.go

@@ -10,24 +10,15 @@ import (
 )
 
 func BenchmarkReadBulkRequest(b *testing.B) {
-	b.Run("encoding:none", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, "")
+	b.Run("gzip:off", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, false)
 	})
-	b.Run("encoding:gzip", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, "gzip")
-	})
-	b.Run("encoding:zstd", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, "zstd")
-	})
-	b.Run("encoding:deflate", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, "deflate")
-	})
-	b.Run("encoding:snappy", func(b *testing.B) {
-		benchmarkReadBulkRequest(b, "snappy")
+	b.Run("gzip:on", func(b *testing.B) {
+		benchmarkReadBulkRequest(b, true)
 	})
 }
 
-func benchmarkReadBulkRequest(b *testing.B, encoding string) {
+func benchmarkReadBulkRequest(b *testing.B, isGzip bool) {
 	data := `{"create":{"_index":"filebeat-8.8.0"}}
 {"@timestamp":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
 {"create":{"_index":"filebeat-8.8.0"}}
@@ -35,8 +26,8 @@ func benchmarkReadBulkRequest(b *testing.B, encoding string) {
 {"create":{"_index":"filebeat-8.8.0"}}
 {"message":"xyz","@timestamp":"2023-06-06T04:48:13.735Z","x":"y"}
 `
-	if encoding != "" {
-		data = compressData(data, encoding)
+	if isGzip {
+		data = compressData(data)
 	}
 	dataBytes := bytesutil.ToUnsafeBytes(data)
 
@@ -50,7 +41,7 @@ func benchmarkReadBulkRequest(b *testing.B, encoding string) {
 		r := &bytes.Reader{}
 		for pb.Next() {
 			r.Reset(dataBytes)
-			_, err := readBulkRequest("test", r, encoding, timeField, msgFields, blp)
+			_, err := readBulkRequest("test", r, isGzip, timeField, msgFields, blp)
 			if err != nil {
 				panic(fmt.Errorf("unexpected error: %w", err))
 			}

app/vlinsert/insertutils/common_params.go

@@ -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 reuse them.
+	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can re-use them.
 	AddRow(timestamp int64, fields, streamFields []logstorage.Field)
 
 	// MustClose() must flush all the remaining fields and free up resources occupied by LogMessageProcessor.
@@ -199,8 +199,8 @@ func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []l
 	lmp.bytesIngestedTotal.Add(n)
 
 	if len(fields) > *MaxFieldsPerLine {
-		line := logstorage.MarshalFieldsToJSON(nil, fields)
-		logger.Warnf("dropping log line with %d fields; it exceeds -insert.maxFieldsPerLine=%d; %s", len(fields), *MaxFieldsPerLine, line)
+		rf := logstorage.RowFormatter(fields)
+		logger.Warnf("dropping log line with %d fields; it exceeds -insert.maxFieldsPerLine=%d; %s", len(fields), *MaxFieldsPerLine, rf)
 		rowsDroppedTotalTooManyFields.Inc()
 		return
 	}

app/vlinsert/insertutils/flags.go

@@ -8,10 +8,8 @@ import (
 
 var (
 	// MaxLineSizeBytes is the maximum length of a single line for /insert/* handlers
-	MaxLineSizeBytes = flagutil.NewBytes("insert.maxLineSizeBytes", 256*1024, "The maximum size of a single line, which can be read by /insert/* handlers; "+
-		"see https://docs.victoriametrics.com/victorialogs/faq/#what-length-a-log-record-is-expected-to-have")
+	MaxLineSizeBytes = flagutil.NewBytes("insert.maxLineSizeBytes", 256*1024, "The maximum size of a single line, which can be read by /insert/* handlers")
 
 	// MaxFieldsPerLine is the maximum number of fields per line for /insert/* handlers
-	MaxFieldsPerLine = flag.Int("insert.maxFieldsPerLine", 1000, "The maximum number of log fields per line, which can be read by /insert/* handlers; "+
-		"see https://docs.victoriametrics.com/victorialogs/faq/#how-many-fields-a-single-log-entry-may-contain")
+	MaxFieldsPerLine = flag.Int("insert.maxFieldsPerLine", 1000, "The maximum number of log fields per line, which can be read by /insert/* handlers")
 )

app/vlinsert/insertutils/line_reader.go

@@ -15,8 +15,6 @@ import (
 // LineReader reads newline-delimited lines from the underlying reader
 type LineReader struct {
 	// Line contains the next line read after the call to NextLine
-	//
-	// The Line contents is valid until the next call to NextLine.
 	Line []byte
 
 	// name is the LineReader name
@@ -28,9 +26,6 @@ type LineReader struct {
 	// buf is a buffer for reading the next line
 	buf []byte
 
-	// bufOffset is the offset at buf to read the next line from
-	bufOffset int
-
 	// err is the last error when reading data from r
 	err error
 
@@ -56,27 +51,26 @@ func NewLineReader(name string, r io.Reader) *LineReader {
 // Check for Err in this case.
 func (lr *LineReader) NextLine() bool {
 	for {
-		if lr.bufOffset >= len(lr.buf) {
+		if len(lr.buf) == 0 {
 			if lr.err != nil || lr.eofReached {
 				return false
 			}
 			if !lr.readMoreData() {
 				return false
 			}
-			if lr.bufOffset >= len(lr.buf) && lr.eofReached {
+			if len(lr.buf) == 0 && lr.eofReached {
 				return false
 			}
 		}
 
-		buf := lr.buf[lr.bufOffset:]
-		if n := bytes.IndexByte(buf, '\n'); n >= 0 {
-			lr.Line = buf[:n]
-			lr.bufOffset += n + 1
+		if n := bytes.IndexByte(lr.buf, '\n'); n >= 0 {
+			lr.Line = append(lr.Line[:0], lr.buf[:n]...)
+			lr.buf = append(lr.buf[:0], lr.buf[n+1:]...)
 			return true
 		}
 		if lr.eofReached {
-			lr.Line = buf
-			lr.bufOffset += len(buf)
+			lr.Line = append(lr.Line[:0], lr.buf...)
+			lr.buf = lr.buf[:0]
 			return true
 		}
 		if !lr.readMoreData() {
@@ -94,11 +88,6 @@ func (lr *LineReader) Err() error {
 }
 
 func (lr *LineReader) readMoreData() bool {
-	if lr.bufOffset > 0 {
-		lr.buf = append(lr.buf[:0], lr.buf[lr.bufOffset:]...)
-		lr.bufOffset = 0
-	}
-
 	bufLen := len(lr.buf)
 	if bufLen >= MaxLineSizeBytes.IntN() {
 		logger.Warnf("%s: the line length exceeds -insert.maxLineSizeBytes=%d; skipping it; line contents=%q", lr.name, MaxLineSizeBytes.IntN(), lr.buf)

app/vlinsert/insertutils/timestamp.go

@@ -4,19 +4,18 @@ import (
 	"fmt"
 	"math"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
-// ExtractTimestampFromFields extracts timestamp in nanoseconds from the field with the name timeField at fields.
+// ExtractTimestampRFC3339NanoFromFields extracts RFC3339 timestamp in nanoseconds from the field with the name timeField at fields.
 //
 // The value for the timeField is set to empty string after returning from the function,
 // so it could be ignored during data ingestion.
 //
 // The current timestamp is returned if fields do not contain a field with timeField name or if the timeField value is empty.
-func ExtractTimestampFromFields(timeField string, fields []logstorage.Field) (int64, error) {
+func ExtractTimestampRFC3339NanoFromFields(timeField string, fields []logstorage.Field) (int64, error) {
 	for i := range fields {
 		f := &fields[i]
 		if f.Name != timeField {
@@ -49,53 +48,22 @@ func parseTimestamp(s string) (int64, error) {
 	return nsecs, nil
 }
 
-// ParseUnixTimestamp parses s as unix timestamp in seconds, milliseconds, microseconds or nanoseconds and returns the parsed timestamp in nanoseconds.
+// ParseUnixTimestamp parses s as unix timestamp in either seconds or milliseconds and returns the parsed timestamp in nanoseconds.
 func ParseUnixTimestamp(s string) (int64, error) {
-	if strings.IndexByte(s, '.') >= 0 {
-		// Parse timestamp as floating-point value
-		f, err := strconv.ParseFloat(s, 64)
-		if err != nil {
-			return 0, fmt.Errorf("cannot parse unix timestamp from %q: %w", s, err)
-		}
-		if f < (1<<31) && f >= (-1<<31) {
-			// The timestamp is in seconds.
-			return int64(f * 1e9), nil
-		}
-		if f < 1e3*(1<<31) && f >= 1e3*(-1<<31) {
-			// The timestamp is in milliseconds.
-			return int64(f * 1e6), nil
-		}
-		if f < 1e6*(1<<31) && f >= 1e6*(-1<<31) {
-			// The timestamp is in microseconds.
-			return int64(f * 1e3), nil
-		}
-		// The timestamp is in nanoseconds
-		if f > math.MaxInt64 {
-			return 0, fmt.Errorf("too big timestamp in nanoseconds: %v; mustn't exceed %v", f, int64(math.MaxInt64))
-		}
-		if f < math.MinInt64 {
-			return 0, fmt.Errorf("too small timestamp in nanoseconds: %v; must be bigger or equal to %v", f, int64(math.MinInt64))
-		}
-		return int64(f), nil
-	}
-
-	// Parse timestamp as integer
 	n, err := strconv.ParseInt(s, 10, 64)
 	if err != nil {
 		return 0, fmt.Errorf("cannot parse unix timestamp from %q: %w", s, err)
 	}
 	if n < (1<<31) && n >= (-1<<31) {
-		// The timestamp is in seconds.
-		return n * 1e9, nil
+		// The timestamp is in seconds. Convert it to milliseconds
+		n *= 1e3
 	}
-	if n < 1e3*(1<<31) && n >= 1e3*(-1<<31) {
-		// The timestamp is in milliseconds.
-		return n * 1e6, nil
+	if n > int64(math.MaxInt64)/1e6 {
+		return 0, fmt.Errorf("too big timestamp in milliseconds: %d; mustn't exceed %d", n, int64(math.MaxInt64)/1e6)
 	}
-	if n < 1e6*(1<<31) && n >= 1e6*(-1<<31) {
-		// The timestamp is in microseconds.
-		return n * 1e3, nil
+	if n < int64(math.MinInt64)/1e6 {
+		return 0, fmt.Errorf("too small timestamp in milliseconds: %d; must be bigger than %d", n, int64(math.MinInt64)/1e6)
 	}
-	// The timestamp is in nanoseconds
+	n *= 1e6
 	return n, nil
 }

app/vlinsert/insertutils/timestamp_test.go

@@ -6,68 +6,11 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 )
 
-func TestParseUnixTimestamp_Success(t *testing.T) {
-	f := func(s string, timestampExpected int64) {
-		t.Helper()
-
-		timestamp, err := ParseUnixTimestamp(s)
-		if err != nil {
-			t.Fatalf("unexpected error in ParseUnixTimestamp(%q): %s", s, err)
-		}
-		if timestamp != timestampExpected {
-			t.Fatalf("unexpected timestamp returned from ParseUnixTimestamp(%q); got %d; want %d", s, timestamp, timestampExpected)
-		}
-	}
-
-	f("0", 0)
-
-	// nanoseconds
-	f("-1234567890123456789", -1234567890123456789)
-	f("1234567890123456789", 1234567890123456789)
-
-	// microseconds
-	f("-1234567890123456", -1234567890123456000)
-	f("1234567890123456", 1234567890123456000)
-	f("1234567890123456.789", 1234567890123456768)
-
-	// milliseconds
-	f("-1234567890123", -1234567890123000000)
-	f("1234567890123", 1234567890123000000)
-	f("1234567890123.456", 1234567890123456000)
-
-	// seconds
-	f("-1234567890", -1234567890000000000)
-	f("1234567890", 1234567890000000000)
-	f("-1234567890.123456", -1234567890123456000)
-}
-
-func TestParseUnixTimestamp_Failure(t *testing.T) {
-	f := func(s string) {
-		t.Helper()
-
-		_, err := ParseUnixTimestamp(s)
-		if err == nil {
-			t.Fatalf("expecting non-nil error in ParseUnixTimestamp(%q)", s)
-		}
-	}
-
-	// non-numeric timestamp
-	f("")
-	f("foobar")
-	f("foo.bar")
-
-	// too big timestamp
-	f("12345678901234567890")
-	f("-12345678901234567890")
-	f("12345678901234567890.235424")
-	f("-12345678901234567890.235424")
-}
-
-func TestExtractTimestampFromFields_Success(t *testing.T) {
+func TestExtractTimestampRFC3339NanoFromFields_Success(t *testing.T) {
 	f := func(timeField string, fields []logstorage.Field, nsecsExpected int64) {
 		t.Helper()
 
-		nsecs, err := ExtractTimestampFromFields(timeField, fields)
+		nsecs, err := ExtractTimestampRFC3339NanoFromFields(timeField, fields)
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
@@ -108,18 +51,6 @@ func TestExtractTimestampFromFields_Success(t *testing.T) {
 		{Name: "foo", Value: "bar"},
 	}, 1718773640123456789)
 
-	// Unix timestamp in nanoseconds
-	f("time", []logstorage.Field{
-		{Name: "foo", Value: "bar"},
-		{Name: "time", Value: "1718773640123456789"},
-	}, 1718773640123456789)
-
-	// Unix timestamp in microseconds
-	f("time", []logstorage.Field{
-		{Name: "foo", Value: "bar"},
-		{Name: "time", Value: "1718773640123456"},
-	}, 1718773640123456000)
-
 	// Unix timestamp in milliseconds
 	f("time", []logstorage.Field{
 		{Name: "foo", Value: "bar"},
@@ -133,14 +64,14 @@ func TestExtractTimestampFromFields_Success(t *testing.T) {
 	}, 1718773640000000000)
 }
 
-func TestExtractTimestampFromFields_Error(t *testing.T) {
+func TestExtractTimestampRFC3339NanoFromFields_Error(t *testing.T) {
 	f := func(s string) {
 		t.Helper()
 
 		fields := []logstorage.Field{
 			{Name: "time", Value: s},
 		}
-		nsecs, err := ExtractTimestampFromFields("time", fields)
+		nsecs, err := ExtractTimestampRFC3339NanoFromFields("time", fields)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -149,7 +80,6 @@ func TestExtractTimestampFromFields_Error(t *testing.T) {
 		}
 	}
 
-	// invalid time
 	f("foobar")
 
 	// incomplete time

app/vlinsert/journald/journald.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"flag"
 	"fmt"
+	"io"
 	"net/http"
 	"regexp"
 	"slices"
@@ -15,30 +16,32 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/writeconcurrencylimiter"
 	"github.com/VictoriaMetrics/metrics"
 )
 
-// See https://github.com/systemd/systemd/blob/main/src/libsystemd/sd-journal/journal-file.c#L1703
-const journaldEntryMaxNameLen = 64
-
-var allowedJournaldEntryNameChars = regexp.MustCompile(`^[A-Z_][A-Z0-9_]*`)
+const (
+	journaldEntryMaxNameLen = 64
+)
 
 var (
-	journaldStreamFields = flagutil.NewArrayString("journald.streamFields", "Comma-separated list of fields to use as log stream fields for logs ingested over journald protocol. "+
-		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#stream-fields")
-	journaldIgnoreFields = flagutil.NewArrayString("journald.ignoreFields", "Comma-separated list of fields to ignore for logs ingested over journald protocol. "+
-		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#dropping-fields")
-	journaldTimeField = flag.String("journald.timeField", "__REALTIME_TIMESTAMP", "Field to use as a log timestamp for logs ingested via journald protocol. "+
-		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#time-field")
-	journaldTenantID = flag.String("journald.tenantID", "0:0", "TenantID for logs ingested via the Journald endpoint. "+
-		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/#multitenancy")
-	journaldIncludeEntryMetadata = flag.Bool("journald.includeEntryMetadata", false, "Include journal entry fields, which with double underscores.")
+	bodyBufferPool                bytesutil.ByteBufferPool
+	allowedJournaldEntryNameChars = regexp.MustCompile(`^[A-Z_][A-Z0-9_]+`)
+)
 
-	maxRequestSize = flagutil.NewBytes("journald.maxRequestSize", 64*1024*1024, "The maximum size in bytes of a single journald request")
+var (
+	journaldStreamFields = flagutil.NewArrayString("journald.streamFields", "Journal fields to be used as stream fields. "+
+		"See the list of allowed fields at https://www.freedesktop.org/software/systemd/man/latest/systemd.journal-fields.html.")
+	journaldIgnoreFields = flagutil.NewArrayString("journald.ignoreFields", "Journal fields to ignore. "+
+		"See the list of allowed fields at https://www.freedesktop.org/software/systemd/man/latest/systemd.journal-fields.html.")
+	journaldTimeField = flag.String("journald.timeField", "__REALTIME_TIMESTAMP", "Journal field to be used as time field. "+
+		"See the list of allowed fields at https://www.freedesktop.org/software/systemd/man/latest/systemd.journal-fields.html.")
+	journaldTenantID             = flag.String("journald.tenantID", "0:0", "TenantID for logs ingested via the Journald endpoint.")
+	journaldIncludeEntryMetadata = flag.Bool("journald.includeEntryMetadata", false, "Include journal entry fields, which with double underscores.")
 )
 
 func getCommonParams(r *http.Request) (*insertutils.CommonParams, error) {
@@ -86,36 +89,45 @@ func handleJournald(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsJournaldTotal.Inc()
 
-	cp, err := getCommonParams(r)
-	if err != nil {
-		errorsTotal.Inc()
-		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
-		return
-	}
-
 	if err := vlstorage.CanWriteData(); err != nil {
-		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 
-	encoding := r.Header.Get("Content-Encoding")
-	err = common.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.NewLogMessageProcessor("journald")
-		err := parseJournaldRequest(data, lmp, cp)
-		lmp.MustClose()
-		return err
-	})
+	reader := r.Body
+	var err error
+
+	wcr := writeconcurrencylimiter.GetReader(reader)
+	data, err := io.ReadAll(wcr)
 	if err != nil {
-		errorsTotal.Inc()
-		httpserver.Errorf(w, r, "cannot read journald protocol data: %s", err)
+		httpserver.Errorf(w, r, "cannot read request body: %s", err)
+		return
+	}
+	writeconcurrencylimiter.PutReader(wcr)
+	bb := bodyBufferPool.Get()
+	defer bodyBufferPool.Put(bb)
+	if r.Header.Get("Content-Encoding") == "zstd" {
+		bb.B, err = zstd.Decompress(bb.B[:0], data)
+		if err != nil {
+			httpserver.Errorf(w, r, "cannot decompress zstd-encoded request with length %d: %s", len(data), err)
+			return
+		}
+		data = bb.B
+	}
+	cp, err := getCommonParams(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return
 	}
 
-	// systemd starting release v258 will support compression, which starts working after negotiation: it expects supported compression
-	// algorithms list in Accept-Encoding response header in a format "<algorithm_1>[:<priority_1>][;<algorithm_2>:<priority_2>]"
-	// See https://github.com/systemd/systemd/pull/34822
-	w.Header().Set("Accept-Encoding", "zstd")
+	lmp := cp.NewLogMessageProcessor("journald")
+	err = parseJournaldRequest(data, lmp, cp)
+	lmp.MustClose()
+	if err != nil {
+		errorsTotal.Inc()
+		httpserver.Errorf(w, r, "cannot parse Journald protobuf request: %s", err)
+		return
+	}
 
 	// update requestJournaldDuration only for successfully parsed requests
 	// There is no need in updating requestJournaldDuration for request errors,
@@ -181,7 +193,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 size
+			// skip binary data sise
 			data = data[idx:]
 			if size == 0 {
 				return fmt.Errorf("unexpected zero binary data size decoded %d", size)
@@ -201,6 +213,7 @@ func parseJournaldRequest(data []byte, lmp insertutils.LogMessageProcessor, cp *
 			}
 			data = data[1:]
 		}
+		// https://github.com/systemd/systemd/blob/main/src/libsystemd/sd-journal/journal-file.c#L1703
 		if len(name) > journaldEntryMaxNameLen {
 			return fmt.Errorf("journald entry name should not exceed %d symbols, got: %q", journaldEntryMaxNameLen, name)
 		}

app/vlinsert/journald/journald_test.go

@@ -35,9 +35,9 @@ func TestPushJournaldOk(t *testing.T) {
 	)
 
 	// Parse binary data
-	f("__CURSOR=s=e0afe8412a6a49d2bfcf66aa7927b588;i=1f06;b=f778b6e2f7584a77b991a2366612a7b5;m=300bdfd420;t=62526e1182354;x=930dc44b370963b7\nE=JobStateChanged\n__REALTIME_TIMESTAMP=1729698775704404\n__MONOTONIC_TIMESTAMP=206357648416\n__SEQNUM=7942\n__SEQNUM_ID=e0afe8412a6a49d2bfcf66aa7927b588\n_BOOT_ID=f778b6e2f7584a77b991a2366612a7b5\n_UID=0\n_GID=0\n_MACHINE_ID=a4a970370c30a925df02a13c67167847\n_HOSTNAME=ecd5e4555787\n_RUNTIME_SCOPE=system\n_TRANSPORT=journal\n_CAP_EFFECTIVE=1ffffffffff\n_SYSTEMD_CGROUP=/init.scope\n_SYSTEMD_UNIT=init.scope\n_SYSTEMD_SLICE=-.slice\nCODE_FILE=<stdin>\nCODE_LINE=1\nCODE_FUNC=<module>\nSYSLOG_IDENTIFIER=python3\n_COMM=python3\n_EXE=/usr/bin/python3.12\n_CMDLINE=python3\nMESSAGE\n\x13\x00\x00\x00\x00\x00\x00\x00foo\nbar\n\n\nasda\nasda\n_PID=2763\n_SOURCE_REALTIME_TIMESTAMP=1729698775704375\n\n",
+	f("__CURSOR=s=e0afe8412a6a49d2bfcf66aa7927b588;i=1f06;b=f778b6e2f7584a77b991a2366612a7b5;m=300bdfd420;t=62526e1182354;x=930dc44b370963b7\n__REALTIME_TIMESTAMP=1729698775704404\n__MONOTONIC_TIMESTAMP=206357648416\n__SEQNUM=7942\n__SEQNUM_ID=e0afe8412a6a49d2bfcf66aa7927b588\n_BOOT_ID=f778b6e2f7584a77b991a2366612a7b5\n_UID=0\n_GID=0\n_MACHINE_ID=a4a970370c30a925df02a13c67167847\n_HOSTNAME=ecd5e4555787\n_RUNTIME_SCOPE=system\n_TRANSPORT=journal\n_CAP_EFFECTIVE=1ffffffffff\n_SYSTEMD_CGROUP=/init.scope\n_SYSTEMD_UNIT=init.scope\n_SYSTEMD_SLICE=-.slice\nCODE_FILE=<stdin>\nCODE_LINE=1\nCODE_FUNC=<module>\nSYSLOG_IDENTIFIER=python3\n_COMM=python3\n_EXE=/usr/bin/python3.12\n_CMDLINE=python3\nMESSAGE\n\x13\x00\x00\x00\x00\x00\x00\x00foo\nbar\n\n\nasda\nasda\n_PID=2763\n_SOURCE_REALTIME_TIMESTAMP=1729698775704375\n\n",
 		[]int64{1729698775704404000},
-		"{\"E\":\"JobStateChanged\",\"_BOOT_ID\":\"f778b6e2f7584a77b991a2366612a7b5\",\"_UID\":\"0\",\"_GID\":\"0\",\"_MACHINE_ID\":\"a4a970370c30a925df02a13c67167847\",\"_HOSTNAME\":\"ecd5e4555787\",\"_RUNTIME_SCOPE\":\"system\",\"_TRANSPORT\":\"journal\",\"_CAP_EFFECTIVE\":\"1ffffffffff\",\"_SYSTEMD_CGROUP\":\"/init.scope\",\"_SYSTEMD_UNIT\":\"init.scope\",\"_SYSTEMD_SLICE\":\"-.slice\",\"CODE_FILE\":\"\\u003cstdin>\",\"CODE_LINE\":\"1\",\"CODE_FUNC\":\"\\u003cmodule>\",\"SYSLOG_IDENTIFIER\":\"python3\",\"_COMM\":\"python3\",\"_EXE\":\"/usr/bin/python3.12\",\"_CMDLINE\":\"python3\",\"_msg\":\"foo\\nbar\\n\\n\\nasda\\nasda\",\"_PID\":\"2763\",\"_SOURCE_REALTIME_TIMESTAMP\":\"1729698775704375\"}",
+		"{\"_BOOT_ID\":\"f778b6e2f7584a77b991a2366612a7b5\",\"_UID\":\"0\",\"_GID\":\"0\",\"_MACHINE_ID\":\"a4a970370c30a925df02a13c67167847\",\"_HOSTNAME\":\"ecd5e4555787\",\"_RUNTIME_SCOPE\":\"system\",\"_TRANSPORT\":\"journal\",\"_CAP_EFFECTIVE\":\"1ffffffffff\",\"_SYSTEMD_CGROUP\":\"/init.scope\",\"_SYSTEMD_UNIT\":\"init.scope\",\"_SYSTEMD_SLICE\":\"-.slice\",\"CODE_FILE\":\"\\u003cstdin>\",\"CODE_LINE\":\"1\",\"CODE_FUNC\":\"\\u003cmodule>\",\"SYSLOG_IDENTIFIER\":\"python3\",\"_COMM\":\"python3\",\"_EXE\":\"/usr/bin/python3.12\",\"_CMDLINE\":\"python3\",\"_msg\":\"foo\\nbar\\n\\n\\nasda\\nasda\",\"_PID\":\"2763\",\"_SOURCE_REALTIME_TIMESTAMP\":\"1729698775704375\"}",
 	)
 }
 

app/vlinsert/jsonline/jsonline.go

@@ -38,23 +38,33 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	encoding := r.Header.Get("Content-Encoding")
-	reader, err := common.GetUncompressedReader(r.Body, encoding)
-	if err != nil {
-		logger.Errorf("cannot decode jsonline request: %s", err)
-		return
+	reader := r.Body
+	if r.Header.Get("Content-Encoding") == "gzip" {
+		zr, err := common.GetGzipReader(reader)
+		if err != nil {
+			logger.Errorf("cannot read gzipped jsonline request: %s", err)
+			return
+		}
+		defer common.PutGzipReader(zr)
+		reader = zr
 	}
-	defer common.PutUncompressedReader(reader)
 
 	lmp := cp.NewLogMessageProcessor("jsonline")
 	streamName := fmt.Sprintf("remoteAddr=%s, requestURI=%q", httpserver.GetQuotedRemoteAddr(r), r.RequestURI)
-	processStreamInternal(streamName, reader, cp.TimeField, cp.MsgFields, lmp)
+	err = processStreamInternal(streamName, reader, cp.TimeField, cp.MsgFields, lmp)
 	lmp.MustClose()
 
-	requestDuration.UpdateDuration(startTime)
+	if err != nil {
+		logger.Errorf("jsonline: %s", err)
+	} else {
+		// update requestDuration only for successfully parsed requests.
+		// There is no need in updating requestDuration for request errors,
+		// since their timings are usually much smaller than the timing for successful request parsing.
+		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 insertutils.LogMessageProcessor) error {
 	wcr := writeconcurrencylimiter.GetReader(r)
 	defer writeconcurrencylimiter.PutReader(wcr)
 
@@ -66,10 +76,10 @@ func processStreamInternal(streamName string, r io.Reader, timeField string, msg
 		wcr.DecConcurrency()
 		if err != nil {
 			errorsTotal.Inc()
-			logger.Warnf("jsonline: cannot read line #%d in /jsonline request: %s", n, err)
+			return fmt.Errorf("cannot read line #%d in /jsonline request: %s", n, err)
 		}
 		if !ok {
-			return
+			return nil
 		}
 		n++
 	}
@@ -86,17 +96,16 @@ func readLine(lr *insertutils.LineReader, timeField string, msgFields []string,
 	}
 
 	p := logstorage.GetJSONParser()
-	defer logstorage.PutJSONParser(p)
-
 	if err := p.ParseLogMessage(line); err != nil {
-		return true, fmt.Errorf("cannot parse json-encoded line: %w; line contents: %q", err, line)
+		return false, fmt.Errorf("cannot parse json-encoded log entry: %w", err)
 	}
-	ts, err := insertutils.ExtractTimestampFromFields(timeField, p.Fields)
+	ts, err := insertutils.ExtractTimestampRFC3339NanoFromFields(timeField, p.Fields)
 	if err != nil {
-		return true, fmt.Errorf("cannot get timestamp from json-encoded line: %w; line contents: %q", err, line)
+		return false, fmt.Errorf("cannot get timestamp: %w", err)
 	}
 	logstorage.RenameField(p.Fields, msgFields, "_msg")
 	lmp.AddRow(ts, p.Fields, nil)
+	logstorage.PutJSONParser(p)
 
 	return true, nil
 }

app/vlinsert/jsonline/jsonline_test.go

@@ -7,14 +7,16 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
 )
 
-func TestProcessStreamInternal(t *testing.T) {
+func TestProcessStreamInternal_Success(t *testing.T) {
 	f := func(data, timeField, msgField string, timestampsExpected []int64, resultExpected string) {
 		t.Helper()
 
 		msgFields := []string{msgField}
 		tlp := &insertutils.TestLogMessageProcessor{}
 		r := bytes.NewBufferString(data)
-		processStreamInternal("test", r, timeField, msgFields, tlp)
+		if err := processStreamInternal("test", r, timeField, msgFields, tlp); err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
 
 		if err := tlp.Verify(timestampsExpected, resultExpected); err != nil {
 			t.Fatal(err)
@@ -43,37 +45,22 @@ func TestProcessStreamInternal(t *testing.T) {
 	resultExpected = `{"log.offset":"71770","log.file.path":"/var/log/auth.log","message":"foobar"}
 {"message":"baz"}`
 	f(data, timeField, msgField, timestampsExpected, resultExpected)
+}
+
+func TestProcessStreamInternal_Failure(t *testing.T) {
+	f := func(data string) {
+		t.Helper()
+
+		tlp := &insertutils.TestLogMessageProcessor{}
+		r := bytes.NewBufferString(data)
+		if err := processStreamInternal("test", r, "time", nil, tlp); err == nil {
+			t.Fatalf("expecting non-nil error")
+		}
+	}
 
 	// invalid json
-	data = "foobar"
-	timeField = "@timestamp"
-	msgField = "aaa"
-	timestampsExpected = nil
-	resultExpected = ``
-	f(data, timeField, msgField, timestampsExpected, resultExpected)
+	f("foobar")
 
 	// invalid timestamp field
-	data = `{"time":"foobar"}`
-	timeField = "time"
-	msgField = "abc"
-	timestampsExpected = nil
-	resultExpected = ``
-	f(data, timeField, msgField, timestampsExpected, resultExpected)
-
-	// invalid lines among valid lines
-	data = `
-dsfodmasd
-
-{"time":"2023-06-06T04:48:11.735Z","log":{"offset":71770,"file":{"path":"/var/log/auth.log"}},"message":"foobar"}
-invalid line
-{"time":"2023-06-06T04:48:12.735+01:00","message":"baz"}
-asbsdf
-
-`
-	timeField = "time"
-	msgField = "message"
-	timestampsExpected = []int64{1686026891735000000, 1686023292735000000}
-	resultExpected = `{"log.offset":"71770","log.file.path":"/var/log/auth.log","_msg":"foobar"}
-{"_msg":"baz"}`
-	f(data, timeField, msgField, timestampsExpected, resultExpected)
+	f(`{"time":"foobar"}`)
 }

app/vlinsert/loki/loki.go

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

app/vlinsert/loki/loki_json.go

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

app/vlinsert/loki/loki_json_test.go

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

app/vlinsert/loki/loki_json_timing_test.go

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

app/vlinsert/loki/loki_protobuf.go

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

app/vlinsert/loki/loki_protobuf_test.go

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

app/vlinsert/loki/loki_protobuf_timing_test.go

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

app/vlinsert/main.go

@@ -46,9 +46,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 	switch {
-	case strings.HasPrefix(path, "/elasticsearch"):
-		// some clients may omit trailing slash
-		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8353
+	case strings.HasPrefix(path, "/elasticsearch/"):
 		path = strings.TrimPrefix(path, "/elasticsearch")
 		return elasticsearch.RequestHandler(path, w, r)
 	case strings.HasPrefix(path, "/loki/"):

app/vlinsert/opentelemetry/opentelemetry.go

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

app/vlinsert/opentelemetry/opentelemetry_test.go

@@ -24,7 +24,6 @@ func TestPushProtoOk(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
-
 	// single line without resource attributes
 	f([]pb.ResourceLogs{
 		{
@@ -40,7 +39,6 @@ func TestPushProtoOk(t *testing.T) {
 		[]int64{1234},
 		`{"_msg":"log-line-message","severity":"Trace"}`,
 	)
-
 	// multi-line with resource attributes
 	f([]pb.ResourceLogs{
 		{
@@ -68,9 +66,9 @@ func TestPushProtoOk(t *testing.T) {
 		},
 	},
 		[]int64{1234, 1235, 1236},
-		`{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message","severity":"Trace"}
-{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Unspecified"}
-{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Unspecified"}`,
+		`{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message","severity":"Trace"}
+{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message-msg-2","severity":"Unspecified"}
+{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message-msg-2","severity":"Unspecified"}`,
 	)
 
 	// multi-scope with resource attributes and multi-line
@@ -108,21 +106,19 @@ func TestPushProtoOk(t *testing.T) {
 				{
 					LogRecords: []pb.LogRecord{
 						{TimeUnixNano: 2347, SeverityNumber: 12, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-0")}},
-						{TraceID: "1234", SpanID: "45", ObservedTimeUnixNano: 2348, SeverityNumber: 12, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-1")}},
-						{TraceID: "4bf92f3577b34da6a3ce929d0e0e4736", SpanID: "00f067aa0ba902b7", ObservedTimeUnixNano: 3333, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-2")}},
+						{ObservedTimeUnixNano: 2348, SeverityNumber: 12, Body: pb.AnyValue{StringValue: ptrTo("log-line-resource-scope-1-1-1")}},
 					},
 				},
 			},
 		},
 	},
-		[]int64{1234, 1235, 2345, 2346, 2347, 2348, 3333},
-		`{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message","severity":"Trace"}
-{"logger":"context","instance_id":"10","node_taints":"{\"role\":\"dev\",\"cluster_load_percent\":0.55}","_msg":"log-line-message-msg-2","severity":"Debug"}
+		[]int64{1234, 1235, 2345, 2346, 2347, 2348},
+		`{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message","severity":"Trace"}
+{"logger":"context","instance_id":"10","node_taints":"[{\"Key\":\"role\",\"Value\":{\"StringValue\":\"dev\",\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":null,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}},{\"Key\":\"cluster_load_percent\",\"Value\":{\"StringValue\":null,\"BoolValue\":null,\"IntValue\":null,\"DoubleValue\":0.55,\"ArrayValue\":null,\"KeyValueList\":null,\"BytesValue\":null}}]","_msg":"log-line-message-msg-2","severity":"Debug"}
 {"_msg":"log-line-resource-scope-1-0-0","severity":"Info2"}
 {"_msg":"log-line-resource-scope-1-0-1","severity":"Info2"}
 {"_msg":"log-line-resource-scope-1-1-0","severity":"Info4"}
-{"_msg":"log-line-resource-scope-1-1-1","trace_id":"1234","span_id":"45","severity":"Info4"}
-{"_msg":"log-line-resource-scope-1-1-2","trace_id":"4bf92f3577b34da6a3ce929d0e0e4736","span_id":"00f067aa0ba902b7","severity":"Unspecified"}`,
+{"_msg":"log-line-resource-scope-1-1-1","severity":"Info4"}`,
 	)
 }
 

app/vlinsert/syslog/syslog.go

@@ -16,6 +16,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/klauspost/compress/gzip"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlinsert/insertutils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vlstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@@ -272,14 +274,14 @@ func runTCPListener(addr string, argIdx int) {
 
 func checkCompressMethod(compressMethod, addr, protocol string) {
 	switch compressMethod {
-	case "", "none", "zstd", "gzip", "deflate":
+	case "", "none", "gzip", "deflate":
 		return
 	default:
-		logger.Fatalf("unsupported -syslog.compressMethod.%s=%q for -syslog.listenAddr.%s=%q; supported values: 'none', 'zstd', 'gzip', 'deflate'", protocol, compressMethod, protocol, addr)
+		logger.Fatalf("unsupported -syslog.compressMethod.%s=%q for -syslog.listenAddr.%s=%q; supported values: 'none', 'gzip', 'deflate'", protocol, compressMethod, protocol, addr)
 	}
 }
 
-func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
+func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
 	gomaxprocs := cgroup.AvailableCPUs()
 	var wg sync.WaitGroup
 	localAddr := ln.LocalAddr()
@@ -312,7 +314,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string,
 				}
 				bb.B = bb.B[:n]
 				udpRequestsTotal.Inc()
-				if err := processStream("udp", bb.NewReader(), encoding, useLocalTimestamp, cp); err != nil {
+				if err := processStream("udp", bb.NewReader(), compressMethod, useLocalTimestamp, cp); err != nil {
 					logger.Errorf("syslog: cannot process UDP data from %s at %s: %s", remoteAddr, localAddr, err)
 				}
 			}
@@ -321,7 +323,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, encoding string,
 	wg.Wait()
 }
 
-func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
+func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
 	var cm ingestserver.ConnsMap
 	cm.Init("syslog")
 
@@ -352,7 +354,7 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, encoding string, us
 		wg.Add(1)
 		go func() {
 			cp := insertutils.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
-			if err := processStream("tcp", c, encoding, useLocalTimestamp, cp); err != nil {
+			if err := processStream("tcp", c, compressMethod, useLocalTimestamp, cp); err != nil {
 				logger.Errorf("syslog: cannot process TCP data at %q: %s", addr, err)
 			}
 
@@ -367,26 +369,49 @@ 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, compressMethod string, useLocalTimestamp bool, cp *insertutils.CommonParams) error {
 	if err := vlstorage.CanWriteData(); err != nil {
 		return err
 	}
 
 	lmp := cp.NewLogMessageProcessor("syslog_" + protocol)
-	err := processStreamInternal(r, encoding, useLocalTimestamp, lmp)
+	err := processStreamInternal(r, compressMethod, 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)
-	if err != nil {
-		return fmt.Errorf("cannot decode syslog data: %w", err)
+func processStreamInternal(r io.Reader, compressMethod string, useLocalTimestamp bool, lmp insertutils.LogMessageProcessor) error {
+	switch compressMethod {
+	case "", "none":
+	case "gzip":
+		zr, err := common.GetGzipReader(r)
+		if err != nil {
+			return fmt.Errorf("cannot read gzipped data: %w", err)
+		}
+		r = zr
+	case "deflate":
+		zr, err := common.GetZlibReader(r)
+		if err != nil {
+			return fmt.Errorf("cannot read deflated data: %w", err)
+		}
+		r = zr
+	default:
+		logger.Panicf("BUG: unsupported compressMethod=%q; supported values: none, gzip, deflate", compressMethod)
 	}
-	defer common.PutUncompressedReader(reader)
 
-	return processUncompressedStream(reader, useLocalTimestamp, lmp)
+	err := processUncompressedStream(r, useLocalTimestamp, lmp)
+
+	switch compressMethod {
+	case "gzip":
+		zr := r.(*gzip.Reader)
+		common.PutGzipReader(zr)
+	case "deflate":
+		zr := r.(io.ReadCloser)
+		common.PutZlibReader(zr)
+	}
+
+	return err
 }
 
 func processUncompressedStream(r io.Reader, useLocalTimestamp bool, lmp insertutils.LogMessageProcessor) error {
@@ -535,7 +560,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 := insertutils.ExtractTimestampRFC3339NanoFromFields("timestamp", p.Fields)
 		if err != nil {
 			return fmt.Errorf("cannot get timestamp from syslog line %q: %w", line, err)
 		}

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 object value: %v; want string", t)
+			return nil, fmt.Errorf("unexpected token read for oject value: %v; want string", t)
 		}
 
 		fields = append(fields, logstorage.Field{
@@ -176,7 +176,7 @@ func writeCompactObject(w io.Writer, fields []logstorage.Field) error {
 		_, err := fmt.Fprintf(w, "%s\n", fields[0].Value)
 		return err
 	}
-	if len(fields) == 2 && (fields[0].Name == "_time" || fields[1].Name == "_time") {
+	if len(fields) == 2 && fields[0].Name == "_time" || fields[1].Name == "_time" {
 		// Write _time\tfieldValue as is
 		if fields[0].Name == "_time" {
 			_, err := fmt.Fprintf(w, "%s\t%s\n", fields[0].Value, fields[1].Value)

app/vlogscli/main.go

@@ -270,7 +270,7 @@ func printCommandsHelp(w io.Writer) {
 \h - show this help
 \s - singleline json output mode
 \m - multiline json output mode
-\c - compact output mode
+\c - compact output
 \logfmt - logfmt output mode
 \wrap_long_lines - toggles wrapping long lines
 \tail <query> - live tail <query> results

app/vlogsgenerator/main.go

@@ -18,7 +18,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/envflag"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
 var (
@@ -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 constant values to generate per each log entry; "+
+	constFieldsPerLog = flag.Int("constFieldsPerLog", 3, "The number of fields with constaint 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")
@@ -45,8 +45,6 @@ var (
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
 	u64FieldsPerLog = flag.Int("u64FieldsPerLog", 1, "The number of fields with uint64 values to generate per each log entry; "+
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
-	i64FieldsPerLog = flag.Int("i64FieldsPerLog", 1, "The number of fields with int64 values to generate per each log entry; "+
-		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
 	floatFieldsPerLog = flag.Int("floatFieldsPerLog", 1, "The number of fields with float64 values to generate per each log entry; "+
 		"see https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model")
 	ipFieldsPerLog = flag.Int("ipFieldsPerLog", 1, "The number of fields with IPv4 values to generate per each log entry; "+
@@ -177,10 +175,7 @@ func generateAndPushLogs(cfg *workerConfig, workerID int) {
 	sw := &statWriter{
 		w: pw,
 	}
-
-	// The 1MB write buffer increases data ingestion performance by reducing the number of send() syscalls
-	bw := bufio.NewWriterSize(sw, 1024*1024)
-
+	bw := bufio.NewWriter(sw)
 	doneCh := make(chan struct{})
 	go func() {
 		generateLogs(bw, workerID, cfg.activeStreams, cfg.totalStreams)
@@ -259,9 +254,6 @@ func generateLogsAtTimestamp(bw *bufio.Writer, workerID int, ts int64, firstStre
 		for j := 0; j < *u64FieldsPerLog; j++ {
 			fmt.Fprintf(bw, `,"u64_%d":"%d"`, j, rand.Uint64())
 		}
-		for j := 0; j < *i64FieldsPerLog; j++ {
-			fmt.Fprintf(bw, `,"i64_%d":"%d"`, j, int64(rand.Uint64()))
-		}
 		for j := 0; j < *floatFieldsPerLog; j++ {
 			fmt.Fprintf(bw, `,"float_%d":"%v"`, j, math.Round(10_000*rand.Float64())/1000)
 		}
@@ -309,7 +301,7 @@ type timeFlag struct {
 }
 
 func (tf *timeFlag) Set(s string) error {
-	msec, err := timeutil.ParseTimeMsec(s)
+	msec, err := promutils.ParseTimeMsec(s)
 	if err != nil {
 		return fmt.Errorf("cannot parse time from %q: %w", s, err)
 	}

app/vlselect/logsql/logsql.go

@@ -22,7 +22,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
 // ProcessFacetsRequest handles /select/logsql/facets request.
@@ -116,7 +116,7 @@ func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	if stepStr == "" {
 		stepStr = "1d"
 	}
-	step, err := timeutil.ParseDuration(stepStr)
+	step, err := promutils.ParseDuration(stepStr)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse 'step' arg: %s", err)
 		return
@@ -131,7 +131,7 @@ func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	if offsetStr == "" {
 		offsetStr = "0s"
 	}
-	offset, err := timeutil.ParseDuration(offsetStr)
+	offset, err := promutils.ParseDuration(offsetStr)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse 'offset' arg: %s", err)
 		return
@@ -665,7 +665,7 @@ func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r
 	if stepStr == "" {
 		stepStr = "1d"
 	}
-	step, err := timeutil.ParseDuration(stepStr)
+	step, err := promutils.ParseDuration(stepStr)
 	if err != nil {
 		err = fmt.Errorf("cannot parse 'step' arg: %s", err)
 		httpserver.SendPrometheusError(w, r, err)
@@ -694,16 +694,13 @@ func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r
 			clonedColumnNames[i] = strings.Clone(c.Name)
 		}
 		for i := range timestamps {
-			// Do not move q.GetTimestamp() outside writeBlock, since ts
-			// must be initialized to query timestamp for every processed log row.
-			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8312
-			ts := q.GetTimestamp()
+			timestamp := q.GetTimestamp()
 			labels := make([]logstorage.Field, 0, len(byFields))
 			for j, c := range columns {
 				if c.Name == "_time" {
 					nsec, ok := logstorage.TryParseTimestampRFC3339Nano(c.Values[i])
 					if ok {
-						ts = nsec
+						timestamp = nsec
 						continue
 					}
 				}
@@ -724,7 +721,7 @@ func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r
 					dst = logstorage.MarshalFieldsToJSON(dst, labels)
 					key := string(dst)
 					p := statsPoint{
-						Timestamp: ts,
+						Timestamp: timestamp,
 						Value:     strings.Clone(c.Values[i]),
 					}
 
@@ -1122,7 +1119,7 @@ func getTimeNsec(r *http.Request, argName string) (int64, bool, error) {
 		return 0, false, nil
 	}
 	currentTimestamp := time.Now().UnixNano()
-	nsecs, err := timeutil.ParseTimeAt(s, currentTimestamp)
+	nsecs, err := promutils.ParseTimeAt(s, currentTimestamp)
 	if err != nil {
 		return 0, false, fmt.Errorf("cannot parse %s=%s: %w", argName, s, err)
 	}

app/vlselect/logsql/logsql_test.go

@@ -28,7 +28,7 @@ func TestParseExtraFilters_Success(t *testing.T) {
 	// LogsQL filter
 	f(`foobar`, `foobar`)
 	f(`foo:bar`, `foo:bar`)
-	f(`foo:(bar or baz) error _time:5m {"foo"=bar,baz="z"}`, `{foo="bar",baz="z"} (foo:bar or foo:baz) error _time:5m`)
+	f(`foo:(bar or baz) error _time:5m {"foo"=bar,baz="z"}`, `(foo:bar or foo:baz) error _time:5m {foo="bar",baz="z"}`)
 }
 
 func TestParseExtraFilters_Failure(t *testing.T) {
@@ -77,7 +77,7 @@ func TestParseExtraStreamFilters_Success(t *testing.T) {
 	// LogsQL filter
 	f(`foobar`, `foobar`)
 	f(`foo:bar`, `foo:bar`)
-	f(`foo:(bar or baz) error _time:5m {"foo"=bar,baz="z"}`, `{foo="bar",baz="z"} (foo:bar or foo:baz) error _time:5m`)
+	f(`foo:(bar or baz) error _time:5m {"foo"=bar,baz="z"}`, `(foo:bar or foo:baz) error _time:5m {foo="bar",baz="z"}`)
 }
 
 func TestParseExtraStreamFilters_Failure(t *testing.T) {

app/vlselect/main.go

@@ -176,62 +176,50 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 
 func processSelectRequest(ctx context.Context, w http.ResponseWriter, r *http.Request, path string) bool {
 	httpserver.EnableCORS(w, r)
-	startTime := time.Now()
 	switch path {
 	case "/select/logsql/facets":
 		logsqlFacetsRequests.Inc()
 		logsql.ProcessFacetsRequest(ctx, w, r)
-		logsqlFacetsDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/field_names":
 		logsqlFieldNamesRequests.Inc()
 		logsql.ProcessFieldNamesRequest(ctx, w, r)
-		logsqlFieldNamesDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/field_values":
 		logsqlFieldValuesRequests.Inc()
 		logsql.ProcessFieldValuesRequest(ctx, w, r)
-		logsqlFieldValuesDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/hits":
 		logsqlHitsRequests.Inc()
 		logsql.ProcessHitsRequest(ctx, w, r)
-		logsqlHitsDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/query":
 		logsqlQueryRequests.Inc()
 		logsql.ProcessQueryRequest(ctx, w, r)
-		logsqlQueryDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/stats_query":
 		logsqlStatsQueryRequests.Inc()
 		logsql.ProcessStatsQueryRequest(ctx, w, r)
-		logsqlStatsQueryDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/stats_query_range":
 		logsqlStatsQueryRangeRequests.Inc()
 		logsql.ProcessStatsQueryRangeRequest(ctx, w, r)
-		logsqlStatsQueryRangeDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/stream_field_names":
 		logsqlStreamFieldNamesRequests.Inc()
 		logsql.ProcessStreamFieldNamesRequest(ctx, w, r)
-		logsqlStreamFieldNamesDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/stream_field_values":
 		logsqlStreamFieldValuesRequests.Inc()
 		logsql.ProcessStreamFieldValuesRequest(ctx, w, r)
-		logsqlStreamFieldValuesDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/stream_ids":
 		logsqlStreamIDsRequests.Inc()
 		logsql.ProcessStreamIDsRequest(ctx, w, r)
-		logsqlStreamIDsDuration.UpdateDuration(startTime)
 		return true
 	case "/select/logsql/streams":
 		logsqlStreamsRequests.Inc()
 		logsql.ProcessStreamsRequest(ctx, w, r)
-		logsqlStreamsDuration.UpdateDuration(startTime)
 		return true
 	default:
 		return false
@@ -252,39 +240,16 @@ func getMaxQueryDuration(r *http.Request) time.Duration {
 }
 
 var (
-	logsqlFacetsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/facets"}`)
-	logsqlFacetsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/facets"}`)
-
-	logsqlFieldNamesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_names"}`)
-	logsqlFieldNamesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/field_names"}`)
-
-	logsqlFieldValuesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_values"}`)
-	logsqlFieldValuesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/field_values"}`)
-
-	logsqlHitsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/hits"}`)
-	logsqlHitsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/hits"}`)
-
-	logsqlQueryRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/query"}`)
-	logsqlQueryDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/query"}`)
-
-	logsqlStatsQueryRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stats_query"}`)
-	logsqlStatsQueryDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stats_query"}`)
-
-	logsqlStatsQueryRangeRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stats_query_range"}`)
-	logsqlStatsQueryRangeDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stats_query_range"}`)
-
-	logsqlStreamFieldNamesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_field_names"}`)
-	logsqlStreamFieldNamesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stream_field_names"}`)
-
+	logsqlFacetsRequests            = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/facets"}`)
+	logsqlFieldNamesRequests        = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_names"}`)
+	logsqlFieldValuesRequests       = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/field_values"}`)
+	logsqlHitsRequests              = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/hits"}`)
+	logsqlQueryRequests             = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/query"}`)
+	logsqlStatsQueryRequests        = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stats_query"}`)
+	logsqlStatsQueryRangeRequests   = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stats_query_range"}`)
+	logsqlStreamFieldNamesRequests  = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_field_names"}`)
 	logsqlStreamFieldValuesRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_field_values"}`)
-	logsqlStreamFieldValuesDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stream_field_values"}`)
-
-	logsqlStreamIDsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_ids"}`)
-	logsqlStreamIDsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/stream_ids"}`)
-
-	logsqlStreamsRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/streams"}`)
-	logsqlStreamsDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/select/logsql/streams"}`)
-
-	// no need to track duration for tail requests, as they usually take long time
-	logsqlTailRequests = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/tail"}`)
+	logsqlStreamIDsRequests         = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/stream_ids"}`)
+	logsqlStreamsRequests           = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/streams"}`)
+	logsqlTailRequests              = metrics.NewCounter(`vl_http_requests_total{path="/select/logsql/tail"}`)
 )

app/vlselect/vmui/asset-manifest.json

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

app/vlselect/vmui/assets/vendor-D1GxaB_c.css

@@ -1 +0,0 @@
-.uplot,.uplot *,.uplot *:before,.uplot *:after{box-sizing:border-box}.uplot{font-family:system-ui,-apple-system,Segoe UI,Roboto,Helvetica Neue,Arial,Noto Sans,sans-serif,"Apple Color Emoji","Segoe UI Emoji",Segoe UI Symbol,"Noto Color Emoji";line-height:1.5;width:min-content}.u-title{text-align:center;font-size:18px;font-weight:700}.u-wrap{position:relative;-webkit-user-select:none;user-select:none}.u-over,.u-under{position:absolute}.u-under{overflow:hidden}.uplot canvas{display:block;position:relative;width:100%;height:100%}.u-axis{position:absolute}.u-legend{font-size:14px;margin:auto;text-align:center}.u-inline{display:block}.u-inline *{display:inline-block}.u-inline tr{margin-right:16px}.u-legend th{font-weight:600}.u-legend th>*{vertical-align:middle;display:inline-block}.u-legend .u-marker{width:1em;height:1em;margin-right:4px;background-clip:padding-box!important}.u-inline.u-live th:after{content:":";vertical-align:middle}.u-inline:not(.u-live) .u-value{display:none}.u-series>*{padding:4px}.u-series th{cursor:pointer}.u-legend .u-off>*{opacity:.3}.u-select{background:#00000012;position:absolute;pointer-events:none}.u-cursor-x,.u-cursor-y{position:absolute;left:0;top:0;pointer-events:none;will-change:transform}.u-hz .u-cursor-x,.u-vt .u-cursor-y{height:100%;border-right:1px dashed #607D8B}.u-hz .u-cursor-y,.u-vt .u-cursor-x{width:100%;border-bottom:1px dashed #607D8B}.u-cursor-pt{position:absolute;top:0;left:0;border-radius:50%;border:0 solid;pointer-events:none;will-change:transform;background-clip:padding-box!important}.u-axis.u-off,.u-select.u-off,.u-cursor-x.u-off,.u-cursor-y.u-off,.u-cursor-pt.u-off{display:none}

app/vlselect/vmui/index.html

@@ -1,57 +1 @@
-<!DOCTYPE html>
-<html lang="en">
-<head>
-  <meta charset="utf-8"/>
-  <link rel="icon" href="./favicon.svg" />
-  <link rel="apple-touch-icon" href="./favicon.svg" />
-  <link rel="mask-icon" href="./favicon.svg" color="#000000">
-
-  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=5"/>
-  <meta name="theme-color" content="#000000"/>
-  <meta name="description" content="Explore your log data with VictoriaLogs UI"/>
-  <!--
-    manifest.json provides metadata used when your web app is installed on a
-    user's mobile device or desktop. See https://developers.google.com/web/fundamentals/web-app-manifest/
-  -->
-  <link rel="manifest" href="./manifest.json"/>
-  <!--
-    Notice the use of in the tags above.
-    It will be replaced with the URL of the `public` folder during the build.
-    Only files inside the `public` folder can be referenced from the HTML.
-
-    Unlike "/favicon.ico" or "favicon.ico", "/favicon.ico" will
-    work correctly both with client-side routing and a non-root public URL.
-    Learn how to configure a non-root public URL by running `npm run build`.
-  -->
-  <title>UI for VictoriaLogs</title>
-
-  <meta name="twitter:card" content="summary">
-  <meta name="twitter:title" content="UI for VictoriaLogs">
-  <meta name="twitter:site" content="@https://victoriametrics.com/products/victorialogs/">
-  <meta name="twitter:description" content="Explore your log data with VictoriaLogs UI">
-  <meta name="twitter:image" content="./preview.jpg">
-
-  <meta property="og:type" content="website">
-  <meta property="og:title" content="UI for VictoriaLogs">
-  <meta property="og:url" content="https://victoriametrics.com/products/victorialogs/">
-  <meta property="og:description" content="Explore your log data with VictoriaLogs UI">
-  <script type="module" crossorigin src="./assets/index-C68hz-qY.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">
-</head>
-<body>
-<noscript>You need to enable JavaScript to run this app.</noscript>
-<div id="root"></div>
-<!--
-  This HTML file is a template.
-  If you open it directly in the browser, you will see an empty page.
-
-  You can add webfonts, meta tags, or analytics to this file.
-  The build step will place the bundled scripts into the <body> tag.
-
-  To begin the development, run `npm start` or `yarn start`.
-  To create a production bundle, use `npm run build` or `yarn build`.
--->
-</body>
-</html>
+<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="./favicon.svg"/><link rel="apple-touch-icon" href="./favicon.svg"/><link rel="mask-icon" href="./favicon.svg" color="#000000"><meta name="viewport" content="width=device-width,initial-scale=1,maximum-scale=5"/><meta name="theme-color" content="#000000"/><meta name="description" content="Explore your log data with VictoriaLogs UI"/><link rel="manifest" href="./manifest.json"/><title>UI for VictoriaLogs</title><meta name="twitter:card" content="summary"><meta name="twitter:title" content="UI for VictoriaLogs"><meta name="twitter:site" content="@https://victoriametrics.com/products/victorialogs/"><meta name="twitter:description" content="Explore your log data with VictoriaLogs UI"><meta name="twitter:image" content="./preview.jpg"><meta property="og:type" content="website"><meta property="og:title" content="UI for VictoriaLogs"><meta property="og:url" content="https://victoriametrics.com/products/victorialogs/"><meta property="og:description" content="Explore your log data with VictoriaLogs UI"><script defer="defer" src="./static/js/main.8ad2bc1f.js"></script><link href="./static/css/main.fa83344e.css" rel="stylesheet"></head><body><noscript>You need to enable JavaScript to run this app.</noscript><div id="root"></div></body></html>

app/vlselect/vmui/static/js/main.8ad2bc1f.js.LICENSE.txt

@@ -0,0 +1,38 @@
+/*!
+	Copyright (c) 2018 Jed Watson.
+	Licensed under the MIT License (MIT), see
+	http://jedwatson.github.io/classnames
+*/
+
+/**
+ * @remix-run/router v1.19.2
+ *
+ * Copyright (c) Remix Software Inc.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE.md file in the root directory of this source tree.
+ *
+ * @license MIT
+ */
+
+/**
+ * React Router DOM v6.26.2
+ *
+ * Copyright (c) Remix Software Inc.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE.md file in the root directory of this source tree.
+ *
+ * @license MIT
+ */
+
+/**
+ * React Router v6.26.2
+ *
+ * Copyright (c) Remix Software Inc.
+ *
+ * This source code is licensed under the MIT license found in the
+ * LICENSE.md file in the root directory of this source tree.
+ *
+ * @license MIT
+ */

app/vlselect/vmui/static/media/MetricsQL.a00044c91d9781cf8557.md

@@ -2349,4 +2349,4 @@ VictoriaMetrics performs the following implicit conversions for incoming queries
   is passed to [rollup function](#rollup-functions), then a [subquery](#subqueries) with `1i` lookbehind window and `1i` step is automatically formed.
   For example, `rate(sum(up))` is automatically converted to `rate((sum(default_rollup(up)))[1i:1i])`.
   This behavior can be disabled or logged via `-search.disableImplicitConversion` and `-search.logImplicitConversion` command-line flags
-  starting from [`v1.102.0-rc2` release](https://docs.victoriametrics.com/changelog/changelog_2024/#v11020-rc2).
+  starting from [`v1.101.0` release](https://docs.victoriametrics.com/changelog/).

app/vmagent/graphite/request_handler.go

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

app/vmagent/influx/request_handler.go

@@ -35,8 +35,8 @@ var (
 // InsertHandlerForReader processes remote write for influx line protocol.
 //
 // See https://github.com/influxdata/telegraf/tree/master/plugins/inputs/socket_listener/
-func InsertHandlerForReader(at *auth.Token, r io.Reader, encoding string) error {
-	return stream.Parse(r, encoding, true, "", "", func(db string, rows []parser.Row) error {
+func InsertHandlerForReader(at *auth.Token, r io.Reader, isGzipped bool) error {
+	return stream.Parse(r, true, isGzipped, "", "", func(db string, rows []parser.Row) error {
 		return insertRows(at, db, rows, nil)
 	})
 }
@@ -49,13 +49,13 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	isStreamMode := req.Header.Get("Stream-Mode") == "1"
 	q := req.URL.Query()
 	precision := q.Get("precision")
 	// Read db tag from https://docs.influxdata.com/influxdb/v1.7/tools/api/#write-http-endpoint
 	db := q.Get("db")
-	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, isStreamMode, isGzipped, precision, db, func(db string, rows []parser.Row) error {
 		return insertRows(at, db, rows, extraLabels)
 	})
 }

app/vmagent/main.go

@@ -148,7 +148,7 @@ func main() {
 	common.StartUnmarshalWorkers()
 	if len(*influxListenAddr) > 0 {
 		influxServer = influxserver.MustStart(*influxListenAddr, *influxUseProxyProtocol, func(r io.Reader) error {
-			return influx.InsertHandlerForReader(nil, r, "")
+			return influx.InsertHandlerForReader(nil, r, false)
 		})
 	}
 	if len(*graphiteListenAddr) > 0 {

app/vmagent/native/request_handler.go

@@ -29,8 +29,8 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, encoding, func(block *stream.Block) error {
+	isGzip := req.Header.Get("Content-Encoding") == "gzip"
+	return stream.Parse(req.Body, isGzip, func(block *stream.Block) error {
 		return insertRows(at, block, extraLabels)
 	})
 }

app/vmagent/newrelic/request_handler.go

@@ -28,8 +28,9 @@ func InsertHandlerForHTTP(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, encoding, func(rows []newrelic.Row) error {
+	ce := req.Header.Get("Content-Encoding")
+	isGzip := ce == "gzip"
+	return stream.Parse(req.Body, isGzip, func(rows []newrelic.Row) error {
 		return insertRows(at, rows, extraLabels)
 	})
 }

app/vmagent/opentelemetry/request_handler.go

@@ -27,7 +27,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	encoding := req.Header.Get("Content-Encoding")
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
 	var processBody func([]byte) ([]byte, error)
 	if req.Header.Get("Content-Type") == "application/json" {
 		if req.Header.Get("X-Amz-Firehose-Protocol-Version") != "" {
@@ -36,7 +36,7 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 			return fmt.Errorf("json encoding isn't supported for opentelemetry format. Use protobuf encoding")
 		}
 	}
-	return stream.ParseStream(req.Body, encoding, processBody, func(tss []prompbmarshal.TimeSeries) error {
+	return stream.ParseStream(req.Body, isGzipped, processBody, func(tss []prompbmarshal.TimeSeries) error {
 		return insertRows(at, tss, extraLabels)
 	})
 }

app/vmagent/prometheusimport/request_handler.go

@@ -31,8 +31,8 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, defaultTimestamp, encoding, true, func(rows []parser.Row) error {
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	return stream.Parse(req.Body, defaultTimestamp, isGzipped, true, func(rows []parser.Row) error {
 		return insertRows(at, rows, extraLabels)
 	}, func(s string) {
 		httpserver.LogError(req, s)

app/vmagent/prometheusimport/request_handler_test.go

@@ -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("unexpected error %s", err)
+		t.Fatalf("unxepected error %s", err)
 	}
 	expectedMsg := "cannot unmarshal Prometheus line"
 	if !strings.Contains(testOutput.String(), expectedMsg) {

app/vmagent/remotewrite/pendingseries.go

@@ -118,7 +118,7 @@ type writeRequest struct {
 }
 
 func (wr *writeRequest) reset() {
-	// Do not reset lastFlushTime, fq, isVMRemoteWrite, significantFigures and roundDigits, since they are reused.
+	// Do not reset lastFlushTime, fq, isVMRemoteWrite, significantFigures and roundDigits, since they are re-used.
 
 	wr.wr.Timeseries = nil
 

app/vmagent/remotewrite/remotewrite_test.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/streamaggr"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -54,7 +53,7 @@ func TestGetLabelsHash_Distribution(t *testing.T) {
 }
 
 func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
-	f := func(streamAggrConfig, relabelConfig string, enableWindows bool, dedupInterval time.Duration, keepInput, dropInput bool, input string) {
+	f := func(streamAggrConfig, relabelConfig string, dedupInterval time.Duration, keepInput, dropInput bool, input string) {
 		t.Helper()
 		perURLRelabel, err := promrelabel.ParseRelabelConfigsData([]byte(relabelConfig))
 		if err != nil {
@@ -78,15 +77,12 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 			rowsDroppedByRelabel:   metrics.GetOrCreateCounter(`bar`),
 		}
 		if dedupInterval > 0 {
-			rwctx.deduplicator = streamaggr.NewDeduplicator(nil, enableWindows, dedupInterval, nil, "dedup-global")
+			rwctx.deduplicator = streamaggr.NewDeduplicator(nil, dedupInterval, nil, "dedup-global")
 		}
 
 		if streamAggrConfig != "" {
 			pushNoop := func(_ []prompbmarshal.TimeSeries) {}
-			opts := streamaggr.Options{
-				EnableWindows: enableWindows,
-			}
-			sas, err := streamaggr.LoadFromData([]byte(streamAggrConfig), pushNoop, &opts, "global")
+			sas, err := streamaggr.LoadFromData([]byte(streamAggrConfig), pushNoop, nil, "global")
 			if err != nil {
 				t.Fatalf("cannot load streamaggr configs: %s", err)
 			}
@@ -95,7 +91,7 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 		}
 
 		offsetMsecs := time.Now().UnixMilli()
-		inputTss := prometheus.MustParsePromMetrics(input, offsetMsecs)
+		inputTss := prompbmarshal.MustParsePromMetrics(input, offsetMsecs)
 		expectedTss := make([]prompbmarshal.TimeSeries, len(inputTss))
 
 		// copy inputTss to make sure it is not mutated during TryPush call
@@ -118,13 +114,13 @@ func TestRemoteWriteContext_TryPush_ImmutableTimeseries(t *testing.T) {
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, false, 0, false, false, `
+`, 0, false, false, `
 metric{env="dev"} 10
 metric{env="bar"} 20
 metric{env="dev"} 15
 metric{env="bar"} 25
 `)
-	f(``, ``, true, time.Hour, false, false, `
+	f(``, ``, time.Hour, false, false, `
 metric{env="dev"} 10
 metric{env="foo"} 20
 metric{env="dev"} 15
@@ -134,7 +130,7 @@ metric{env="foo"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, true, time.Hour, false, false, `
+`, time.Hour, false, false, `
 metric{env="dev"} 10
 metric{env="bar"} 20
 metric{env="dev"} 15
@@ -144,7 +140,7 @@ metric{env="bar"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, true, time.Hour, true, false, `
+`, time.Hour, true, false, `
 metric{env="test"} 10
 metric{env="dev"} 20
 metric{env="foo"} 15
@@ -154,7 +150,7 @@ metric{env="dev"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, true, time.Hour, false, true, `
+`, time.Hour, false, true, `
 metric{env="foo"} 10
 metric{env="dev"} 20
 metric{env="foo"} 15
@@ -164,7 +160,7 @@ metric{env="dev"} 25
 - action: keep
   source_labels: [env]
   regex: "dev"
-`, true, time.Hour, true, true, `
+`, time.Hour, true, true, `
 metric{env="dev"} 10
 metric{env="test"} 20
 metric{env="dev"} 15

app/vmagent/remotewrite/streamaggr.go

@@ -35,9 +35,6 @@ var (
 		"clients pushing data into the vmagent. See https://docs.victoriametrics.com/stream-aggregation/#ignore-aggregation-intervals-on-start")
 	streamAggrGlobalDropInputLabels = flagutil.NewArrayString("streamAggr.dropInputLabels", "An optional list of labels to drop from samples for aggregator "+
 		"before stream de-duplication and aggregation . See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
-	streamAggrGlobalEnableWindows = flag.Bool("streamAggr.enableWindows", false, "Enables aggregation within fixed windows for all global aggregators. "+
-		"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
-		"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
 
 	// Per URL config
 	streamAggrConfig = flagutil.NewArrayString("remoteWrite.streamAggr.config", "Optional path to file with stream aggregation config for the corresponding -remoteWrite.url. "+
@@ -56,15 +53,12 @@ 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 buffered delayed data from clients pushing data into the vmagent. "+
+		"you observe incorrect aggregation results after vmagent restarts. It could be caused by receiving bufferred 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. "+
 		"Multiple labels per remoteWrite.url must be delimited by '^^': -remoteWrite.streamAggr.dropInputLabels='replica^^az,replica'. "+
 		"See https://docs.victoriametrics.com/stream-aggregation/#dropping-unneeded-labels")
-	streamAggrEnableWindows = flagutil.NewArrayBool("remoteWrite.streamAggr.enableWindows", "Enables aggregation within fixed windows for all remote write's aggregators. "+
-		"This allows to get more precise results, but impacts resource usage as it requires twice more memory to store two states. "+
-		"See https://docs.victoriametrics.com/stream-aggregation/#aggregation-windows.")
 )
 
 // CheckStreamAggrConfigs checks -remoteWrite.streamAggr.config and -streamAggr.config.
@@ -131,7 +125,7 @@ func reloadStreamAggrConfigGlobal() {
 func initStreamAggrConfigGlobal() {
 	sas, err := newStreamAggrConfigGlobal()
 	if err != nil {
-		logger.Fatalf("cannot initialize global stream aggregators: %s", err)
+		logger.Fatalf("cannot initialize gloabl stream aggregators: %s", err)
 	}
 	if sas != nil {
 		filePath := sas.FilePath()
@@ -141,7 +135,7 @@ func initStreamAggrConfigGlobal() {
 	}
 	dedupInterval := *streamAggrGlobalDedupInterval
 	if dedupInterval > 0 {
-		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, *streamAggrGlobalEnableWindows, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
+		deduplicatorGlobal = streamaggr.NewDeduplicator(pushToRemoteStoragesTrackDropped, dedupInterval, *streamAggrGlobalDropInputLabels, "dedup-global")
 	}
 }
 
@@ -167,7 +161,7 @@ func (rwctx *remoteWriteCtx) initStreamAggrConfig() {
 		if streamAggrDropInputLabels.GetOptionalArg(idx) != "" {
 			dropLabels = strings.Split(streamAggrDropInputLabels.GetOptionalArg(idx), "^^")
 		}
-		rwctx.deduplicator = streamaggr.NewDeduplicator(rwctx.pushInternalTrackDropped, *streamAggrGlobalEnableWindows, dedupInterval, dropLabels, alias)
+		rwctx.deduplicator = streamaggr.NewDeduplicator(rwctx.pushInternalTrackDropped, dedupInterval, dropLabels, alias)
 	}
 }
 
@@ -213,7 +207,6 @@ func newStreamAggrConfigGlobal() (*streamaggr.Aggregators, error) {
 		IgnoreOldSamples:     *streamAggrGlobalIgnoreOldSamples,
 		IgnoreFirstIntervals: *streamAggrGlobalIgnoreFirstIntervals,
 		KeepInput:            *streamAggrGlobalKeepInput,
-		EnableWindows:        *streamAggrGlobalEnableWindows,
 	}
 
 	sas, err := streamaggr.LoadFromFile(path, pushToRemoteStoragesTrackDropped, opts, "global")
@@ -247,7 +240,6 @@ func newStreamAggrConfigPerURL(idx int, pushFunc streamaggr.PushFunc) (*streamag
 		IgnoreOldSamples:     streamAggrIgnoreOldSamples.GetOptionalArg(idx),
 		IgnoreFirstIntervals: streamAggrIgnoreFirstIntervals.GetOptionalArg(idx),
 		KeepInput:            streamAggrKeepInput.GetOptionalArg(idx),
-		EnableWindows:        streamAggrEnableWindows.GetOptionalArg(idx),
 	}
 
 	sas, err := streamaggr.LoadFromFile(path, pushFunc, opts, alias)

app/vmagent/vmimport/request_handler.go

@@ -30,8 +30,8 @@ func InsertHandler(at *auth.Token, req *http.Request) error {
 	if err != nil {
 		return err
 	}
-	encoding := req.Header.Get("Content-Encoding")
-	return stream.Parse(req.Body, encoding, func(rows []parser.Row) error {
+	isGzipped := req.Header.Get("Content-Encoding") == "gzip"
+	return stream.Parse(req.Body, isGzipped, func(rows []parser.Row) error {
 		return insertRows(at, rows, extraLabels)
 	})
 }

app/vmalert-tool/main.go

@@ -51,11 +51,6 @@ Examples:
 						Usage:    `Optional external URL to template in rule's labels or annotations.`,
 						Required: false,
 					},
-					&cli.StringFlag{
-						Name:     "httpListenPort",
-						Usage:    `Optional local port for incoming HTTP requests. If not specified, a random unoccupied port will be used.`,
-						Required: false,
-					},
 					&cli.StringFlag{
 						Name:     "loggerLevel",
 						Usage:    `Minimum level of errors to log. Possible values: INFO, WARN, ERROR, FATAL, PANIC (default "ERROR").`,
@@ -63,7 +58,7 @@ Examples:
 					},
 				},
 				Action: func(c *cli.Context) error {
-					if failed := unittest.UnitTest(c.StringSlice("files"), c.Bool("disableAlertgroupLabel"), c.StringSlice("external.label"), c.String("external.url"), c.String("httpListenPort"), c.String("loggerLevel")); failed {
+					if failed := unittest.UnitTest(c.StringSlice("files"), c.Bool("disableAlertgroupLabel"), c.StringSlice("external.label"), c.String("external.url"), c.String("loggerLevel")); failed {
 						return fmt.Errorf("unittest failed")
 					}
 					return nil

app/vmalert-tool/unittest/unittest.go

@@ -4,18 +4,13 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"net"
 	"net/http"
-	"net/http/httptest"
 	"net/url"
 	"os"
-	"os/signal"
 	"path/filepath"
 	"reflect"
 	"sort"
-	"strconv"
 	"strings"
-	"syscall"
 	"time"
 
 	"gopkg.in/yaml.v2"
@@ -33,6 +28,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/promql"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -42,9 +38,15 @@ import (
 
 var (
 	storagePath    string
-	httpListenAddr string
+	httpListenAddr = ":8880"
 	// insert series from 1970-01-01T00:00:00
-	testStartTime          = time.Unix(0, 0).UTC()
+	testStartTime = time.Unix(0, 0).UTC()
+
+	testPromWriteHTTPPath = "http://127.0.0.1" + httpListenAddr + "/api/v1/write"
+	testDataSourcePath    = "http://127.0.0.1" + httpListenAddr + "/prometheus"
+	testRemoteWritePath   = "http://127.0.0.1" + httpListenAddr
+	testHealthHTTPPath    = "http://127.0.0.1" + httpListenAddr + "/health"
+
 	testLogLevel           = "ERROR"
 	disableAlertgroupLabel bool
 )
@@ -54,7 +56,7 @@ const (
 )
 
 // UnitTest runs unittest for files
-func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, externalURL, httpListenPort, logLevel string) bool {
+func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, externalURL, logLevel string) bool {
 	if logLevel != "" {
 		testLogLevel = logLevel
 	}
@@ -65,42 +67,7 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	if err := templates.Load([]string{}, *eu); err != nil {
 		logger.Fatalf("failed to load template: %v", err)
 	}
-
-	// set up http server
-	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		switch r.URL.Path {
-		case "/prometheus/api/v1/query":
-			if err := prometheus.QueryHandler(nil, time.Now(), w, r); err != nil {
-				httpserver.Errorf(w, r, "%s", err)
-			}
-		case "/prometheus/api/v1/write", "/api/v1/write":
-			if err := promremotewrite.InsertHandler(r); err != nil {
-				httpserver.Errorf(w, r, "%s", err)
-			}
-		default:
-		}
-	})
-	if httpListenPort == "" {
-		server := httptest.NewServer(handler)
-		httpListenAddr = strings.Split(server.URL, ":")[2]
-		defer server.Close()
-	} else {
-		httpListenAddr = httpListenPort
-		ln, err := net.Listen("tcp", fmt.Sprintf(":%s", httpListenPort))
-		if err != nil {
-			logger.Fatalf("cannot listen on port %s: %v", httpListenPort, err)
-		}
-		go func() {
-			err = http.Serve(ln, handler)
-			if err != nil {
-				logger.Fatalf("cannot start http server: %v", err)
-			}
-			defer ln.Close()
-		}()
-	}
-
-	// adding time.Now().UnixNano() to avoid possible file conflict when multiple processes run on a single host
-	storagePath = filepath.Join(os.TempDir(), testStoragePath, strconv.FormatInt(time.Now().UnixNano(), 10))
+	storagePath = filepath.Join(os.TempDir(), testStoragePath)
 	processFlags()
 	vminsert.Init()
 	vmselect.Init()
@@ -135,34 +102,16 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	}
 
 	var failed bool
-	runTest := func() bool {
-		for fileName, file := range testfiles {
-			if err := ruleUnitTest(fileName, file, labels); err != nil {
-				fmt.Println("FAILED")
-				fmt.Printf("failed to run unit test for file %q: \n%v", fileName, err)
-				return true
-			}
-			fmt.Println("SUCCESS")
+	for fileName, file := range testfiles {
+		if err := ruleUnitTest(fileName, file, labels); err != nil {
+			fmt.Println("FAILED")
+			fmt.Printf("failed to run unit test for file %q: \n%v", fileName, err)
+			failed = true
+		} else {
+			fmt.Println("  SUCCESS")
 		}
-		return false
 	}
 
-	finishCh := make(chan struct{}, 1)
-	go func() {
-		failed = runTest()
-		finishCh <- struct{}{}
-	}()
-
-	sigs := make(chan os.Signal, 1)
-	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
-	select {
-	case sig := <-sigs:
-		fmt.Printf("received signal %s\n", sig)
-		failed = true
-		break
-	case <-finishCh:
-		break
-	}
 	return failed
 }
 
@@ -262,8 +211,8 @@ func processFlags() {
 		{flag: "search.disableCache", value: "true"},
 		// set storage retention time to 100 years, allow to store series from 1970-01-01T00:00:00.
 		{flag: "retentionPeriod", value: "100y"},
-		{flag: "datasource.url", value: fmt.Sprintf("http://127.0.0.1:%s/prometheus", httpListenAddr)},
-		{flag: "remoteWrite.url", value: fmt.Sprintf("http://127.0.0.1:%s", httpListenAddr)},
+		{flag: "datasource.url", value: testDataSourcePath},
+		{flag: "remoteWrite.url", value: testRemoteWritePath},
 		{flag: "notifier.blackhole", value: "true"},
 	} {
 		// panics if flag doesn't exist
@@ -275,10 +224,27 @@ func processFlags() {
 
 func setUp() {
 	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
+	var ab flagutil.ArrayBool
+	go httpserver.Serve([]string{httpListenAddr}, &ab, func(w http.ResponseWriter, r *http.Request) bool {
+		switch r.URL.Path {
+		case "/prometheus/api/v1/query":
+			if err := prometheus.QueryHandler(nil, time.Now(), w, r); err != nil {
+				httpserver.Errorf(w, r, "%s", err)
+			}
+			return true
+		case "/prometheus/api/v1/write", "/api/v1/write":
+			if err := promremotewrite.InsertHandler(r); err != nil {
+				httpserver.Errorf(w, r, "%s", err)
+			}
+			return true
+		default:
+		}
+		return false
+	})
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	readyCheckFunc := func() bool {
-		resp, err := http.Get(fmt.Sprintf("http://127.0.0.1:%s/health", httpListenAddr))
+		resp, err := http.Get(testHealthHTTPPath)
 		if err != nil {
 			return false
 		}
@@ -300,6 +266,9 @@ checkCheck:
 }
 
 func tearDown() {
+	if err := httpserver.Stop([]string{httpListenAddr}); err != nil {
+		logger.Errorf("cannot stop the webservice: %s", err)
+	}
 	vmstorage.Stop()
 	metrics.UnregisterAllMetrics()
 	fs.MustRemoveAll(storagePath)
@@ -314,7 +283,7 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 	if tg.Interval == nil {
 		tg.Interval = promutils.NewDuration(evalInterval)
 	}
-	err := writeInputSeries(tg.InputSeries, tg.Interval, testStartTime, fmt.Sprintf("http://127.0.0.1:%s/api/v1/write", httpListenAddr))
+	err := writeInputSeries(tg.InputSeries, tg.Interval, testStartTime, testPromWriteHTTPPath)
 	if err != nil {
 		return []error{err}
 	}

app/vmalert-tool/unittest/unittest_test.go

@@ -8,7 +8,7 @@ func TestUnitTest_Failure(t *testing.T) {
 	f := func(files []string) {
 		t.Helper()
 
-		failed := UnitTest(files, false, nil, "", "", "")
+		failed := UnitTest(files, false, nil, "", "")
 		if !failed {
 			t.Fatalf("expecting failed test")
 		}
@@ -20,20 +20,19 @@ func TestUnitTest_Failure(t *testing.T) {
 }
 
 func TestUnitTest_Success(t *testing.T) {
-	f := func(disableGroupLabel bool, files []string, externalLabels []string, externalURL, httpPort string) {
+	f := func(disableGroupLabel bool, files []string, externalLabels []string, externalURL string) {
 		t.Helper()
 
-		failed := UnitTest(files, disableGroupLabel, externalLabels, externalURL, httpPort, "")
+		failed := UnitTest(files, disableGroupLabel, externalLabels, externalURL, "")
 		if failed {
 			t.Fatalf("unexpected failed test")
 		}
 	}
 
-	// run multi files with random http port
-	f(false, []string{"./testdata/test1.yaml", "./testdata/test2.yaml"}, []string{"cluster=prod"}, "http://grafana:3000", "")
+	// run multi files
+	f(false, []string{"./testdata/test1.yaml", "./testdata/test2.yaml"}, []string{"cluster=prod"}, "http://grafana:3000")
 
 	// disable group label
 	// template with null external values
-	// specify httpListenAddr
-	f(true, []string{"./testdata/disable-group-label.yaml"}, nil, "", "8880")
+	f(true, []string{"./testdata/disable-group-label.yaml"}, nil, "")
 }

app/vmalert/config/config.go

@@ -88,9 +88,6 @@ func (g *Group) Validate(validateTplFn ValidateTplFn, validateExpressions bool)
 	if g.EvalOffset.Duration() > g.Interval.Duration() {
 		return fmt.Errorf("eval_offset should be smaller than interval; now eval_offset: %v, interval: %v", g.EvalOffset.Duration(), g.Interval.Duration())
 	}
-	if g.EvalOffset != nil && g.EvalDelay != nil {
-		return fmt.Errorf("eval_offset cannot be used with eval_delay")
-	}
 	if g.Limit < 0 {
 		return fmt.Errorf("invalid limit %d, shouldn't be less than 0", g.Limit)
 	}

app/vmalert/config/config_test.go

@@ -157,19 +157,6 @@ func TestGroupValidate_Failure(t *testing.T) {
 
 	f(&Group{}, false, "group name must be set")
 
-	f(&Group{
-		Name: "both record and alert are not set",
-		Rules: []Rule{
-			{
-				Expr: "sum(up == 0 ) by (host)",
-				For:  promutils.NewDuration(10 * time.Millisecond),
-			},
-			{
-				Expr: "sumSeries(time('foo.bar',10))",
-			},
-		},
-	}, false, "invalid rule")
-
 	f(&Group{
 		Name:     "negative interval",
 		Interval: promutils.NewDuration(-1),
@@ -253,6 +240,59 @@ func TestGroupValidate_Failure(t *testing.T) {
 		},
 	}, false, "duplicate")
 
+	f(&Group{
+		Name: "test graphite with prometheus expr",
+		Type: NewGraphiteType(),
+		Rules: []Rule{
+			{
+				Expr: "sum(up == 0 ) by (host)",
+				For:  promutils.NewDuration(10 * time.Millisecond),
+			},
+			{
+				Expr: "sumSeries(time('foo.bar',10))",
+			},
+		},
+	}, false, "invalid rule")
+
+	f(&Group{
+		Name: "test graphite inherit",
+		Type: NewGraphiteType(),
+		Rules: []Rule{
+			{
+				Expr: "sumSeries(time('foo.bar',10))",
+				For:  promutils.NewDuration(10 * time.Millisecond),
+			},
+			{
+				Expr: "sum(up == 0 ) by (host)",
+			},
+		},
+	}, false, "either `record` or `alert` must be set")
+
+	f(&Group{
+		Name: "test vlogs with prometheus expr",
+		Type: NewVLogsType(),
+		Rules: []Rule{
+			{
+				Expr: "sum(up == 0 ) by (host)",
+				For:  promutils.NewDuration(10 * time.Millisecond),
+			},
+			{
+				Expr: "sumSeries(time('foo.bar',10))",
+			},
+		},
+	}, false, "invalid rule")
+
+	// validate expressions
+	f(&Group{
+		Name: "test",
+		Rules: []Rule{
+			{
+				Record: "record",
+				Expr:   "up | 0",
+			},
+		},
+	}, true, "invalid expression")
+
 	f(&Group{
 		Name: "test thanos",
 		Type: NewRawType("thanos"),
@@ -263,20 +303,8 @@ func TestGroupValidate_Failure(t *testing.T) {
 		},
 	}, true, "unknown datasource type")
 
-	// validate expressions
 	f(&Group{
-		Name: "test prometheus expr",
-		Type: NewPrometheusType(),
-		Rules: []Rule{
-			{
-				Record: "record",
-				Expr:   "up | 0",
-			},
-		},
-	}, true, "bad prometheus expr")
-
-	f(&Group{
-		Name: "test graphite expr",
+		Name: "test graphite",
 		Type: NewGraphiteType(),
 		Rules: []Rule{
 			{Alert: "alert", Expr: "up == 1", Labels: map[string]string{
@@ -286,63 +314,14 @@ func TestGroupValidate_Failure(t *testing.T) {
 	}, true, "bad graphite expr")
 
 	f(&Group{
-		Name: "test vlogs expr",
+		Name: "test vlogs",
 		Type: NewVLogsType(),
 		Rules: []Rule{
-			{Alert: "alert", Expr: "stats count(*) as requests"},
+			{Alert: "alert", Expr: "stats count(*) as requests", Labels: map[string]string{
+				"description": "some-description",
+			}},
 		},
 	}, true, "bad LogsQL expr")
-
-	f(&Group{
-		Name: "test vlogs expr",
-		Type: NewVLogsType(),
-		Rules: []Rule{
-			{Alert: "alert", Expr: "_time: 1m | stats by (path, _time: 1m) count(*) as requests"},
-		},
-	}, true, "bad LogsQL expr")
-
-	f(&Group{
-		Name: "test graphite with prometheus expr",
-		Type: NewGraphiteType(),
-		Rules: []Rule{
-			{
-				Record: "r1",
-				ID:     1,
-				Expr:   "sumSeries(time('foo.bar',10))",
-				For:    promutils.NewDuration(10 * time.Millisecond),
-			},
-			{
-				Record: "r2",
-				ID:     2,
-				Expr:   "sum(up == 0 ) by (host)",
-			},
-		},
-	}, true, "bad graphite expr")
-
-	f(&Group{
-		Name: "test vlogs with prometheus exp",
-		Type: NewVLogsType(),
-		Rules: []Rule{
-			{
-				Record: "r1",
-				Expr:   "sum(up == 0 ) by (host)",
-				For:    promutils.NewDuration(10 * time.Millisecond),
-			},
-		},
-	}, true, "bad LogsQL expr")
-
-	f(&Group{
-		Name: "test prometheus with vlogs exp",
-		Type: NewPrometheusType(),
-		Rules: []Rule{
-			{
-				Record: "r1",
-				Expr:   "* | stats by (path) count()",
-				For:    promutils.NewDuration(10 * time.Millisecond),
-			},
-		},
-	}, true, "bad prometheus expr")
-
 }
 
 func TestGroupValidate_Success(t *testing.T) {

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

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

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

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

app/vmalert/config/types.go

@@ -71,18 +71,9 @@ func (t *Type) ValidateExpr(expr string) error {
 			return fmt.Errorf("bad prometheus expr: %q, err: %w", expr, err)
 		}
 	case "vlogs":
-		q, err := logstorage.ParseStatsQuery(expr, 0)
-		if err != nil {
+		if _, err := logstorage.ParseStatsQuery(expr, 0); err != nil {
 			return fmt.Errorf("bad LogsQL expr: %q, err: %w", expr, err)
 		}
-		fields, _ := q.GetStatsByFields()
-		for i := range fields {
-			// VictoriaLogs inserts `_time` field as a label in result when query with `stats by (_time:step)`,
-			// making the result meaningless and may lead to cardinality issues.
-			if fields[i] == "_time" {
-				return fmt.Errorf("bad LogsQL expr: %q, err: cannot contain time buckets stats pipe `stats by (_time:step)`", expr)
-			}
-		}
 	default:
 		return fmt.Errorf("unknown datasource type=%q", t.Name)
 	}

app/vmalert/main.go

@@ -405,9 +405,6 @@ func configsEqual(a, b []config.Group) bool {
 		if a[i].Checksum != b[i].Checksum {
 			return false
 		}
-		if a[i].File != b[i].File {
-			return false
-		}
 	}
 	return true
 }

app/vmalert/manager.go

@@ -160,8 +160,8 @@ func (m *manager) update(ctx context.Context, groupsCfg []config.Group, restore
 			// it is important to call InterruptEval before the update, because cancel fn
 			// can be re-assigned during the update.
 			item.old.InterruptEval()
-			go func(oldGroup *rule.Group, newGroup *rule.Group) {
-				oldGroup.UpdateWith(newGroup)
+			go func(old *rule.Group, new *rule.Group) {
+				old.UpdateWith(new)
 				wg.Done()
 			}(item.old, item.new)
 		}

app/vmalert/notifier/alertmanager.go

@@ -10,12 +10,9 @@ import (
 	"strings"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
@@ -31,36 +28,25 @@ type AlertManager struct {
 	// stores already parsed RelabelConfigs object
 	relabelConfigs *promrelabel.ParsedConfigs
 
-	metrics *notifierMetrics
+	metrics *metrics
 }
 
-type notifierMetrics struct {
-	set *metrics.Set
-
-	alertsSent         *metrics.Counter
-	alertsSendErrors   *metrics.Counter
-	alertsSendDuration *metrics.Histogram
+type metrics struct {
+	alertsSent       *utils.Counter
+	alertsSendErrors *utils.Counter
 }
 
-func newNotifierMetrics(addr string) *notifierMetrics {
-	set := metrics.NewSet()
-	metrics.RegisterSet(set)
-
-	return &notifierMetrics{
-		set:                set,
-		alertsSent:         set.NewCounter(fmt.Sprintf("vmalert_alerts_sent_total{addr=%q}", addr)),
-		alertsSendErrors:   set.NewCounter(fmt.Sprintf("vmalert_alerts_send_errors_total{addr=%q}", addr)),
-		alertsSendDuration: set.NewHistogram(fmt.Sprintf("vmalert_alerts_send_duration_seconds{addr=%q}", addr)),
+func newMetrics(addr string) *metrics {
+	return &metrics{
+		alertsSent:       utils.GetOrCreateCounter(fmt.Sprintf("vmalert_alerts_sent_total{addr=%q}", addr)),
+		alertsSendErrors: utils.GetOrCreateCounter(fmt.Sprintf("vmalert_alerts_send_errors_total{addr=%q}", addr)),
 	}
 }
 
-func (nm *notifierMetrics) close() {
-	metrics.UnregisterSet(nm.set, true)
-}
-
 // Close is a destructor method for AlertManager
 func (am *AlertManager) Close() {
-	am.metrics.close()
+	am.metrics.alertsSent.Unregister()
+	am.metrics.alertsSendErrors.Unregister()
 }
 
 // Addr returns address where alerts are sent.
@@ -74,9 +60,7 @@ func (am AlertManager) Addr() string {
 // Send an alert or resolve message
 func (am *AlertManager) Send(ctx context.Context, alerts []Alert, headers map[string]string) error {
 	am.metrics.alertsSent.Add(len(alerts))
-	startTime := time.Now()
 	err := am.send(ctx, alerts, headers)
-	am.metrics.alertsSendDuration.UpdateDuration(startTime)
 	if err != nil {
 		am.metrics.alertsSendErrors.Add(len(alerts))
 	}
@@ -85,17 +69,7 @@ func (am *AlertManager) Send(ctx context.Context, alerts []Alert, headers map[st
 
 func (am *AlertManager) send(ctx context.Context, alerts []Alert, headers map[string]string) error {
 	b := &bytes.Buffer{}
-	alertsToSend := make([]Alert, 0, len(alerts))
-	lblss := make([][]prompbmarshal.Label, 0, len(alerts))
-	for _, a := range alerts {
-		lbls := a.applyRelabelingIfNeeded(am.relabelConfigs)
-		if len(lbls) == 0 {
-			continue
-		}
-		alertsToSend = append(alertsToSend, a)
-		lblss = append(lblss, lbls)
-	}
-	writeamRequest(b, alertsToSend, am.argFunc, lblss)
+	writeamRequest(b, alerts, am.argFunc, am.relabelConfigs)
 
 	req, err := http.NewRequest(http.MethodPost, am.addr.String(), b)
 	if err != nil {
@@ -195,6 +169,6 @@ func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, authCfg proma
 		relabelConfigs: relabelCfg,
 		client:         &http.Client{Transport: tr},
 		timeout:        timeout,
-		metrics:        newNotifierMetrics(alertManagerURL),
+		metrics:        newMetrics(alertManagerURL),
 	}, nil
 }

app/vmalert/notifier/alertmanager_request.qtpl

@@ -1,14 +1,15 @@
 {% import (
     "time"
 
-    "github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+    "github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 ) %}
 {% stripspace %}
 
-{% func amRequest(alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) %}
+{% func amRequest(alerts []Alert, generatorURL func(Alert) string, relabelCfg *promrelabel.ParsedConfigs) %}
 [
 {% for i, alert := range alerts %}
-{% code lbls := lblss[i] %}
+{% code lbls := alert.applyRelabelingIfNeeded(relabelCfg) %}
+{% if len(lbls) == 0 %} {% continue %} {% endif %}
 {
     "startsAt":{%q= alert.Start.Format(time.RFC3339Nano) %},
     "generatorURL": {%q= generatorURL(alert) %},

app/vmalert/notifier/alertmanager_request.qtpl.go

@@ -8,7 +8,7 @@ package notifier
 import (
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 )
 
 //line app/vmalert/notifier/alertmanager_request.qtpl:8
@@ -25,116 +25,122 @@ var (
 )
 
 //line app/vmalert/notifier/alertmanager_request.qtpl:8
-func streamamRequest(qw422016 *qt422016.Writer, alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) {
+func streamamRequest(qw422016 *qt422016.Writer, alerts []Alert, generatorURL func(Alert) string, relabelCfg *promrelabel.ParsedConfigs) {
 //line app/vmalert/notifier/alertmanager_request.qtpl:8
 	qw422016.N().S(`[`)
 //line app/vmalert/notifier/alertmanager_request.qtpl:10
 	for i, alert := range alerts {
 //line app/vmalert/notifier/alertmanager_request.qtpl:11
-		lbls := lblss[i]
+		lbls := alert.applyRelabelingIfNeeded(relabelCfg)
 
-//line app/vmalert/notifier/alertmanager_request.qtpl:11
-		qw422016.N().S(`{"startsAt":`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:13
-		qw422016.N().Q(alert.Start.Format(time.RFC3339Nano))
-//line app/vmalert/notifier/alertmanager_request.qtpl:13
-		qw422016.N().S(`,"generatorURL":`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:14
-		qw422016.N().Q(generatorURL(alert))
-//line app/vmalert/notifier/alertmanager_request.qtpl:14
-		qw422016.N().S(`,`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:15
-		if !alert.End.IsZero() {
-//line app/vmalert/notifier/alertmanager_request.qtpl:15
-			qw422016.N().S(`"endsAt":`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:16
-			qw422016.N().Q(alert.End.Format(time.RFC3339Nano))
-//line app/vmalert/notifier/alertmanager_request.qtpl:16
-			qw422016.N().S(`,`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:17
+//line app/vmalert/notifier/alertmanager_request.qtpl:12
+		if len(lbls) == 0 {
+//line app/vmalert/notifier/alertmanager_request.qtpl:12
+			continue
+//line app/vmalert/notifier/alertmanager_request.qtpl:12
 		}
+//line app/vmalert/notifier/alertmanager_request.qtpl:12
+		qw422016.N().S(`{"startsAt":`)
+//line app/vmalert/notifier/alertmanager_request.qtpl:14
+		qw422016.N().Q(alert.Start.Format(time.RFC3339Nano))
+//line app/vmalert/notifier/alertmanager_request.qtpl:14
+		qw422016.N().S(`,"generatorURL":`)
+//line app/vmalert/notifier/alertmanager_request.qtpl:15
+		qw422016.N().Q(generatorURL(alert))
+//line app/vmalert/notifier/alertmanager_request.qtpl:15
+		qw422016.N().S(`,`)
+//line app/vmalert/notifier/alertmanager_request.qtpl:16
+		if !alert.End.IsZero() {
+//line app/vmalert/notifier/alertmanager_request.qtpl:16
+			qw422016.N().S(`"endsAt":`)
 //line app/vmalert/notifier/alertmanager_request.qtpl:17
+			qw422016.N().Q(alert.End.Format(time.RFC3339Nano))
+//line app/vmalert/notifier/alertmanager_request.qtpl:17
+			qw422016.N().S(`,`)
+//line app/vmalert/notifier/alertmanager_request.qtpl:18
+		}
+//line app/vmalert/notifier/alertmanager_request.qtpl:18
 		qw422016.N().S(`"labels": {`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:19
+//line app/vmalert/notifier/alertmanager_request.qtpl:20
 		ll := len(lbls)
 
-//line app/vmalert/notifier/alertmanager_request.qtpl:20
+//line app/vmalert/notifier/alertmanager_request.qtpl:21
 		for idx, l := range lbls {
-//line app/vmalert/notifier/alertmanager_request.qtpl:21
+//line app/vmalert/notifier/alertmanager_request.qtpl:22
 			qw422016.N().Q(l.Name)
-//line app/vmalert/notifier/alertmanager_request.qtpl:21
+//line app/vmalert/notifier/alertmanager_request.qtpl:22
 			qw422016.N().S(`:`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:21
+//line app/vmalert/notifier/alertmanager_request.qtpl:22
 			qw422016.N().Q(l.Value)
-//line app/vmalert/notifier/alertmanager_request.qtpl:21
+//line app/vmalert/notifier/alertmanager_request.qtpl:22
 			if idx != ll-1 {
-//line app/vmalert/notifier/alertmanager_request.qtpl:21
+//line app/vmalert/notifier/alertmanager_request.qtpl:22
 				qw422016.N().S(`,`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:21
+//line app/vmalert/notifier/alertmanager_request.qtpl:22
 			}
-//line app/vmalert/notifier/alertmanager_request.qtpl:22
+//line app/vmalert/notifier/alertmanager_request.qtpl:23
 		}
-//line app/vmalert/notifier/alertmanager_request.qtpl:22
+//line app/vmalert/notifier/alertmanager_request.qtpl:23
 		qw422016.N().S(`},"annotations": {`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:25
+//line app/vmalert/notifier/alertmanager_request.qtpl:26
 		c := len(alert.Annotations)
 
-//line app/vmalert/notifier/alertmanager_request.qtpl:26
-		for k, v := range alert.Annotations {
 //line app/vmalert/notifier/alertmanager_request.qtpl:27
+		for k, v := range alert.Annotations {
+//line app/vmalert/notifier/alertmanager_request.qtpl:28
 			c = c - 1
 
-//line app/vmalert/notifier/alertmanager_request.qtpl:28
+//line app/vmalert/notifier/alertmanager_request.qtpl:29
 			qw422016.N().Q(k)
-//line app/vmalert/notifier/alertmanager_request.qtpl:28
+//line app/vmalert/notifier/alertmanager_request.qtpl:29
 			qw422016.N().S(`:`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:28
+//line app/vmalert/notifier/alertmanager_request.qtpl:29
 			qw422016.N().Q(v)
-//line app/vmalert/notifier/alertmanager_request.qtpl:28
+//line app/vmalert/notifier/alertmanager_request.qtpl:29
 			if c > 0 {
-//line app/vmalert/notifier/alertmanager_request.qtpl:28
+//line app/vmalert/notifier/alertmanager_request.qtpl:29
 				qw422016.N().S(`,`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:28
+//line app/vmalert/notifier/alertmanager_request.qtpl:29
 			}
-//line app/vmalert/notifier/alertmanager_request.qtpl:29
+//line app/vmalert/notifier/alertmanager_request.qtpl:30
 		}
-//line app/vmalert/notifier/alertmanager_request.qtpl:29
+//line app/vmalert/notifier/alertmanager_request.qtpl:30
 		qw422016.N().S(`}}`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:32
+//line app/vmalert/notifier/alertmanager_request.qtpl:33
 		if i != len(alerts)-1 {
-//line app/vmalert/notifier/alertmanager_request.qtpl:32
+//line app/vmalert/notifier/alertmanager_request.qtpl:33
 			qw422016.N().S(`,`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:32
+//line app/vmalert/notifier/alertmanager_request.qtpl:33
 		}
-//line app/vmalert/notifier/alertmanager_request.qtpl:33
+//line app/vmalert/notifier/alertmanager_request.qtpl:34
 	}
-//line app/vmalert/notifier/alertmanager_request.qtpl:33
+//line app/vmalert/notifier/alertmanager_request.qtpl:34
 	qw422016.N().S(`]`)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 }
 
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
-func writeamRequest(qq422016 qtio422016.Writer, alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) {
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
+func writeamRequest(qq422016 qtio422016.Writer, alerts []Alert, generatorURL func(Alert) string, relabelCfg *promrelabel.ParsedConfigs) {
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
-	streamamRequest(qw422016, alerts, generatorURL, lblss)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
+	streamamRequest(qw422016, alerts, generatorURL, relabelCfg)
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 	qt422016.ReleaseWriter(qw422016)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 }
 
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
-func amRequest(alerts []Alert, generatorURL func(Alert) string, lblss [][]prompbmarshal.Label) string {
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
+func amRequest(alerts []Alert, generatorURL func(Alert) string, relabelCfg *promrelabel.ParsedConfigs) string {
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
-	writeamRequest(qb422016, alerts, generatorURL, lblss)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
+	writeamRequest(qb422016, alerts, generatorURL, relabelCfg)
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 	qs422016 := string(qb422016.B)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 	return qs422016
-//line app/vmalert/notifier/alertmanager_request.qtpl:35
+//line app/vmalert/notifier/alertmanager_request.qtpl:36
 }

app/vmalert/notifier/alertmanager_test.go

@@ -105,16 +105,6 @@ func TestAlertManager_Send(t *testing.T) {
 			if r.Header.Get(headerKey) != "bar" {
 				t.Fatalf("expected header %q to be set to %q; got %q instead", headerKey, "bar", r.Header.Get(headerKey))
 			}
-		case 4:
-			var a []struct {
-				Labels map[string]string `json:"labels"`
-			}
-			if err := json.NewDecoder(r.Body).Decode(&a); err != nil {
-				t.Fatalf("can not unmarshal data into alert %s", err)
-			}
-			if len(a) != 1 {
-				t.Fatalf("expected 1 alert in array got %d", len(a))
-			}
 		}
 	})
 	srv := httptest.NewServer(mux)
@@ -178,20 +168,7 @@ func TestAlertManager_Send(t *testing.T) {
 		t.Fatalf("unexpected error %s", err)
 	}
 
-	if err := am.Send(context.Background(), []Alert{
-		{
-			Name:   "alert1",
-			Labels: map[string]string{"rule": "test"},
-		},
-		{
-			Name:   "alert2",
-			Labels: map[string]string{},
-		},
-	}, map[string]string{}); err != nil {
-		t.Fatalf("unexpected error %s", err)
-	}
-
-	if c != 4 {
-		t.Fatalf("expected 4 calls(count from zero) to server got %d", c)
+	if c != 3 {
+		t.Fatalf("expected 3 calls(count from zero) to server got %d", c)
 	}
 }

app/vmalert/notifier/init.go

@@ -54,7 +54,6 @@ var (
 		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
 	oauth2Scopes = flagutil.NewArrayString("notifier.oauth2.scopes", "Optional OAuth2 scopes to use for -notifier.url. Scopes must be delimited by ';'. "+
 		"If multiple args are set, then they are applied independently for the corresponding -notifier.url")
-	sendTimeout = flagutil.NewArrayDuration("notifier.sendTimeout", 10*time.Second, "Timeout when sending alerts to the corresponding -notifier.url")
 )
 
 // cw holds a configWatcher for configPath configuration file
@@ -176,7 +175,7 @@ func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
 		}
 
 		addr = strings.TrimSuffix(addr, "/")
-		am, err := NewAlertManager(addr+alertManagerPath, gen, authCfg, nil, sendTimeout.GetOptionalArg(i))
+		am, err := NewAlertManager(addr+alertManagerPath, gen, authCfg, nil, time.Second*10)
 		if err != nil {
 			return nil, err
 		}

app/vmalert/notifier/notifier_blackhole.go

@@ -6,7 +6,7 @@ import "context"
 // to be sent.
 type blackHoleNotifier struct {
 	addr    string
-	metrics *notifierMetrics
+	metrics *metrics
 }
 
 // Send will send no notifications, but increase the metric.
@@ -22,7 +22,8 @@ func (bh blackHoleNotifier) Addr() string {
 
 // Close unregister the metrics
 func (bh *blackHoleNotifier) Close() {
-	bh.metrics.close()
+	bh.metrics.alertsSent.Unregister()
+	bh.metrics.alertsSendErrors.Unregister()
 }
 
 // newBlackHoleNotifier creates a new blackHoleNotifier
@@ -30,6 +31,6 @@ func newBlackHoleNotifier() *blackHoleNotifier {
 	address := "blackhole"
 	return &blackHoleNotifier{
 		addr:    address,
-		metrics: newNotifierMetrics(address),
+		metrics: newMetrics(address),
 	}
 }

app/vmalert/remotewrite/init.go

@@ -36,7 +36,7 @@ var (
 
 	maxQueueSize  = flag.Int("remoteWrite.maxQueueSize", defaultMaxQueueSize, "Defines the max number of pending datapoints to remote write endpoint")
 	maxBatchSize  = flag.Int("remoteWrite.maxBatchSize", defaultMaxBatchSize, "Defines max number of timeseries to be flushed at once")
-	concurrency   = flag.Int("remoteWrite.concurrency", defaultConcurrency, "Defines number of writers for concurrent writing into remote write endpoint. Default value depends on the number of available CPU cores.")
+	concurrency   = flag.Int("remoteWrite.concurrency", defaultConcurrency, "Defines number of writers for concurrent writing into remote write endpoint")
 	flushInterval = flag.Duration("remoteWrite.flushInterval", defaultFlushInterval, "Defines interval of flushes to remote write endpoint")
 
 	tlsInsecureSkipVerify = flag.Bool("remoteWrite.tlsInsecureSkipVerify", false, "Whether to skip tls verification when connecting to -remoteWrite.url")

app/vmalert/rule/alerting.go

@@ -9,8 +9,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -59,69 +57,6 @@ type alertingRuleMetrics struct {
 	seriesFetched *utils.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),
-		func() float64 {
-			ar.alertsMu.RLock()
-			defer ar.alertsMu.RUnlock()
-			var num int
-			for _, a := range ar.alerts {
-				if a.State == notifier.StatePending {
-					num++
-				}
-			}
-			return float64(num)
-		})
-	arm.active = utils.NewGauge(set, fmt.Sprintf(`vmalert_alerts_firing{%s}`, labels),
-		func() float64 {
-			ar.alertsMu.RLock()
-			defer ar.alertsMu.RUnlock()
-			var num int
-			for _, a := range ar.alerts {
-				if a.State == notifier.StateFiring {
-					num++
-				}
-			}
-			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),
-		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),
-		func() float64 {
-			e := ar.state.getLast()
-			if e.SeriesFetched == nil {
-				// means seriesFetched is unsupported
-				return -1
-			}
-			seriesFetched := float64(*e.SeriesFetched)
-			if seriesFetched == 0 && e.Samples > 0 {
-				// `alert: 0.95` will fetch no series
-				// but will get one time series in response.
-				seriesFetched = float64(e.Samples)
-			}
-			return seriesFetched
-		})
-	return arm
-}
-
-func (arm *alertingRuleMetrics) close() {
-	if arm == nil {
-		return
-	}
-	arm.errors.Unregister()
-	arm.active.Unregister()
-	arm.pending.Unregister()
-	arm.samples.Unregister()
-	arm.seriesFetched.Unregister()
-}
-
 // NewAlertingRule creates a new AlertingRule
 func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule) *AlertingRule {
 	ar := &AlertingRule{
@@ -146,7 +81,8 @@ func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 			Headers:                   group.Headers,
 			Debug:                     cfg.Debug,
 		}),
-		alerts: make(map[uint64]*notifier.Alert),
+		alerts:  make(map[uint64]*notifier.Alert),
+		metrics: &alertingRuleMetrics{},
 	}
 
 	entrySize := *ruleUpdateEntriesLimit
@@ -159,17 +95,63 @@ func NewAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
 	ar.state = &ruleState{
 		entries: make([]StateEntry, entrySize),
 	}
-	ar.metrics = newAlertingRuleMetrics(group.metrics.set, ar)
+
+	labels := fmt.Sprintf(`alertname=%q, group=%q, file=%q, id="%d"`, ar.Name, group.Name, group.File, ar.ID())
+	ar.metrics.pending = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerts_pending{%s}`, labels),
+		func() float64 {
+			ar.alertsMu.RLock()
+			defer ar.alertsMu.RUnlock()
+			var num int
+			for _, a := range ar.alerts {
+				if a.State == notifier.StatePending {
+					num++
+				}
+			}
+			return float64(num)
+		})
+	ar.metrics.active = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerts_firing{%s}`, labels),
+		func() float64 {
+			ar.alertsMu.RLock()
+			defer ar.alertsMu.RUnlock()
+			var num int
+			for _, a := range ar.alerts {
+				if a.State == notifier.StateFiring {
+					num++
+				}
+			}
+			return float64(num)
+		})
+	ar.metrics.errors = utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_alerting_rules_errors_total{%s}`, labels))
+	ar.metrics.samples = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_samples{%s}`, labels),
+		func() float64 {
+			e := ar.state.getLast()
+			return float64(e.Samples)
+		})
+	ar.metrics.seriesFetched = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_alerting_rules_last_evaluation_series_fetched{%s}`, labels),
+		func() float64 {
+			e := ar.state.getLast()
+			if e.SeriesFetched == nil {
+				// means seriesFetched is unsupported
+				return -1
+			}
+			seriesFetched := float64(*e.SeriesFetched)
+			if seriesFetched == 0 && e.Samples > 0 {
+				// `alert: 0.95` will fetch no series
+				// but will get one time series in response.
+				seriesFetched = float64(e.Samples)
+			}
+			return seriesFetched
+		})
 	return ar
 }
 
-func (ar *AlertingRule) registerMetrics(g *Group) {
-	ar.metrics = newAlertingRuleMetrics(g.metrics.set, ar)
-}
-
 // close unregisters rule metrics
-func (ar *AlertingRule) unregisterMetrics() {
-	ar.metrics.close()
+func (ar *AlertingRule) close() {
+	ar.metrics.active.Unregister()
+	ar.metrics.pending.Unregister()
+	ar.metrics.errors.Unregister()
+	ar.metrics.samples.Unregister()
+	ar.metrics.seriesFetched.Unregister()
 }
 
 // String implements Stringer interface
@@ -632,7 +614,7 @@ func (ar *AlertingRule) alertToTimeSeries(a *notifier.Alert, timestamp int64) []
 }
 
 func alertToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSeries {
-	labels := make([]prompbmarshal.Label, 0, len(a.Labels)+2)
+	var labels []prompbmarshal.Label
 	for k, v := range a.Labels {
 		labels = append(labels, prompbmarshal.Label{
 			Name:  k,
@@ -652,7 +634,7 @@ func alertToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSer
 // alertForToTimeSeries returns a time series that represents
 // state of active alerts, where value is time when alert become active
 func alertForToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.TimeSeries {
-	labels := make([]prompbmarshal.Label, 0, len(a.Labels)+1)
+	var labels []prompbmarshal.Label
 	for k, v := range a.Labels {
 		labels = append(labels, prompbmarshal.Label{
 			Name:  k,
@@ -668,24 +650,21 @@ func alertForToTimeSeries(a *notifier.Alert, timestamp int64) prompbmarshal.Time
 // for alerts which changed their state from Pending to Inactive or Firing.
 func pendingAlertStaleTimeSeries(ls map[string]string, timestamp int64, includeAlertForState bool) []prompbmarshal.TimeSeries {
 	var result []prompbmarshal.TimeSeries
-	baseLabels := make([]prompbmarshal.Label, 0, len(ls)+1)
+	var baseLabels []prompbmarshal.Label
 	for k, v := range ls {
 		baseLabels = append(baseLabels, prompbmarshal.Label{
 			Name:  k,
 			Value: v,
 		})
 	}
-
-	alertsLabels := make([]prompbmarshal.Label, 0, len(ls)+2)
-	alertsLabels = append(alertsLabels, baseLabels...)
 	// __name__ already been dropped, no need to check duplication
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
+	alertsLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
 	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: alertStateLabel, Value: notifier.StatePending.String()})
 	result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsLabels))
 
 	if includeAlertForState {
-		baseLabels = append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
-		result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, baseLabels))
+		alertsForStateLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
+		result = append(result, newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsForStateLabels))
 	}
 	return result
 }
@@ -693,25 +672,22 @@ func pendingAlertStaleTimeSeries(ls map[string]string, timestamp int64, includeA
 // firingAlertStaleTimeSeries returns stale `ALERTS` and `ALERTS_FOR_STATE` time series
 // for alerts which changed their state from Firing to Inactive.
 func firingAlertStaleTimeSeries(ls map[string]string, timestamp int64) []prompbmarshal.TimeSeries {
-	baseLabels := make([]prompbmarshal.Label, 0, len(ls)+1)
+	var baseLabels []prompbmarshal.Label
 	for k, v := range ls {
 		baseLabels = append(baseLabels, prompbmarshal.Label{
 			Name:  k,
 			Value: v,
 		})
 	}
-
-	alertsLabels := make([]prompbmarshal.Label, 0, len(ls)+2)
-	alertsLabels = append(alertsLabels, baseLabels...)
 	// __name__ already been dropped, no need to check duplication
-	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
+	alertsLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertMetricName})
 	alertsLabels = append(alertsLabels, prompbmarshal.Label{Name: alertStateLabel, Value: notifier.StateFiring.String()})
 
-	baseLabels = append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
+	alertsForStateLabels := append(baseLabels, prompbmarshal.Label{Name: "__name__", Value: alertForStateMetricName})
 
 	return []prompbmarshal.TimeSeries{
 		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsLabels),
-		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, baseLabels),
+		newTimeSeries([]float64{decimal.StaleNaN}, []int64{timestamp}, alertsForStateLabels),
 	}
 }
 

app/vmalert/rule/alerting_test.go

@@ -11,8 +11,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -254,14 +252,10 @@ func TestAlertingRule_Exec(t *testing.T) {
 	},
 		map[int][]prompbmarshal.TimeSeries{
 			0: {
-				{
-					Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "empty_labels"}, {Name: "alertstate", Value: "firing"}},
-					Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
-				},
-				{
-					Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "empty_labels"}},
-					Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
-				},
+				{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "empty_labels"}, {Name: "alertstate", Value: "firing"}},
+					Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
+				{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "empty_labels"}},
+					Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
 			},
 		})
 
@@ -279,34 +273,22 @@ func TestAlertingRule_Exec(t *testing.T) {
 		4: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateInactive}}},
 	}, map[int][]prompbmarshal.TimeSeries{
 		0: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
 		},
 		1: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
 		},
 		2: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "single-firing=>inactive=>firing=>inactive=>inactive"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
 		},
 	})
 
@@ -362,54 +344,34 @@ func TestAlertingRule_Exec(t *testing.T) {
 		},
 	}, map[int][]prompbmarshal.TimeSeries{
 		0: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
 		},
 		1: {
 			// stale time series for foo, `firing -> inactive`
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
 			// new time series for foo1
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
 		},
 		2: {
 			// stale time series for foo1
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo1"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
 			// new time series for foo2
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo2"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo2"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo2"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "multiple-steps-firing"}, {Name: "name", Value: "foo2"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(2 * defaultStep).Unix()), Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
 		},
 	})
 
@@ -427,72 +389,50 @@ func TestAlertingRule_Exec(t *testing.T) {
 		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StateFiring}}},
 	}, map[int][]prompbmarshal.TimeSeries{
 		0: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
 		},
 		1: {
 			// stale time series for `pending -> firing`
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "alertstate", Value: "firing"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-fired"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Add(defaultStep).Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
 		},
 	})
 
 	f(newTestAlertingRule("for-pending=>empty", time.Second), [][]datasource.Metric{
-		{metricWithLabels(t, "name", "foo", "a1", "b1", "a2", "b2", "a3", "b3")},
-		{metricWithLabels(t, "name", "foo", "a1", "b1", "a2", "b2", "a3", "b3")},
+		{metricWithLabels(t, "name", "foo")},
+		{metricWithLabels(t, "name", "foo")},
 		// empty step to delete pending alerts
 		{},
 	}, map[int][]testAlert{
-		0: {{labels: []string{"name", "foo", "a1", "b1", "a2", "b2", "a3", "b3"}, alert: &notifier.Alert{State: notifier.StatePending}}},
-		1: {{labels: []string{"name", "foo", "a1", "b1", "a2", "b2", "a3", "b3"}, alert: &notifier.Alert{State: notifier.StatePending}}},
+		0: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
+		1: {{labels: []string{"name", "foo"}, alert: &notifier.Alert{State: notifier.StatePending}}},
 		2: {},
 	}, map[int][]prompbmarshal.TimeSeries{
 		0: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.UnixNano() / 1e6}}},
 		},
 		1: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: 1, Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: float64(ts.Unix()), Timestamp: ts.Add(defaultStep).UnixNano() / 1e6}}},
 		},
 		// stale time series for `pending -> inactive`
 		2: {
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
-			{
-				Labels:  []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "a1", Value: "b1"}, {Name: "a2", Value: "b2"}, {Name: "a3", Value: "b3"}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
-				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}},
-			},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "alertstate", Value: "pending"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
+			{Labels: []prompbmarshal.Label{{Name: "__name__", Value: alertForStateMetricName}, {Name: "alertname", Value: "for-pending=>empty"}, {Name: "name", Value: "foo"}},
+				Samples: []prompbmarshal.Sample{{Value: decimal.StaleNaN, Timestamp: ts.Add(2*defaultStep).UnixNano() / 1e6}}},
 		},
 	})
 
@@ -930,7 +870,7 @@ func TestGroup_Restore(t *testing.T) {
 			},
 		})
 
-	// one active alert with restore labels mismatch
+	// one active alert with restore labels missmatch
 	ts = time.Now().Truncate(time.Hour)
 	fqr.Set(`default_rollup(ALERTS_FOR_STATE{alertgroup="TestRestore",alertname="foo",env="dev"}[3600s])`,
 		stateMetric("foo", ts, "env", "dev", "team", "foo"))
@@ -975,7 +915,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 instead", expErr, err)
+		t.Fatalf("expected to get err %q; got %q insterad", expErr, err)
 	}
 }
 
@@ -1250,17 +1190,13 @@ func newTestAlertingRule(name string, waitFor time.Duration) *AlertingRule {
 		EvalInterval: waitFor,
 		alerts:       make(map[uint64]*notifier.Alert),
 		state:        &ruleState{entries: make([]StateEntry, 10)},
-		metrics:      getTestAlertingRuleMetrics(name),
+		metrics: &alertingRuleMetrics{
+			errors: utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_alerting_rules_errors_total{alertname=%q}`, name)),
+		},
 	}
 	return &rule
 }
 
-func getTestAlertingRuleMetrics(name string) *alertingRuleMetrics {
-	m := &alertingRuleMetrics{}
-	m.errors = utils.NewCounter(metrics.NewSet(), fmt.Sprintf(`vmalert_alerting_rules_errors_total{alertname=%q}`, name))
-	return m
-}
-
 func newTestAlertingRuleWithCustomFields(name string, waitFor, evalInterval, keepFiringFor time.Duration, annotation map[string]string) *AlertingRule {
 	rule := newTestAlertingRule(name, waitFor)
 	if evalInterval != 0 {

app/vmalert/rule/group.go

@@ -13,8 +13,6 @@ import (
 
 	"github.com/cheggaaa/pb/v3"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -22,20 +20,20 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/metrics"
 )
 
 var (
 	ruleUpdateEntriesLimit = flag.Int("rule.updateEntriesLimit", 20, "Defines the max number of rule's state updates stored in-memory. "+
 		"Rule's updates are available on rule's Details page and are used for debugging purposes. The number of stored updates can be overridden per rule via update_entries_limit param.")
-	resendDelay        = flag.Duration("rule.resendDelay", 0, "MiniMum amount of time to wait before resending an alert to notifier.")
+	resendDelay        = flag.Duration("rule.resendDelay", 0, "MiniMum amount of time to wait before resending an alert to notifier")
 	maxResolveDuration = flag.Duration("rule.maxResolveDuration", 0, "Limits the maxiMum duration for automatic alert expiration, "+
 		"which by default is 4 times evaluationInterval of the parent group")
-	evalDelay = flag.Duration("rule.evalDelay", 30*time.Second, "Adjustment of the `time` parameter for rule evaluation requests to compensate intentional data delay from the datasource. "+
-		"Normally, should be equal to `-search.latencyOffset` (cmd-line flag configured for VictoriaMetrics single-node or vmselect). "+
-		"This doesn't apply to groups with eval_offset specified.")
+	evalDelay = flag.Duration("rule.evalDelay", 30*time.Second, "Adjustment of the `time` parameter for rule evaluation requests to compensate intentional data delay from the datasource."+
+		"Normally, should be equal to `-search.latencyOffset` (cmd-line flag configured for VictoriaMetrics single-node or vmselect).")
 	disableAlertGroupLabel = flag.Bool("disableAlertgroupLabel", false, "Whether to disable adding group's Name as label to generated alerts and time series.")
-	remoteReadLookBack     = flag.Duration("remoteRead.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries. "+
-		"For example, if lookback=1h then range from now() to now()-1h will be scanned.")
+	remoteReadLookBack     = flag.Duration("remoteRead.lookback", time.Hour, "Lookback defines how far to look into past for alerts timeseries."+
+		" For example, if lookback=1h then range from now() to now()-1h will be scanned.")
 )
 
 // Group is an entity for grouping rules
@@ -76,23 +74,19 @@ type Group struct {
 }
 
 type groupMetrics struct {
-	set *metrics.Set
-
-	iterationTotal    *metrics.Counter
-	iterationDuration *metrics.Summary
-	iterationMissed   *metrics.Counter
-	iterationInterval *metrics.Gauge
+	iterationTotal    *utils.Counter
+	iterationDuration *utils.Summary
+	iterationMissed   *utils.Counter
+	iterationInterval *utils.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 {
+	m.iterationTotal = utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_iteration_total{%s}`, labels))
+	m.iterationDuration = utils.GetOrCreateSummary(fmt.Sprintf(`vmalert_iteration_duration_seconds{%s}`, labels))
+	m.iterationMissed = utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_iteration_missed_total{%s}`, labels))
+	m.iterationInterval = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_iteration_interval_seconds{%s}`, labels), func() float64 {
 		g.mu.RLock()
 		i := g.Interval.Seconds()
 		g.mu.RUnlock()
@@ -101,14 +95,6 @@ func newGroupMetrics(g *Group) *groupMetrics {
 	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 {
@@ -251,7 +237,7 @@ func (g *Group) updateWith(newGroup *Group) error {
 		if !ok {
 			// old rule is not present in the new list
 			// so we mark it for removing
-			g.Rules[i].unregisterMetrics()
+			g.Rules[i].close()
 			g.Rules[i] = nil
 			continue
 		}
@@ -271,7 +257,6 @@ func (g *Group) updateWith(newGroup *Group) error {
 	}
 	// add the rest of rules from registry
 	for _, nr := range rulesRegistry {
-		nr.registerMetrics(g)
 		newRules = append(newRules, nr)
 	}
 	// note that g.Interval is not updated here
@@ -310,7 +295,13 @@ func (g *Group) Close() {
 	g.InterruptEval()
 	<-g.finishedCh
 
-	g.metrics.close()
+	g.metrics.iterationDuration.Unregister()
+	g.metrics.iterationTotal.Unregister()
+	g.metrics.iterationMissed.Unregister()
+	g.metrics.iterationInterval.Unregister()
+	for _, rule := range g.Rules {
+		rule.close()
+	}
 }
 
 // SkipRandSleepOnGroupStart will skip random sleep delay in group first evaluation
@@ -320,8 +311,6 @@ var SkipRandSleepOnGroupStart bool
 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.
@@ -376,7 +365,6 @@ func (g *Group) Start(ctx context.Context, nts func() []notifier.Notifier, rw re
 		}
 
 		resolveDuration := getResolveDuration(g.Interval, *resendDelay, *maxResolveDuration)
-		// adjust request timestamp using evalDelay and evalAlignment if necessary
 		ts = g.adjustReqTimestamp(ts)
 		errs := e.execConcurrently(ctx, g.Rules, ts, g.Concurrency, resolveDuration, g.Limit)
 		for err := range errs {
@@ -455,8 +443,8 @@ func (g *Group) Start(ctx context.Context, nts func() []notifier.Notifier, rw re
 }
 
 // UpdateWith inserts new group to updateCh
-func (g *Group) UpdateWith(newGroup *Group) {
-	g.updateCh <- newGroup
+func (g *Group) UpdateWith(new *Group) {
+	g.updateCh <- new
 }
 
 // DeepCopy returns a deep copy of group
@@ -470,18 +458,10 @@ func (g *Group) DeepCopy() *Group {
 	return &newG
 }
 
-// if offset is specified, delayBeforeStart returns a duration to help aligning timestamp with offset;
-// otherwise, it returns a random duration between [0..interval] based on group key.
+// delayBeforeStart returns a duration on the interval between [ts..ts+interval].
+// delayBeforeStart accounts for `offset`, so returned duration should be always
+// bigger than the `offset`.
 func delayBeforeStart(ts time.Time, key uint64, interval time.Duration, offset *time.Duration) time.Duration {
-	if offset != nil {
-		currentOffsetPoint := ts.Truncate(interval).Add(*offset)
-		if currentOffsetPoint.Before(ts) {
-			// wait until the next offset point
-			return currentOffsetPoint.Add(interval).Sub(ts)
-		}
-		return currentOffsetPoint.Sub(ts)
-	}
-
 	var randSleep time.Duration
 	randSleep = time.Duration(float64(interval) * (float64(key) / (1 << 64)))
 	sleepOffset := time.Duration(ts.UnixNano() % interval.Nanoseconds())
@@ -489,6 +469,15 @@ func delayBeforeStart(ts time.Time, key uint64, interval time.Duration, offset *
 		randSleep += interval
 	}
 	randSleep -= sleepOffset
+	// check if `ts` after randSleep is before `offset`,
+	// if it is, add extra eval_offset to randSleep.
+	// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3409.
+	if offset != nil {
+		tmpEvalTS := ts.Add(randSleep)
+		if tmpEvalTS.Before(tmpEvalTS.Truncate(interval).Add(*offset)) {
+			randSleep += *offset
+		}
+	}
 	return randSleep
 }
 
@@ -594,14 +583,26 @@ func getResolveDuration(groupInterval, delta, maxDuration time.Duration) time.Du
 }
 
 func (g *Group) adjustReqTimestamp(timestamp time.Time) time.Time {
-	// if `eval_offset` is specified, timestamp is already aligned with offset, do nothing
 	if g.EvalOffset != nil {
-		return timestamp
+		// calculate the min timestamp on the evaluationInterval
+		intervalStart := timestamp.Truncate(g.Interval)
+		ts := intervalStart.Add(*g.EvalOffset)
+		if timestamp.Before(ts) {
+			// if passed timestamp is before the expected evaluation offset,
+			// then we should adjust it to the previous evaluation round.
+			// E.g. request with evaluationInterval=1h and evaluationOffset=30m
+			// was evaluated at 11:20. Then the timestamp should be adjusted
+			// to 10:30, to the previous evaluationInterval.
+			return ts.Add(-g.Interval)
+		}
+		// when `eval_offset` is using, ts shouldn't be effect by `eval_alignment` and `eval_delay`
+		// since it should be always aligned.
+		return ts
 	}
 
 	timestamp = timestamp.Add(-g.getEvalDelay())
 
-	// apply the alignment as the last step
+	// always apply the alignment as a last step
 	if g.evalAlignment == nil || *g.evalAlignment {
 		// align query time with interval to get similar result with grafana when plotting time series.
 		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5049

app/vmalert/rule/group_test.go

@@ -10,7 +10,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
 	"gopkg.in/yaml.v2"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
@@ -42,7 +41,6 @@ func TestUpdateWith(t *testing.T) {
 		g := &Group{
 			Name: "test",
 		}
-		g.metrics = newGroupMetrics(g)
 		qb := &datasource.FakeQuerier{}
 		for _, r := range currentRules {
 			r.ID = config.HashRule(r)
@@ -52,7 +50,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))
@@ -198,7 +195,6 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 		Interval: 100 * time.Hour,
 		updateCh: make(chan *Group),
 	}
-	g.metrics = newGroupMetrics(g)
 	go g.Start(context.Background(), nil, nil, nil)
 
 	rule1 := AlertingRule{
@@ -213,7 +209,6 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 			&rule1,
 		},
 	}
-	g1.metrics = newGroupMetrics(g1)
 	g.updateCh <- g1
 	time.Sleep(10 * time.Millisecond)
 	g.mu.RLock()
@@ -223,9 +218,8 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 	g.mu.RUnlock()
 
 	rule2 := AlertingRule{
-		RuleID: 1,
-		Name:   "jobDown",
-		Expr:   "up{job=\"vmagent\"}==0",
+		Name: "jobDown",
+		Expr: "up{job=\"vmagent\"}==0",
 		Labels: map[string]string{
 			"foo": "bar",
 			"baz": "qux",
@@ -233,32 +227,17 @@ func TestUpdateDuringRandSleep(t *testing.T) {
 	}
 	g2 := &Group{
 		Rules: []Rule{
-			&rule1,
 			&rule2,
 		},
 	}
-	g2.metrics = newGroupMetrics(g2)
 	g.updateCh <- g2
 	time.Sleep(10 * time.Millisecond)
 	g.mu.RLock()
-	if len(g.Rules) != 2 {
-		t.Fatalf("expected to have updated rules")
-	}
-
-	if len(g.Rules[1].(*AlertingRule).Labels) != 2 {
+	if len(g.Rules[0].(*AlertingRule).Labels) != 2 {
 		t.Fatalf("expected to have updated labels")
 	}
 	g.mu.RUnlock()
 
-	metricsAfter := metrics.GetDefaultSet().ListMetricNames()
-	metricsRegistry := make(map[string]struct{}, len(metricsAfter))
-	for _, m := range metricsAfter {
-		if _, ok := metricsRegistry[m]; ok {
-			t.Fatalf("duplicate metric name %q", m)
-		}
-		metricsRegistry[m] = struct{}{}
-	}
-
 	g.Close()
 }
 
@@ -533,14 +512,34 @@ func TestGroupStartDelay(t *testing.T) {
 	f("2023-01-01T00:00:29.000+00:00", "2023-01-01T00:00:30.000+00:00")
 	f("2023-01-01T00:00:31.000+00:00", "2023-01-01T00:05:30.000+00:00")
 
-	// test group with offset
-	offset := 3 * time.Minute
+	// test group with offset smaller than above fixed randSleep,
+	// this way randSleep will always be enough
+	offset := 20 * time.Second
 	g.EvalOffset = &offset
 
-	f("2023-01-01T00:00:15.000+00:00", "2023-01-01T00:03:00.000+00:00")
-	f("2023-01-01T00:01:00.000+00:00", "2023-01-01T00:03:00.000+00:00")
-	f("2023-01-01T00:03:30.000+00:00", "2023-01-01T00:08:00.000+00:00")
-	f("2023-01-01T00:08:00.000+00:00", "2023-01-01T00:08:00.000+00:00")
+	f("2023-01-01T00:00:00.000+00:00", "2023-01-01T00:00:30.000+00:00")
+	f("2023-01-01T00:00:29.000+00:00", "2023-01-01T00:00:30.000+00:00")
+	f("2023-01-01T00:00:31.000+00:00", "2023-01-01T00:05:30.000+00:00")
+
+	// test group with offset bigger than above fixed randSleep,
+	// this way offset will be added to delay
+	offset = 3 * time.Minute
+	g.EvalOffset = &offset
+
+	f("2023-01-01T00:00:00.000+00:00", "2023-01-01T00:03:30.000+00:00")
+	f("2023-01-01T00:00:29.000+00:00", "2023-01-01T00:03:30.000+00:00")
+	f("2023-01-01T00:01:00.000+00:00", "2023-01-01T00:08:30.000+00:00")
+	f("2023-01-01T00:03:30.000+00:00", "2023-01-01T00:08:30.000+00:00")
+	f("2023-01-01T00:07:30.000+00:00", "2023-01-01T00:13:30.000+00:00")
+
+	offset = 10 * time.Minute
+	g.EvalOffset = &offset
+	// interval of 1h and key generate a static delay of 6m
+	g.Interval = time.Hour
+
+	f("2023-01-01T00:00:00.000+00:00", "2023-01-01T00:16:00.000+00:00")
+	f("2023-01-01T00:05:00.000+00:00", "2023-01-01T00:16:00.000+00:00")
+	f("2023-01-01T00:30:00.000+00:00", "2023-01-01T01:16:00.000+00:00")
 }
 
 func TestGetPrometheusReqTimestamp(t *testing.T) {
@@ -570,11 +569,17 @@ func TestGetPrometheusReqTimestamp(t *testing.T) {
 		evalAlignment: &disableAlign,
 	}, "2023-08-28T11:11:00+00:00", "2023-08-28T11:10:30+00:00")
 
-	// with eval_offset
+	// with eval_offset, find previous offset point + default evalDelay
 	f(&Group{
 		EvalOffset: &offset,
 		Interval:   time.Hour,
-	}, "2023-08-28T11:30:00+00:00", "2023-08-28T11:30:00+00:00")
+	}, "2023-08-28T11:11:00+00:00", "2023-08-28T10:30:00+00:00")
+
+	// with eval_offset + default evalDelay
+	f(&Group{
+		EvalOffset: &offset,
+		Interval:   time.Hour,
+	}, "2023-08-28T11:41:00+00:00", "2023-08-28T11:30:00+00:00")
 
 	// 1h interval with eval_delay
 	f(&Group{

app/vmalert/rule/recording.go

@@ -6,8 +6,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
@@ -47,28 +45,6 @@ type recordingRuleMetrics struct {
 	samples *utils.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),
-		func() float64 {
-			e := rr.state.getLast()
-			return float64(e.Samples)
-		})
-
-	return rmr
-}
-
-func (m *recordingRuleMetrics) close() {
-	if m == nil {
-		return
-	}
-	m.errors.Unregister()
-	m.samples.Unregister()
-}
-
 // String implements Stringer interface
 func (rr *RecordingRule) String() string {
 	return rr.Name
@@ -91,6 +67,7 @@ func NewRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rul
 		GroupID:   group.ID(),
 		GroupName: group.Name,
 		File:      group.File,
+		metrics:   &recordingRuleMetrics{},
 		q: qb.BuildWithParams(datasource.QuerierParams{
 			DataSourceType:            group.Type.String(),
 			ApplyIntervalAsTimeFilter: setIntervalAsTimeFilter(group.Type.String(), cfg.Expr),
@@ -110,18 +87,21 @@ func NewRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rul
 	rr.state = &ruleState{
 		entries: make([]StateEntry, entrySize),
 	}
-	rr.metrics = newRecordingRuleMetrics(group.metrics.set, rr)
 
+	labels := fmt.Sprintf(`recording=%q, group=%q, file=%q, id="%d"`, rr.Name, group.Name, group.File, rr.ID())
+	rr.metrics.errors = utils.GetOrCreateCounter(fmt.Sprintf(`vmalert_recording_rules_errors_total{%s}`, labels))
+	rr.metrics.samples = utils.GetOrCreateGauge(fmt.Sprintf(`vmalert_recording_rules_last_evaluation_samples{%s}`, labels),
+		func() float64 {
+			e := rr.state.getLast()
+			return float64(e.Samples)
+		})
 	return rr
 }
 
-func (rr *RecordingRule) registerMetrics(g *Group) {
-	rr.metrics = newRecordingRuleMetrics(g.metrics.set, rr)
-}
-
 // close unregisters rule metrics
-func (rr *RecordingRule) unregisterMetrics() {
-	rr.metrics.close()
+func (rr *RecordingRule) close() {
+	rr.metrics.errors.Unregister()
+	rr.metrics.samples.Unregister()
 }
 
 // execRange executes recording rule on the given time range similarly to Exec.

app/vmalert/rule/recording_test.go

@@ -7,9 +7,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/metrics"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
@@ -194,7 +193,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 mismatch: %s", err)
+			t.Fatalf("timeseries missmatch: %s", err)
 		}
 	}
 
@@ -307,12 +306,15 @@ func TestRecordingRuleLimit_Failure(t *testing.T) {
 
 		fq := &datasource.FakeQuerier{}
 		fq.Add(testMetrics...)
+
 		rule := &RecordingRule{Name: "job:foo",
 			state: &ruleState{entries: make([]StateEntry, 10)},
 			Labels: map[string]string{
 				"source": "test_limit",
 			},
-			metrics: getTestRecordingRuleMetrics(),
+			metrics: &recordingRuleMetrics{
+				errors: utils.GetOrCreateCounter(`vmalert_recording_rules_errors_total{alertname="job:foo"}`),
+			},
 		}
 		rule.q = fq
 
@@ -342,12 +344,15 @@ func TestRecordingRuleLimit_Success(t *testing.T) {
 
 		fq := &datasource.FakeQuerier{}
 		fq.Add(testMetrics...)
+
 		rule := &RecordingRule{Name: "job:foo",
 			state: &ruleState{entries: make([]StateEntry, 10)},
 			Labels: map[string]string{
 				"source": "test_limit",
 			},
-			metrics: getTestRecordingRuleMetrics(),
+			metrics: &recordingRuleMetrics{
+				errors: utils.GetOrCreateCounter(`vmalert_recording_rules_errors_total{alertname="job:foo"}`),
+			},
 		}
 		rule.q = fq
 
@@ -361,19 +366,16 @@ func TestRecordingRuleLimit_Success(t *testing.T) {
 	f(-1)
 }
 
-func getTestRecordingRuleMetrics() *recordingRuleMetrics {
-	m := newRecordingRuleMetrics(metrics.NewSet(), &RecordingRule{})
-	return m
-}
-
 func TestRecordingRuleExec_Negative(t *testing.T) {
 	rr := &RecordingRule{
 		Name: "job:foo",
 		Labels: map[string]string{
 			"job": "test",
 		},
-		state:   &ruleState{entries: make([]StateEntry, 10)},
-		metrics: getTestRecordingRuleMetrics(),
+		state: &ruleState{entries: make([]StateEntry, 10)},
+		metrics: &recordingRuleMetrics{
+			errors: utils.GetOrCreateCounter(`vmalert_recording_rules_errors_total{alertname="job:foo"}`),
+		},
 	}
 	fq := &datasource.FakeQuerier{}
 	expErr := "connection reset by peer"
@@ -384,7 +386,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 instead", expErr, err)
+		t.Fatalf("expected to get err %q; got %q insterad", expErr, err)
 	}
 
 	fq.Reset()

app/vmalert/rule/rule.go

@@ -27,10 +27,9 @@ type Rule interface {
 	// updateWith performs modification of current Rule
 	// with fields of the given Rule.
 	updateWith(Rule) error
-	// unregister Rule metrics
-	unregisterMetrics()
-	// register Rule metrics with the given group
-	registerMetrics(g *Group)
+	// close performs the shutdown procedures for rule
+	// such as metrics unregister
+	close()
 }
 
 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/templates/template.go

@@ -33,7 +33,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/formatutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
 // go template execution fails when it's tree is empty
@@ -259,7 +259,7 @@ func templateFuncs() textTpl.FuncMap {
 
 		// parseDuration parses a duration string such as "1h" into the number of seconds it represents
 		"parseDuration": func(s string) (float64, error) {
-			d, err := timeutil.ParseDuration(s)
+			d, err := promutils.ParseDuration(s)
 			if err != nil {
 				return 0, err
 			}
@@ -268,7 +268,7 @@ func templateFuncs() textTpl.FuncMap {
 
 		// same with parseDuration but returns a time.Duration
 		"parseDurationTime": func(s string) (time.Duration, error) {
-			d, err := timeutil.ParseDuration(s)
+			d, err := promutils.ParseDuration(s)
 			if err != nil {
 				return 0, err
 			}

app/vmalert/utils/metrics.go

@@ -4,13 +4,11 @@ import "github.com/VictoriaMetrics/metrics"
 
 type namedMetric struct {
 	Name string
-
-	set *metrics.Set
 }
 
 // Unregister removes the metric by name from default registry
 func (nm namedMetric) Unregister() {
-	nm.set.UnregisterMetric(nm.Name)
+	metrics.UnregisterMetric(nm.Name)
 }
 
 // Gauge is a metrics.Gauge with Name
@@ -19,11 +17,11 @@ type Gauge struct {
 	*metrics.Gauge
 }
 
-// NewGauge creates a new Gauge with the given name
-func NewGauge(set *metrics.Set, name string, f func() float64) *Gauge {
+// GetOrCreateGauge creates a new Gauge with the given name
+func GetOrCreateGauge(name string, f func() float64) *Gauge {
 	return &Gauge{
-		namedMetric: namedMetric{Name: name, set: set},
-		Gauge:       set.NewGauge(name, f),
+		namedMetric: namedMetric{Name: name},
+		Gauge:       metrics.GetOrCreateGauge(name, f),
 	}
 }
 
@@ -33,11 +31,11 @@ type Counter struct {
 	*metrics.Counter
 }
 
-// NewCounter creates a new Counter with the given name
-func NewCounter(set *metrics.Set, name string) *Counter {
+// GetOrCreateCounter creates a new Counter with the given name
+func GetOrCreateCounter(name string) *Counter {
 	return &Counter{
-		namedMetric: namedMetric{Name: name, set: set},
-		Counter:     set.NewCounter(name),
+		namedMetric: namedMetric{Name: name},
+		Counter:     metrics.GetOrCreateCounter(name),
 	}
 }
 
@@ -47,10 +45,10 @@ type Summary struct {
 	*metrics.Summary
 }
 
-// NewSummary creates a new Summary with the given name
-func NewSummary(set *metrics.Set, name string) *Summary {
+// GetOrCreateSummary creates a new Summary with the given name
+func GetOrCreateSummary(name string) *Summary {
 	return &Summary{
-		namedMetric: namedMetric{Name: name, set: set},
-		Summary:     set.NewSummary(name),
+		namedMetric: namedMetric{Name: name},
+		Summary:     metrics.GetOrCreateSummary(name),
 	}
 }

app/vmalert/web.qtpl

@@ -97,7 +97,7 @@ btn-primary
                     <a href="#group-{%s g.ID %}">{%s g.Name %}{% if g.Type != "prometheus" %} ({%s g.Type %}){% endif %} (every {%f.0 g.Interval %}s) #</a>
                      {% if rNotOk[g.ID] > 0 %}<span class="badge bg-danger" title="Number of rules with status Error">{%d rNotOk[g.ID] %}</span> {% endif %}
                      {% if rNoMatch[g.ID] > 0 %}<span class="badge bg-warning" title="Number of rules with status NoMatch">{%d rNoMatch[g.ID] %}</span> {% endif %}
-                    <span class="badge bg-success" title="Number of rules with status Ok">{%d rOk[g.ID] %}</span>
+                    <span class="badge bg-success" title="Number of rules withs status Ok">{%d rOk[g.ID] %}</span>
                     <p class="fs-6 fw-lighter">{%s g.File %}</p>
                     {% if len(g.Params) > 0 %}
                         <div class="fs-6 fw-lighter">Extra params

app/vmalert/web.qtpl.go

@@ -355,7 +355,7 @@ func StreamListGroups(qw422016 *qt422016.Writer, r *http.Request, originGroups [
 			}
 //line app/vmalert/web.qtpl:99
 			qw422016.N().S(`
-                    <span class="badge bg-success" title="Number of rules with status Ok">`)
+                    <span class="badge bg-success" title="Number of rules withs status Ok">`)
 //line app/vmalert/web.qtpl:100
 			qw422016.N().D(rOk[g.ID])
 //line app/vmalert/web.qtpl:100

app/vmalert/web_test.go

@@ -21,18 +21,14 @@ func TestHandler(t *testing.T) {
 	fq.Add(datasource.Metric{
 		Values: []float64{1}, Timestamps: []int64{0},
 	})
-	g := rule.NewGroup(config.Group{
+	g := &rule.Group{
 		Name:        "group",
 		File:        "rules.yaml",
 		Concurrency: 1,
-		Rules: []config.Rule{
-			{ID: 0, Alert: "alert"},
-			{ID: 1, Record: "record"},
-		},
-	}, fq, 1*time.Minute, nil)
-	ar := g.Rules[0].(*rule.AlertingRule)
-	rr := g.Rules[1].(*rule.RecordingRule)
-
+	}
+	ar := rule.NewAlertingRule(fq, g, config.Rule{ID: 0, Alert: "alert"})
+	rr := rule.NewRecordingRule(fq, g, config.Rule{ID: 1, Record: "record"})
+	g.Rules = []rule.Rule{ar, rr}
 	g.ExecOnce(context.Background(), func() []notifier.Notifier { return nil }, nil, time.Time{})
 
 	m := &manager{groups: map[uint64]*rule.Group{
@@ -166,7 +162,7 @@ func TestHandler(t *testing.T) {
 
 		gotRuleWithUpdates := apiRuleWithUpdates{}
 		getResp(t, ts.URL+"/"+expRule.APILink(), &gotRuleWithUpdates, 200)
-		if len(gotRuleWithUpdates.StateUpdates) < 1 {
+		if gotRuleWithUpdates.StateUpdates == nil || len(gotRuleWithUpdates.StateUpdates) < 1 {
 			t.Fatalf("expected %+v to have state updates field not empty", gotRuleWithUpdates.StateUpdates)
 		}
 	})

app/vmalert/web_types_test.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"reflect"
 	"testing"
-	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
@@ -16,22 +15,20 @@ func TestRecordingToApi(t *testing.T) {
 	fq.Add(datasource.Metric{
 		Values: []float64{1}, Timestamps: []int64{0},
 	})
-	entriesLimit := 44
-	g := rule.NewGroup(config.Group{
+	g := &rule.Group{
 		Name:        "group",
 		File:        "rules.yaml",
 		Concurrency: 1,
-		Rules: []config.Rule{
-			{
-				ID:                 1248,
-				Record:             "record_name",
-				Expr:               "up",
-				Labels:             map[string]string{"label": "value"},
-				UpdateEntriesLimit: &entriesLimit,
-			},
-		},
-	}, fq, 1*time.Minute, nil)
-	rr := g.Rules[0].(*rule.RecordingRule)
+	}
+
+	entriesLimit := 44
+	rr := rule.NewRecordingRule(fq, g, config.Rule{
+		ID:                 1248,
+		Record:             "record_name",
+		Expr:               "up",
+		Labels:             map[string]string{"label": "value"},
+		UpdateEntriesLimit: &entriesLimit,
+	})
 
 	expectedRes := apiRule{
 		Name:           "record_name",

app/vmauth/auth_config.go

@@ -7,13 +7,11 @@ import (
 	"flag"
 	"fmt"
 	"math"
-	"net"
 	"net/http"
 	"net/url"
 	"os"
 	"regexp"
 	"sort"
-	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -141,7 +139,7 @@ func (h *Header) UnmarshalYAML(f func(any) error) error {
 
 	n := strings.IndexByte(s, ':')
 	if n < 0 {
-		return fmt.Errorf("missing separator char ':' between Name and Value in the header %q; expected format - 'Name: Value'", s)
+		return fmt.Errorf("missing speparator char ':' between Name and Value in the header %q; expected format - 'Name: Value'", s)
 	}
 	h.Name = strings.TrimSpace(s[:n])
 	h.Value = strings.TrimSpace(s[n+1:])
@@ -173,7 +171,7 @@ type URLMap struct {
 	// DiscoverBackendIPs instructs discovering URLPrefix backend IPs via DNS.
 	DiscoverBackendIPs *bool `yaml:"discover_backend_ips,omitempty"`
 
-	// HeadersConf is the config for augmenting request and response headers.
+	// HeadersConf is the config for augumenting request and response headers.
 	HeadersConf HeadersConf `yaml:",inline"`
 
 	// RetryStatusCodes is the list of response status codes used for retrying requests.
@@ -350,7 +348,6 @@ func (up *URLPrefix) discoverBackendAddrsIfNeeded() {
 	hostToAddrs := make(map[string][]string)
 	for _, bu := range up.busOriginal {
 		host := bu.Hostname()
-		port := bu.Port()
 		if hostToAddrs[host] != nil {
 			// ips for the given host have been already discovered
 			continue
@@ -367,11 +364,7 @@ func (up *URLPrefix) discoverBackendAddrsIfNeeded() {
 			} else {
 				resolvedAddrs = make([]string, len(addrs))
 				for i, addr := range addrs {
-					hostPort := port
-					if hostPort == "" && addr.Port > 0 {
-						hostPort = strconv.FormatUint(uint64(addr.Port), 10)
-					}
-					resolvedAddrs[i] = net.JoinHostPort(addr.Target, hostPort)
+					resolvedAddrs[i] = fmt.Sprintf("%s:%d", addr.Target, addr.Port)
 				}
 			}
 		} else {
@@ -382,7 +375,7 @@ func (up *URLPrefix) discoverBackendAddrsIfNeeded() {
 			} else {
 				resolvedAddrs = make([]string, len(addrs))
 				for i, addr := range addrs {
-					resolvedAddrs[i] = net.JoinHostPort(addr.String(), port)
+					resolvedAddrs[i] = addr.String()
 				}
 			}
 		}
@@ -396,9 +389,17 @@ func (up *URLPrefix) discoverBackendAddrsIfNeeded() {
 	var busNew []*backendURL
 	for _, bu := range up.busOriginal {
 		host := bu.Hostname()
+		port := bu.Port()
 		for _, addr := range hostToAddrs[host] {
 			buCopy := *bu
 			buCopy.Host = addr
+			if port != "" {
+				if n := strings.IndexByte(buCopy.Host, ':'); n >= 0 {
+					// Drop the discovered port and substitute it the port specified in bu.
+					buCopy.Host = buCopy.Host[:n]
+				}
+				buCopy.Host += ":" + port
+			}
 			busNew = append(busNew, &backendURL{
 				url: &buCopy,
 			})
@@ -438,7 +439,7 @@ func getFirstAvailableBackendURL(bus []*backendURL) *backendURL {
 		return bu
 	}
 
-	// Slow path - the first url is temporarily unavailable. Fall back to the remaining urls.
+	// Slow path - the first url is temporarily unavailabel. Fall back to the remaining urls.
 	for i := 1; i < len(bus); i++ {
 		if !bus[i].isBroken() {
 			bu = bus[i]

app/vmauth/auth_config_test.go

@@ -3,14 +3,12 @@ package main
 import (
 	"bytes"
 	"fmt"
-	"net"
 	"net/url"
 	"testing"
 
 	"gopkg.in/yaml.v2"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 )
 
 func TestParseAuthConfigFailure(t *testing.T) {
@@ -801,75 +799,6 @@ func TestBrokenBackend(t *testing.T) {
 	}
 }
 
-func TestDiscoverBackendIPsWithIPV6(t *testing.T) {
-	f := func(actualUrl, expectedUrl string) {
-		t.Helper()
-		up := mustParseURL(actualUrl)
-		up.discoverBackendIPs = true
-		up.loadBalancingPolicy = "least_loaded"
-
-		up.discoverBackendAddrsIfNeeded()
-		pbus := up.bus.Load()
-		bus := *pbus
-
-		if len(bus) != 1 {
-			t.Fatalf("expected url list to be of size 1; got %d instead", len(bus))
-		}
-
-		got := bus[0].url.Host
-		if got != expectedUrl {
-			t.Fatalf(`expected url to be %q; got %q instead`, expectedUrl, bus[0].url.Host)
-		}
-	}
-
-	// Discover backendURL with SRV hostnames
-	customResolver := &fakeResolver{
-		Resolver: &net.Resolver{},
-		// SRV records must return hostname
-		// not an IP address
-		lookupSRVResults: map[string][]*net.SRV{
-			"_vmselect._tcp.selectwithport.": {
-				{
-					Target: "vmselect.local",
-					Port:   8481,
-				},
-			},
-			"_vmselect._tcp.selectwoport.": {
-				{
-					Target: "vmselect.local",
-				},
-			},
-		},
-		lookupIPAddrResults: map[string][]net.IPAddr{
-			"vminsert.local": {
-				{
-					IP: net.ParseIP("10.0.10.13"),
-				},
-			},
-			"ipv6.vminsert.local": {
-				{
-					IP: net.ParseIP("2607:f8b0:400a:80b::200e"),
-				},
-			},
-		},
-	}
-	origResolver := netutil.Resolver
-	netutil.Resolver = customResolver
-	defer func() {
-		netutil.Resolver = origResolver
-	}()
-	f("http://srv+_vmselect._tcp.selectwithport.:8080", "vmselect.local:8080")
-	f("http://srv+_vmselect._tcp.selectwithport.:", "vmselect.local:8481")
-	f("http://srv+_vmselect._tcp.selectwoport.:8080", "vmselect.local:8080")
-	f("http://srv+_vmselect._tcp.selectwoport.", "vmselect.local:")
-
-	f("http://vminsert.local:8080", "10.0.10.13:8080")
-	f("http://vminsert.local", "10.0.10.13:")
-	f("http://ipv6.vminsert.local:8080", "[2607:f8b0:400a:80b::200e]:8080")
-	f("http://ipv6.vminsert.local", "[2607:f8b0:400a:80b::200e]:")
-
-}
-
 func getRegexs(paths []string) []*Regex {
 	var sps []*Regex
 	for _, path := range paths {

app/vmauth/main.go

@@ -31,11 +31,7 @@ import (
 )
 
 var (
-	httpListenAddrs = flagutil.NewArrayString("httpListenAddr", "TCP address to listen for incoming http requests. "+
-		"By default, serves internal API and proxy requests. "+
-		" See also -tls, -httpListenAddr.useProxyProtocol and -httpInternalListenAddr.")
-	httpInternalListenAddr = flagutil.NewArrayString("httpInternalListenAddr", "TCP address to listen for incoming internal API http requests. Such as /health, /-/reload, /debug/pprof, etc. "+
-		"If flag is set, vmauth no longer serves internal API at -httpListenAddr.")
+	httpListenAddrs  = flagutil.NewArrayString("httpListenAddr", "TCP address to listen for incoming http requests. See also -tls and -httpListenAddr.useProxyProtocol")
 	useProxyProtocol = flagutil.NewArrayBool("httpListenAddr.useProxyProtocol", "Whether to use proxy protocol for connections accepted at the corresponding -httpListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt . "+
 		"With enabled proxy protocol http server cannot serve regular /metrics endpoint. Use -pushmetrics.url for metrics pushing")
@@ -95,21 +91,7 @@ func main() {
 	logger.Infof("starting vmauth at %q...", listenAddrs)
 	startTime := time.Now()
 	initAuthConfig()
-	disableInternalRoutes := len(*httpInternalListenAddr) > 0
-	rh := requestHandlerWithInternalRoutes
-	if disableInternalRoutes {
-		rh = requestHandler
-	}
-
-	serveOpts := httpserver.ServeOptions{
-		UseProxyProtocol:     useProxyProtocol,
-		DisableBuiltinRoutes: disableInternalRoutes,
-	}
-	go httpserver.ServeWithOpts(listenAddrs, rh, serveOpts)
-
-	if len(*httpInternalListenAddr) > 0 {
-		go httpserver.Serve(*httpInternalListenAddr, nil, internalRequestHandler)
-	}
+	go httpserver.Serve(listenAddrs, useProxyProtocol, requestHandler)
 	logger.Infof("started vmauth in %.3f seconds", time.Since(startTime).Seconds())
 
 	pushmetrics.Init()
@@ -127,7 +109,7 @@ func main() {
 	logger.Infof("successfully stopped vmauth in %.3f seconds", time.Since(startTime).Seconds())
 }
 
-func internalRequestHandler(w http.ResponseWriter, r *http.Request) bool {
+func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 	switch r.URL.Path {
 	case "/-/reload":
 		if !httpserver.CheckAuthFlag(w, r, reloadAuthKey) {
@@ -138,17 +120,6 @@ func internalRequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		w.WriteHeader(http.StatusOK)
 		return true
 	}
-	return false
-}
-
-func requestHandlerWithInternalRoutes(w http.ResponseWriter, r *http.Request) bool {
-	if internalRequestHandler(w, r) {
-		return true
-	}
-	return requestHandler(w, r)
-}
-
-func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 
 	ats := getAuthTokensFromRequest(r)
 	if len(ats) == 0 {
@@ -251,7 +222,8 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 		isDefault = true
 	}
 
-	rtb := newReadTrackingBody(r.Body, maxRequestBodySizeToRetry.IntN())
+	rtb := getReadTrackingBody(r.Body, maxRequestBodySizeToRetry.IntN())
+	defer putReadTrackingBody(rtb)
 	r.Body = rtb
 
 	maxAttempts := up.getBackendsCount()
@@ -587,11 +559,22 @@ type readTrackingBody struct {
 	bufComplete bool
 }
 
-func newReadTrackingBody(r io.ReadCloser, maxBodySize int) *readTrackingBody {
-	// do not use sync.Pool there
-	// since http.RoundTrip may still use request body after return
-	// See this issue for details https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8051
-	rtb := &readTrackingBody{}
+func (rtb *readTrackingBody) reset() {
+	rtb.maxBodySize = 0
+	rtb.r = nil
+	rtb.buf = rtb.buf[:0]
+	rtb.readBuf = nil
+	rtb.cannotRetry = false
+	rtb.bufComplete = false
+}
+
+func getReadTrackingBody(r io.ReadCloser, maxBodySize int) *readTrackingBody {
+	v := readTrackingBodyPool.Get()
+	if v == nil {
+		v = &readTrackingBody{}
+	}
+	rtb := v.(*readTrackingBody)
+
 	if maxBodySize < 0 {
 		maxBodySize = 0
 	}
@@ -614,6 +597,13 @@ func (r *zeroReader) Close() error {
 	return nil
 }
 
+func putReadTrackingBody(rtb *readTrackingBody) {
+	rtb.reset()
+	readTrackingBodyPool.Put(rtb)
+}
+
+var readTrackingBodyPool sync.Pool
+
 // Read implements io.Reader interface.
 func (rtb *readTrackingBody) Read(p []byte) (int, error) {
 	if len(rtb.readBuf) > 0 {

app/vmauth/main_test.go

@@ -52,7 +52,7 @@ func TestRequestHandler(t *testing.T) {
 		r.Header.Set("Pass-Header", "abc")
 
 		w := &fakeResponseWriter{}
-		if !requestHandlerWithInternalRoutes(w, r) {
+		if !requestHandler(w, r) {
 			t.Fatalf("unexpected false is returned from requestHandler")
 		}
 
@@ -195,7 +195,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=401
-Expected to receive non-empty authKey when -reloadAuthKey is set`
+The provided authKey doesn't match -reloadAuthKey`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 	if err := reloadAuthKey.Set(origAuthKey); err != nil {
 		t.Fatalf("unexpected error: %s", err)
@@ -545,7 +545,8 @@ func TestReadTrackingBody_RetrySuccess(t *testing.T) {
 	f := func(s string, maxBodySize int) {
 		t.Helper()
 
-		rtb := newReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		rtb := getReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		defer putReadTrackingBody(rtb)
 
 		if !rtb.canRetry() {
 			t.Fatalf("canRetry() must return true before reading anything")
@@ -580,7 +581,8 @@ func TestReadTrackingBody_RetrySuccessPartialRead(t *testing.T) {
 		t.Helper()
 
 		// Check the case with partial read
-		rtb := newReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		rtb := getReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		defer putReadTrackingBody(rtb)
 
 		for i := 0; i < len(s); i++ {
 			buf := make([]byte, i)
@@ -629,7 +631,8 @@ func TestReadTrackingBody_RetryFailureTooBigBody(t *testing.T) {
 	f := func(s string, maxBodySize int) {
 		t.Helper()
 
-		rtb := newReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		rtb := getReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		defer putReadTrackingBody(rtb)
 
 		if !rtb.canRetry() {
 			t.Fatalf("canRetry() must return true before reading anything")
@@ -678,7 +681,8 @@ func TestReadTrackingBody_RetryFailureZeroOrNegativeMaxBodySize(t *testing.T) {
 	f := func(s string, maxBodySize int) {
 		t.Helper()
 
-		rtb := newReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		rtb := getReadTrackingBody(io.NopCloser(bytes.NewBufferString(s)), maxBodySize)
+		defer putReadTrackingBody(rtb)
 
 		if !rtb.canRetry() {
 			t.Fatalf("canRetry() must return true before reading anything")

app/vmauth/target_url_test.go

@@ -98,7 +98,6 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 		up, hc := ui.getURLPrefixAndHeaders(u, u.Host, nil)
 		if up == nil {
 			t.Fatalf("cannot match available backend: %s", err)
-			return
 		}
 		bu := up.getBackendURL()
 		target := mergeURLs(bu.url, u, up.dropSrcPathPrefixParts)
@@ -310,7 +309,6 @@ func TestUserInfoGetBackendURL_SRV(t *testing.T) {
 		up, _ := ui.getURLPrefixAndHeaders(u, u.Host, nil)
 		if up == nil {
 			t.Fatalf("cannot match available backend: %s", err)
-			return
 		}
 		bu := up.getBackendURL()
 		target := mergeURLs(bu.url, u, up.dropSrcPathPrefixParts)

app/vmctl/flags.go

@@ -344,7 +344,7 @@ var (
 		},
 		&cli.BoolFlag{
 			Name:  influxSkipDatabaseLabel,
-			Usage: "Whether to skip adding the label 'db' to timeseries.",
+			Usage: "Wether to skip adding the label 'db' to timeseries.",
 			Value: false,
 		},
 		&cli.BoolFlag{
@@ -596,8 +596,7 @@ var (
 		&cli.Int64Flag{
 			Name: vmRateLimit,
 			Usage: "Optional data transfer rate limit in bytes per second.\n" +
-				"By default, the rate limit is disabled. It can be useful for limiting load on source or destination databases. \n" +
-				"Rate limit is applied per worker, see `--vm-concurrency`.",
+				"By default, the rate limit is disabled. It can be useful for limiting load on source or destination databases.",
 		},
 		&cli.BoolFlag{
 			Name: vmInterCluster,

app/vmctl/influx/influx.go

@@ -144,9 +144,9 @@ func timeFilter(start, end string) string {
 // by checking available (non-empty) tags, fields and measurements
 // which unique combination represents all possible
 // time series existing in database.
-// Explore is required to reduce the load on influx
+// The explore required to reduce the load on influx
 // by querying field of the exact time series at once,
-// instead of fetching all the values over and over.
+// instead of fetching all of the values over and over.
 //
 // May contain non-existing time series.
 func (c *Client) Explore() ([]*Series, error) {
@@ -180,7 +180,11 @@ func (c *Client) Explore() ([]*Series, error) {
 			log.Printf("skip measurement %q since it has no fields", s.Measurement)
 			continue
 		}
-		emptyTags := getEmptyTags(measurementTags[s.Measurement], s.LabelPairs)
+		tags, ok := measurementTags[s.Measurement]
+		if !ok {
+			return nil, fmt.Errorf("failed to find tags of measurement %s", s.Measurement)
+		}
+		emptyTags := getEmptyTags(tags, s.LabelPairs)
 		for _, field := range fields {
 			is := &Series{
 				Measurement: s.Measurement,
@@ -197,16 +201,11 @@ func (c *Client) Explore() ([]*Series, error) {
 // getEmptyTags returns tags of a measurement that are missing in a specific series.
 // Tags represent all tags of a measurement. LabelPairs represent tags of a specific series.
 func getEmptyTags(tags map[string]struct{}, LabelPairs []LabelPair) []string {
-	if len(tags) == 0 {
-		// fast path: the measurement does not contain any tag
-		return nil
-	}
-
 	labelMap := make(map[string]struct{})
 	for _, pair := range LabelPairs {
 		labelMap[pair.Name] = struct{}{}
 	}
-	var result []string
+	result := make([]string, 0, len(labelMap)-len(LabelPairs))
 	for tag := range tags {
 		if _, ok := labelMap[tag]; !ok {
 			result = append(result, tag)
@@ -340,7 +339,10 @@ func (c *Client) fieldsByMeasurement() (map[string][]string, error) {
 }
 
 func (c *Client) getSeries() ([]*Series, error) {
-	com := c.getSeriesCommand()
+	com := "show series"
+	if c.filterSeries != "" {
+		com = fmt.Sprintf("%s %s", com, c.filterSeries)
+	}
 	q := influx.Query{
 		Command:         com,
 		Database:        c.database,
@@ -386,21 +388,6 @@ func (c *Client) getSeries() ([]*Series, error) {
 	return result, nil
 }
 
-func (c *Client) getSeriesCommand() string {
-	com := "show series"
-	if c.filterSeries != "" {
-		com = fmt.Sprintf("%s %s", com, c.filterSeries)
-	}
-	if c.filterTime != "" {
-		joinStatement := " where "
-		if strings.Contains(strings.ToLower(com), joinStatement) {
-			joinStatement = " AND "
-		}
-		com = fmt.Sprintf("%s%s%s", com, joinStatement, c.filterTime)
-	}
-	return com
-}
-
 // getMeasurementTags get the tags for each measurement.
 // tags are placed in a map without values (similar to a set) for quick lookups:
 // {"measurement1": {"tag1", "tag2"}, "measurement2": {"tag3", "tag4"}}

app/vmctl/influx/influx_test.go

@@ -103,26 +103,3 @@ func TestTimeFilter(t *testing.T) {
 	// both start and end filters
 	f("2020-01-01T20:07:00Z", "2020-01-01T21:07:00Z", "time >= '2020-01-01T20:07:00Z' and time <= '2020-01-01T21:07:00Z'")
 }
-
-func TestGetSeriesCommand(t *testing.T) {
-	f := func(filterSeries, filterTime, expCommand string) {
-		t.Helper()
-
-		c := &Client{
-			filterTime:   filterTime,
-			filterSeries: filterSeries,
-		}
-		gotCommand := c.getSeriesCommand()
-		if gotCommand != expCommand {
-			t.Fatalf("unexpected command\ngot\n%s\nwant\n%s", gotCommand, expCommand)
-		}
-	}
-
-	f("", "", "show series")
-	f("from cpu", "", "show series from cpu")
-	f("from cpu where arch='x86'", "", "show series from cpu where arch='x86'")
-	f("", "time >= '2020-01-01T20:07:00Z'", "show series where time >= '2020-01-01T20:07:00Z'")
-	f("from cpu", "time >= '2020-01-01T20:07:00Z'", "show series from cpu where time >= '2020-01-01T20:07:00Z'")
-	f("from cpu where arch='x86'", "time >= '2020-01-01T20:07:00Z'", "show series from cpu where arch='x86' AND time >= '2020-01-01T20:07:00Z'")
-	f("from cpu where arch='x86' AND hostname='host_2753'", "time >= '2020-01-01T20:07:00Z'", "show series from cpu where arch='x86' AND hostname='host_2753' AND time >= '2020-01-01T20:07:00Z'")
-}

app/vmctl/main.go

@@ -293,7 +293,7 @@ func main() {
 						auth.WithBearer(c.String(vmNativeSrcBearerToken)),
 						auth.WithHeaders(c.String(vmNativeSrcHeaders)))
 					if err != nil {
-						return fmt.Errorf("error initialize auth config for source: %s", srcAddr)
+						return fmt.Errorf("error initilize auth config for source: %s", srcAddr)
 					}
 
 					// create TLS config
@@ -320,7 +320,7 @@ func main() {
 						auth.WithBearer(c.String(vmNativeDstBearerToken)),
 						auth.WithHeaders(c.String(vmNativeDstHeaders)))
 					if err != nil {
-						return fmt.Errorf("error initialize auth config for destination: %s", dstAddr)
+						return fmt.Errorf("error initilize auth config for destination: %s", dstAddr)
 					}
 
 					// create TLS config
@@ -384,10 +384,7 @@ func main() {
 				Action: func(c *cli.Context) error {
 					common.StartUnmarshalWorkers()
 					blockPath := c.Args().First()
-					encoding := ""
-					if c.Bool("gunzip") {
-						encoding = "gzip"
-					}
+					isBlockGzipped := c.Bool("gunzip")
 					if len(blockPath) == 0 {
 						return cli.Exit("you must provide path for exported data block", 1)
 					}
@@ -398,7 +395,7 @@ func main() {
 					}
 					defer f.Close()
 					var blocksCount atomic.Uint64
-					if err := stream.Parse(f, encoding, func(_ *stream.Block) error {
+					if err := stream.Parse(f, isBlockGzipped, func(_ *stream.Block) error {
 						blocksCount.Add(1)
 						return nil
 					}); err != nil {

app/vmctl/prometheus/prometheus.go

@@ -40,15 +40,15 @@ type filter struct {
 	labelValue string
 }
 
-func (f filter) inRange(minV, maxV int64) bool {
+func (f filter) inRange(min, max int64) bool {
 	fmin, fmax := f.min, f.max
-	if minV == 0 {
-		fmin = minV
+	if min == 0 {
+		fmin = min
 	}
 	if fmax == 0 {
-		fmax = maxV
+		fmax = max
 	}
-	return minV <= fmax && fmin <= maxV
+	return min <= fmax && fmin <= max
 }
 
 // NewClient creates and validates new Client
@@ -59,13 +59,13 @@ func NewClient(cfg Config) (*Client, error) {
 		return nil, fmt.Errorf("failed to open snapshot %q: %s", cfg.Snapshot, err)
 	}
 	c := &Client{DBReadOnly: db}
-	minTime, maxTime, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
+	min, max, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse time in filter: %s", err)
 	}
 	c.filter = filter{
-		min:        minTime,
-		max:        maxTime,
+		min:        min,
+		max:        max,
 		label:      cfg.Filter.Label,
 		labelValue: cfg.Filter.LabelValue,
 	}