apptest: add an apptest for vmselect retry

Signed-off-by: Artem Fetishev <rtm@victoriametrics.com>

.github/copilot-instructions.md

@@ -1,23 +0,0 @@
-# Project Overview
-
-VictoriaMetrics is a fast, cost-saving, and scalable solution for monitoring and managing time series data. It delivers high performance and reliability, making it an ideal choice for businesses of all sizes.
-
-## Folder Structure
-
-- `/app`: Contains the compilable binaries.
-- `/lib`: Contains the golang reusable libraries
-- `/docs/victoriametrics`: Contains documentation for the project.
-- `/apptest/tests`: Contains integration tests.
-
-## Libraries and Frameworks
-
-- Backend: Golang, no framework. Use third-party libraries sparingly.
-- Frontend: React.
-
-## Code review guidelines
-
-Ensure the feature or bugfix includes a changelog entry in /docs/victoriametrics/changelog/CHANGELOG.md.
-Verify the entry is under the ## tip section and matches the structure and style of existing entries.
-Chore-only changes may be omitted from the changelog.
-
-

.github/workflows/build.yml

@@ -65,7 +65,8 @@ jobs:
             go.sum
             Makefile
             app/**/Makefile
-          go-version: stable
+          go-version-file: 'go.mod'
+      - run: go version
 
       - name: Build vmcluster for ${{ matrix.os }}-${{ matrix.arch }}
         run: make vmcluster-${{ matrix.os }}-${{ matrix.arch }}

.github/workflows/check-licenses.yml

@@ -21,9 +21,11 @@ jobs:
         id: go
         uses: actions/setup-go@v6
         with:
-          go-version: stable
+          go-version-file: 'go.mod'
           cache: false
 
+      - run: go version
+
       - name: Cache Go artifacts
         uses: actions/cache@v4
         with:
@@ -32,7 +34,7 @@ jobs:
             ~/go/pkg/mod
             ~/go/bin
           key: go-artifacts-${{ runner.os }}-check-licenses-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-check-licenses-
+          restore-keys: go-artifacts-${{ runner.os }}-check-licenses-${{ steps.go.outputs.go-version }}-
 
       - name: Check License
         run: make check-licenses

.github/workflows/codeql-analysis-go.yml

@@ -36,7 +36,8 @@ jobs:
         uses: actions/setup-go@v6
         with:
           cache: false
-          go-version: stable
+          go-version-file: 'go.mod'
+      - run: go version
 
       - name: Cache Go artifacts
         uses: actions/cache@v4
@@ -46,7 +47,7 @@ jobs:
             ~/go/bin
             ~/go/pkg/mod
           key: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-${{ hashFiles('go.sum', 'Makefile', 'app/**/Makefile') }}
-          restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-
+          restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-
 
       - name: Initialize CodeQL
         uses: github/codeql-action/init@v4

.github/workflows/docs.yaml

@@ -28,7 +28,7 @@ jobs:
           path: __vm-docs
 
       - name: Import GPG key
-        uses: crazy-max/ghaction-import-gpg@v6
+        uses: crazy-max/ghaction-import-gpg@v7
         id: import-gpg
         with:
           gpg_private_key: ${{ secrets.VM_BOT_GPG_PRIVATE_KEY }}

.github/workflows/test.yml

@@ -42,8 +42,9 @@ jobs:
             go.sum
             Makefile
             app/**/Makefile
-          go-version: stable
+          go-version-file: 'go.mod'
 
+      - run: go version
 
       - name: Cache golangci-lint
         uses: actions/cache@v4
@@ -51,7 +52,7 @@ jobs:
           path: |
             ~/.cache/golangci-lint
             ~/go/bin
-          key: golangci-lint-${{ runner.os }}-${{ hashFiles('.golangci.yml') }}
+          key: golangci-lint-${{ runner.os }}-${{ steps.go.outputs.go-version }}-${{ hashFiles('.golangci.yml') }}
 
       - name: Run check-all
         run: |
@@ -81,19 +82,20 @@ jobs:
             go.sum
             Makefile
             app/**/Makefile
-          go-version: stable
+          go-version-file: 'go.mod'
+      - run: go version
 
       - 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
 
-  integration:
-    name: integration
-    runs-on: ubuntu-latest
+  apptest:
+    name: apptest
+    runs-on: apptest
 
     steps:
       - name: Code checkout
@@ -107,7 +109,8 @@ jobs:
             go.sum
             Makefile
             app/**/Makefile
-          go-version: stable
+          go-version-file: 'go.mod'
+      - run: go version
 
-      - name: Run integration tests
-        run: make integration-test
+      - name: Run app tests
+        run: make apptest

.github/workflows/vmui.yml

@@ -34,33 +34,39 @@ jobs:
       - name: Code checkout
         uses: actions/checkout@v6
 
-      - name: Setup Node
-        uses: actions/setup-node@v6
+      - name: Cache node_modules
+        id: cache
+        uses: actions/cache@v5
         with:
-          node-version: '24.x'
+          path: app/vmui/packages/vmui/node_modules
+          key: vmui-deps-${{ runner.os }}-${{ hashFiles('app/vmui/packages/vmui/package-lock.json', 'app/vmui/Dockerfile-build') }}
+          restore-keys: |
+            vmui-deps-${{ runner.os }}-
 
-      - name: Cache node-modules
-        uses: actions/cache@v4
-        with:
-          path: |
-            app/vmui/packages/vmui/node_modules
-          key: vmui-artifacts-${{ runner.os }}-${{ hashFiles('package-lock.json') }}
-          restore-keys: vmui-artifacts-${{ runner.os }}-
+      - name: Install dependencies
+        if: steps.cache.outputs.cache-hit != 'true'
+        run: make vmui-install
 
       - name: Run lint
         id: lint
         run: make vmui-lint
         continue-on-error: true
+        env:
+          VMUI_SKIP_INSTALL: true
 
       - name: Run tests
         id: test
         run: make vmui-test
         continue-on-error: true
+        env:
+          VMUI_SKIP_INSTALL: true
 
       - name: Run typecheck
         id: typecheck
         run: make vmui-typecheck
         continue-on-error: true
+        env:
+          VMUI_SKIP_INSTALL: true
 
       - name: Annotate Code Linting Results
         uses: ataylorme/eslint-annotate-action@v3

LICENSE

@@ -175,7 +175,7 @@
 
    END OF TERMS AND CONDITIONS
 
-   Copyright 2019-2025 VictoriaMetrics, Inc.
+   Copyright 2019-2026 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

@@ -17,7 +17,7 @@ EXTRA_GO_BUILD_TAGS ?=
 GO_BUILDINFO = -X '$(PKG_PREFIX)/lib/buildinfo.Version=$(APP_NAME)-$(DATEINFO_TAG)-$(BUILDINFO_TAG)'
 TAR_OWNERSHIP ?= --owner=1000 --group=1000
 
-GOLANGCI_LINT_VERSION := 2.7.2
+GOLANGCI_LINT_VERSION := 2.9.0
 
 .PHONY: $(MAKECMDGOALS)
 
@@ -251,7 +251,7 @@ fmt:
 	gofmt -l -w -s ./apptest
 
 vet:
-	GOEXPERIMENT=synctest go vet ./lib/...
+	go vet -tags 'synctest' ./lib/...
 	go vet ./app/...
 	go vet ./apptest/...
 
@@ -260,28 +260,25 @@ check-all: fmt vet golangci-lint govulncheck
 clean-checkers: remove-golangci-lint remove-govulncheck
 
 test:
-	GOEXPERIMENT=synctest go test ./lib/... ./app/...
+	go test -tags 'synctest' ./lib/... ./app/...
 
 test-race:
-	GOEXPERIMENT=synctest go test -race ./lib/... ./app/...
+	go test -tags 'synctest' -race ./lib/... ./app/...
 
 test-pure:
-	GOEXPERIMENT=synctest CGO_ENABLED=0 go test ./lib/... ./app/...
+	CGO_ENABLED=0 go test -tags 'synctest' ./lib/... ./app/...
 
 test-full:
-	GOEXPERIMENT=synctest go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
+	go test -tags 'synctest' -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
 test-full-386:
-	GOEXPERIMENT=synctest GOARCH=386 go test -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
-
-integration-test:
-	$(MAKE) apptest
+	GOARCH=386 go test -tags 'synctest' -coverprofile=coverage.txt -covermode=atomic ./lib/... ./app/...
 
 apptest:
 	$(MAKE) all vmctl vmbackup vmrestore
 	go test ./apptest/... -skip="^Test(Single|Legacy).*"
 
-integration-test-legacy: all vmbackup vmrestore
+apptest-legacy: all vmbackup vmrestore
 	OS=$$(uname | tr '[:upper:]' '[:lower:]'); \
 	ARCH=$$(uname -m | tr '[:upper:]' '[:lower:]' | sed 's/x86_64/amd64/'); \
 	VERSION=v1.132.0; \
@@ -298,17 +295,17 @@ integration-test-legacy: all vmbackup vmrestore
 	go test ./apptest/tests -run="^TestLegacyCluster.*"
 
 benchmark:
-	GOEXPERIMENT=synctest go test -bench=. ./lib/...
-	go test -bench=. ./app/...
+	go test -run=NO_TESTS -bench=. ./lib/...
+	go test -run=NO_TESTS -bench=. ./app/...
 
 benchmark-pure:
-	GOEXPERIMENT=synctest CGO_ENABLED=0 go test -bench=. ./lib/...
-	CGO_ENABLED=0 go test -bench=. ./app/...
+	CGO_ENABLED=0 go test -run=NO_TESTS -bench=. ./lib/...
+	CGO_ENABLED=0 go test -run=NO_TESTS -bench=. ./app/...
 
 vendor-update:
 	go get -u ./lib/...
 	go get -u ./app/...
-	go mod tidy -compat=1.24
+	go mod tidy -compat=1.26
 	go mod vendor
 
 app-local:
@@ -332,7 +329,7 @@ install-qtc:
 
 
 golangci-lint: install-golangci-lint
-	GOEXPERIMENT=synctest golangci-lint run
+	golangci-lint run --build-tags 'synctest'
 
 install-golangci-lint:
 	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)

README.md

@@ -16,16 +16,21 @@
   <img src="docs/victoriametrics/logo.webp" width="300" alt="VictoriaMetrics logo">
 </picture>
 
-VictoriaMetrics is a fast, cost-saving, and scalable solution for monitoring and managing time series data. It delivers high performance and reliability, making it an ideal choice for businesses of all sizes.
+VictoriaMetrics is a fast, cost-effective, and scalable solution for monitoring and managing time series data. It delivers high performance and reliability, making it an ideal choice for businesses of all sizes.
 
 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/victoriametrics/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)
-- Deployment types: [Single-node version](https://docs.victoriametrics.com/), [Cluster version](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/), and [Enterprise version](https://docs.victoriametrics.com/victoriametrics/enterprise/)
-- Changelog: [CHANGELOG](https://docs.victoriametrics.com/victoriametrics/changelog/), and [How to upgrade](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#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)
+- **Case studies**: [Grammarly, Roblox, Wix, Spotify,...](https://docs.victoriametrics.com/victoriametrics/casestudies/).
+- **Available**: [Binary releases](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/latest), Docker images on [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).
+- **Deployment types**: [Single-node version](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/) and [Cluster version](https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/) under [Apache License 2.0](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/LICENSE).
+- **Getting started:** Read [key concepts](https://docs.victoriametrics.com/victoriametrics/keyconcepts/) and follow the
+  [quick start guide](https://docs.victoriametrics.com/victoriametrics/quick-start/).
+- **Community**: [Slack](https://slack.victoriametrics.com/) (join via [Slack Inviter](https://slack.victoriametrics.com/)), [X (Twitter)](https://x.com/VictoriaMetrics), [YouTube](https://www.youtube.com/@VictoriaMetrics). See full list [here](https://docs.victoriametrics.com/victoriametrics/#community-and-contributions).
+- **Changelog**: Project evolves fast - check the [CHANGELOG](https://docs.victoriametrics.com/victoriametrics/changelog/), and [How to upgrade](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-upgrade-victoriametrics).
+- **Enterprise support:** [Contact us](mailto:info@victoriametrics.com) for commercial support with additional [enterprise features](https://docs.victoriametrics.com/victoriametrics/enterprise/).
+- **Enterprise releases:** Enterprise and [long-term support releases (LTS)](https://docs.victoriametrics.com/victoriametrics/lts-releases/) are publicly available and can be evaluated for free
+  using a [free trial license](https://victoriametrics.com/products/enterprise/trial/).
+- **Security:** we achieved [security certifications](https://victoriametrics.com/security/) for Database Software Development and Software-Based Monitoring Services.
 
 Yes, we open-source both the single-node VictoriaMetrics and the cluster version.
 

SECURITY.md

@@ -12,6 +12,31 @@ The following versions of VictoriaMetrics receive regular security fixes:
 
 See [this page](https://victoriametrics.com/security/) for more details.
 
+## Software Bill of Materials (SBOM)
+
+Every VictoriaMetrics container{{% available_from "#" %}} image published to
+[Docker Hub](https://hub.docker.com/u/victoriametrics)
+and [Quay.io](https://quay.io/organization/victoriametrics)
+includes an [SPDX](https://spdx.dev/) SBOM attestation
+generated automatically by BuildKit during
+`docker buildx build`.
+
+To inspect the SBOM for an image:
+
+```sh
+docker buildx imagetools inspect \
+  docker.io/victoriametrics/victoria-metrics:latest \
+  --format "{{ json .SBOM }}"
+```
+
+To scan an image using its SBOM attestation with
+[Trivy](https://github.com/aquasecurity/trivy):
+
+```sh
+trivy image --sbom-sources oci \
+  docker.io/victoriametrics/victoria-metrics:latest
+```
+
 ## Reporting a Vulnerability
 
 Please report any security issues to <security@victoriametrics.com>

app/vmagent/datadogsketches/request_handler.go

@@ -49,6 +49,11 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 				Name:  "__name__",
 				Value: m.Name,
 			})
+			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10557
+			labels = append(labels, prompb.Label{
+				Name:  "host",
+				Value: sketch.Host,
+			})
 			for _, label := range m.Labels {
 				labels = append(labels, prompb.Label{
 					Name:  label.Name,
@@ -57,9 +62,6 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 			}
 			for _, tag := range sketch.Tags {
 				name, value := datadogutil.SplitTag(tag)
-				if name == "host" {
-					name = "exported_host"
-				}
 				labels = append(labels, prompb.Label{
 					Name:  name,
 					Value: value,

app/vmagent/main.go

@@ -245,6 +245,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		}
 		w.Header().Add("Content-Type", "text/html; charset=utf-8")
 		fmt.Fprintf(w, "<h2>vmagent</h2>")
+		fmt.Fprintf(w, "Version %s<br>", buildinfo.Version)
 		fmt.Fprintf(w, "See docs at <a href='https://docs.victoriametrics.com/victoriametrics/vmagent/'>https://docs.victoriametrics.com/victoriametrics/vmagent/</a></br>")
 		fmt.Fprintf(w, "Useful endpoints:</br>")
 		httpserver.WriteAPIHelp(w, [][2]string{

app/vmagent/remotewrite/client.go

@@ -202,14 +202,10 @@ func (c *client) init(argIdx, concurrency int, sanitizedURL string) {
 	c.retriesCount = metrics.GetOrCreateCounter(fmt.Sprintf(`vmagent_remotewrite_retries_count_total{url=%q}`, c.sanitizedURL))
 	c.sendDuration = metrics.GetOrCreateFloatCounter(fmt.Sprintf(`vmagent_remotewrite_send_duration_seconds_total{url=%q}`, c.sanitizedURL))
 	metrics.GetOrCreateGauge(fmt.Sprintf(`vmagent_remotewrite_queues{url=%q}`, c.sanitizedURL), func() float64 {
-		return float64(*queues)
+		return float64(concurrency)
 	})
-	for i := 0; i < concurrency; i++ {
-		c.wg.Add(1)
-		go func() {
-			defer c.wg.Done()
-			c.runWorker()
-		}()
+	for range concurrency {
+		c.wg.Go(c.runWorker)
 	}
 	logger.Infof("initialized client for -remoteWrite.url=%q", c.sanitizedURL)
 }

app/vmagent/remotewrite/client_test.go

@@ -18,7 +18,7 @@ func TestCalculateRetryDuration(t *testing.T) {
 	f := func(retryAfterDuration, retryDuration time.Duration, n int, expectMinDuration time.Duration) {
 		t.Helper()
 
-		for i := 0; i < n; i++ {
+		for range n {
 			retryDuration = getRetryDuration(retryAfterDuration, retryDuration, time.Minute)
 		}
 

app/vmagent/remotewrite/pendingseries.go

@@ -48,11 +48,7 @@ func newPendingSeries(fq *persistentqueue.FastQueue, isVMRemoteWrite *atomic.Boo
 	ps.wr.significantFigures = significantFigures
 	ps.wr.roundDigits = roundDigits
 	ps.stopCh = make(chan struct{})
-	ps.periodicFlusherWG.Add(1)
-	go func() {
-		defer ps.periodicFlusherWG.Done()
-		ps.periodicFlusher()
-	}()
+	ps.periodicFlusherWG.Go(ps.periodicFlusher)
 	return &ps
 }
 

app/vmagent/remotewrite/pendingseries_test.go

@@ -51,9 +51,9 @@ func testPushWriteRequest(t *testing.T, rowsCount, expectedBlockLenProm, expecte
 
 func newTestWriteRequest(seriesCount, labelsCount int) *prompb.WriteRequest {
 	var wr prompb.WriteRequest
-	for i := 0; i < seriesCount; i++ {
+	for i := range seriesCount {
 		var labels []prompb.Label
-		for j := 0; j < labelsCount; j++ {
+		for j := range labelsCount {
 			labels = append(labels, prompb.Label{
 				Name:  fmt.Sprintf("label_%d_%d", i, j),
 				Value: fmt.Sprintf("value_%d_%d", i, j),

app/vmagent/remotewrite/relabel.go

@@ -20,8 +20,7 @@ import (
 )
 
 var (
-	unparsedLabelsGlobal = flagutil.NewArrayString("remoteWrite.label", "Optional label in the form 'name=value' to add to all the metrics before sending them to -remoteWrite.url. "+
-		"Pass multiple -remoteWrite.label flags in order to add multiple labels to metrics before sending them to remote storage")
+	unparsedLabelsGlobal    = flagutil.NewArrayString("remoteWrite.label", "Optional label in the form 'name=value' to add to all the metrics before sending them to all -remoteWrite.url.")
 	relabelConfigPathGlobal = flag.String("remoteWrite.relabelConfig", "", "Optional path to file with relabeling configs, which are applied "+
 		"to all the metrics before sending them to -remoteWrite.url. See also -remoteWrite.urlRelabelConfig. "+
 		"The path can point either to local file or to http url. "+
@@ -39,7 +38,7 @@ var (
 	labelsGlobal []prompb.Label
 
 	remoteWriteRelabelConfigData    atomic.Pointer[[]byte]
-	remoteWriteURLRelabelConfigData atomic.Pointer[[]interface{}]
+	remoteWriteURLRelabelConfigData atomic.Pointer[[]any]
 
 	relabelConfigReloads      *metrics.Counter
 	relabelConfigReloadErrors *metrics.Counter
@@ -91,8 +90,8 @@ func WriteURLRelabelConfigData(w io.Writer) {
 		return
 	}
 	type urlRelabelCfg struct {
-		Url           string      `yaml:"url"`
-		RelabelConfig interface{} `yaml:"relabel_config"`
+		Url           string `yaml:"url"`
+		RelabelConfig any    `yaml:"relabel_config"`
 	}
 	var cs []urlRelabelCfg
 	for i, url := range *remoteWriteURLs {
@@ -145,7 +144,7 @@ func loadRelabelConfigs() (*relabelConfigs, error) {
 			len(*relabelConfigPaths), (len(*remoteWriteURLs)))
 	}
 
-	var urlRelabelCfgs []interface{}
+	var urlRelabelCfgs []any
 	rcs.perURL = make([]*promrelabel.ParsedConfigs, len(*remoteWriteURLs))
 	for i, path := range *relabelConfigPaths {
 		if len(path) == 0 {
@@ -158,7 +157,7 @@ func loadRelabelConfigs() (*relabelConfigs, error) {
 		}
 		rcs.perURL[i] = prc
 
-		var parsedCfg interface{}
+		var parsedCfg any
 		_ = yaml.Unmarshal(rawCfg, &parsedCfg)
 		urlRelabelCfgs = append(urlRelabelCfgs, parsedCfg)
 	}

app/vmagent/remotewrite/remotewrite.go

@@ -59,7 +59,7 @@ var (
 		"See also -remoteWrite.maxDiskUsagePerURL and -remoteWrite.disableOnDiskQueue")
 	keepDanglingQueues = flag.Bool("remoteWrite.keepDanglingQueues", false, "Keep persistent queues contents at -remoteWrite.tmpDataPath in case there are no matching -remoteWrite.url. "+
 		"Useful when -remoteWrite.url is changed temporarily and persistent queue files will be needed later on.")
-	queues = flag.Int("remoteWrite.queues", cgroup.AvailableCPUs()*2, "The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
+	queues = flagutil.NewArrayInt("remoteWrite.queues", cgroup.AvailableCPUs()*2, "The number of concurrent queues to each -remoteWrite.url. Set more queues if default number of queues "+
 		"isn't enough for sending high volume of collected data to remote storage. "+
 		"Default value depends on the number of available CPU cores. It should work fine in most cases since it minimizes resource usage")
 	showRemoteWriteURL = flag.Bool("remoteWrite.showURL", false, "Whether to show -remoteWrite.url in the exported metrics. "+
@@ -176,13 +176,6 @@ func Init() {
 		})
 	}
 
-	if *queues > maxQueues {
-		*queues = maxQueues
-	}
-	if *queues <= 0 {
-		*queues = 1
-	}
-
 	if len(*shardByURLLabels) > 0 && len(*shardByURLIgnoreLabels) > 0 {
 		logger.Fatalf("-remoteWrite.shardByURL.labels and -remoteWrite.shardByURL.ignoreLabels cannot be set simultaneously; " +
 			"see https://docs.victoriametrics.com/victoriametrics/vmagent/#sharding-among-remote-storages")
@@ -215,9 +208,7 @@ func Init() {
 	dropDanglingQueues()
 
 	// Start config reloader.
-	configReloaderWG.Add(1)
-	go func() {
-		defer configReloaderWG.Done()
+	configReloaderWG.Go(func() {
 		for {
 			select {
 			case <-configReloaderStopCh:
@@ -227,7 +218,7 @@ func Init() {
 			reloadRelabelConfigs()
 			reloadStreamAggrConfigs()
 		}
-	}()
+	})
 }
 
 func dropDanglingQueues() {
@@ -267,17 +258,6 @@ func initRemoteWriteCtxs(urls []string) {
 	if len(urls) == 0 {
 		logger.Panicf("BUG: urls must be non-empty")
 	}
-
-	maxInmemoryBlocks := memory.Allowed() / len(urls) / *maxRowsPerBlock / 100
-	if maxInmemoryBlocks / *queues > 100 {
-		// There is no much sense in keeping higher number of blocks in memory,
-		// since this means that the producer outperforms consumer and the queue
-		// will continue growing. It is better storing the queue to file.
-		maxInmemoryBlocks = 100 * *queues
-	}
-	if maxInmemoryBlocks < 2 {
-		maxInmemoryBlocks = 2
-	}
 	rwctxs := make([]*remoteWriteCtx, len(urls))
 	rwctxIdx := make([]int, len(urls))
 	if retryMaxTime.String() != "" {
@@ -292,7 +272,7 @@ func initRemoteWriteCtxs(urls []string) {
 		if *showRemoteWriteURL {
 			sanitizedURL = fmt.Sprintf("%d:%s", i+1, remoteWriteURL)
 		}
-		rwctxs[i] = newRemoteWriteCtx(i, remoteWriteURL, maxInmemoryBlocks, sanitizedURL)
+		rwctxs[i] = newRemoteWriteCtx(i, remoteWriteURL, sanitizedURL)
 		rwctxIdx[i] = i
 	}
 
@@ -558,11 +538,9 @@ func tryPushMetadataToRemoteStorages(rwctxs []*remoteWriteCtx, mms []prompb.Metr
 	// Push metadata to remote storage systems in parallel to reduce
 	// the time needed for sending the data to multiple remote storage systems.
 	var wg sync.WaitGroup
-	wg.Add(len(rwctxs))
 	var anyPushFailed atomic.Bool
 	for _, rwctx := range rwctxs {
-		go func(rwctx *remoteWriteCtx) {
-			defer wg.Done()
+		wg.Go(func() {
 			if !rwctx.tryPushMetadataInternal(mms) {
 				rwctx.pushFailures.Inc()
 				if forceDropSamplesOnFailure {
@@ -571,7 +549,7 @@ func tryPushMetadataToRemoteStorages(rwctxs []*remoteWriteCtx, mms []prompb.Metr
 				}
 				anyPushFailed.Store(true)
 			}
-		}(rwctx)
+		})
 	}
 	wg.Wait()
 	return !anyPushFailed.Load()
@@ -603,15 +581,13 @@ func tryPushTimeSeriesToRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock []prom
 	// Push tssBlock to remote storage systems in parallel to reduce
 	// the time needed for sending the data to multiple remote storage systems.
 	var wg sync.WaitGroup
-	wg.Add(len(rwctxs))
 	var anyPushFailed atomic.Bool
 	for _, rwctx := range rwctxs {
-		go func(rwctx *remoteWriteCtx) {
-			defer wg.Done()
+		wg.Go(func() {
 			if !rwctx.TryPushTimeSeries(tssBlock, forceDropSamplesOnFailure) {
 				anyPushFailed.Store(true)
 			}
-		}(rwctx)
+		})
 	}
 	wg.Wait()
 	return !anyPushFailed.Load()
@@ -633,13 +609,11 @@ func tryShardingTimeSeriesAmongRemoteStorages(rwctxs []*remoteWriteCtx, tssBlock
 		if len(shard) == 0 {
 			continue
 		}
-		wg.Add(1)
-		go func(rwctx *remoteWriteCtx, tss []prompb.TimeSeries) {
-			defer wg.Done()
-			if !rwctx.TryPushTimeSeries(tss, forceDropSamplesOnFailure) {
+		wg.Go(func() {
+			if !rwctx.TryPushTimeSeries(shard, forceDropSamplesOnFailure) {
 				anyPushFailed.Store(true)
 			}
-		}(rwctx, shard)
+		})
 	}
 	wg.Wait()
 	return !anyPushFailed.Load()
@@ -848,7 +822,7 @@ type remoteWriteCtx struct {
 	rowsDroppedOnPushFailure     *metrics.Counter
 }
 
-func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks int, sanitizedURL string) *remoteWriteCtx {
+func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, sanitizedURL string) *remoteWriteCtx {
 	// strip query params, otherwise changing params resets pq
 	pqURL := *remoteWriteURL
 	pqURL.RawQuery = ""
@@ -863,6 +837,23 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in
 	}
 
 	isPQDisabled := disableOnDiskQueue.GetOptionalArg(argIdx)
+	queuesSize := queues.GetOptionalArg(argIdx)
+	if queuesSize > maxQueues {
+		queuesSize = maxQueues
+	} else if queuesSize <= 0 {
+		queuesSize = 1
+	}
+
+	maxInmemoryBlocks := memory.Allowed() / len(*remoteWriteURLs) / *maxRowsPerBlock / 100
+	if maxInmemoryBlocks/queuesSize > 100 {
+		// There is no much sense in keeping higher number of blocks in memory,
+		// since this means that the producer outperforms consumer and the queue
+		// will continue growing. It is better storing the queue to file.
+		maxInmemoryBlocks = 100 * queuesSize
+	}
+	if maxInmemoryBlocks < 2 {
+		maxInmemoryBlocks = 2
+	}
 	fq := persistentqueue.MustOpenFastQueue(queuePath, sanitizedURL, maxInmemoryBlocks, maxPendingBytes, isPQDisabled)
 	_ = metrics.GetOrCreateGauge(fmt.Sprintf(`vmagent_remotewrite_pending_data_bytes{path=%q, url=%q}`, queuePath, sanitizedURL), func() float64 {
 		return float64(fq.GetPendingBytes())
@@ -880,16 +871,16 @@ func newRemoteWriteCtx(argIdx int, remoteWriteURL *url.URL, maxInmemoryBlocks in
 	var c *client
 	switch remoteWriteURL.Scheme {
 	case "http", "https":
-		c = newHTTPClient(argIdx, remoteWriteURL.String(), sanitizedURL, fq, *queues)
+		c = newHTTPClient(argIdx, remoteWriteURL.String(), sanitizedURL, fq, queuesSize)
 	default:
 		logger.Fatalf("unsupported scheme: %s for remoteWriteURL: %s, want `http`, `https`", remoteWriteURL.Scheme, sanitizedURL)
 	}
-	c.init(argIdx, *queues, sanitizedURL)
+	c.init(argIdx, queuesSize, sanitizedURL)
 
 	// Initialize pss
 	sf := significantFigures.GetOptionalArg(argIdx)
 	rd := roundDigits.GetOptionalArg(argIdx)
-	pssLen := *queues
+	pssLen := queuesSize
 	if n := cgroup.AvailableCPUs(); pssLen > n {
 		// There is no sense in running more than availableCPUs concurrent pendingSeries,
 		// since every pendingSeries can saturate up to a single CPU.
@@ -1089,7 +1080,7 @@ func (rwctx *remoteWriteCtx) tryPushTimeSeriesInternal(tss []prompb.TimeSeries)
 	}()
 
 	if len(labelsGlobal) > 0 {
-		// Make a copy of tss before adding extra labels in order to prevent
+		// Make a copy of tss before adding extra labels to prevent
 		// from affecting time series for other remoteWrite.url configs.
 		rctx = getRelabelCtx()
 		v = tssPool.Get().(*[]prompb.TimeSeries)

app/vmagent/remotewrite/remotewrite_test.go

@@ -28,12 +28,12 @@ func TestGetLabelsHash_Distribution(t *testing.T) {
 		itemsCount := 1_000 * bucketsCount
 		m := make([]int, bucketsCount)
 		var labels []prompb.Label
-		for i := 0; i < itemsCount; i++ {
+		for i := range itemsCount {
 			labels = append(labels[:0], prompb.Label{
 				Name:  "__name__",
 				Value: fmt.Sprintf("some_name_%d", i),
 			})
-			for j := 0; j < 10; j++ {
+			for j := range 10 {
 				labels = append(labels, prompb.Label{
 					Name:  fmt.Sprintf("label_%d", j),
 					Value: fmt.Sprintf("value_%d_%d", i, j),
@@ -248,7 +248,7 @@ func TestShardAmountRemoteWriteCtx(t *testing.T) {
 		seriesCount := 100000
 		// build 1000000 series
 		tssBlock := make([]prompb.TimeSeries, 0, seriesCount)
-		for i := 0; i < seriesCount; i++ {
+		for i := range seriesCount {
 			tssBlock = append(tssBlock, prompb.TimeSeries{
 				Labels: []prompb.Label{
 					{
@@ -269,7 +269,7 @@ func TestShardAmountRemoteWriteCtx(t *testing.T) {
 		// build active time series set
 		nodes := make([]string, 0, remoteWriteCount)
 		activeTimeSeriesByNodes := make([]map[string]struct{}, remoteWriteCount)
-		for i := 0; i < remoteWriteCount; i++ {
+		for i := range remoteWriteCount {
 			nodes = append(nodes, fmt.Sprintf("node%d", i))
 			activeTimeSeriesByNodes[i] = make(map[string]struct{})
 		}

app/vmalert/config/config.go

@@ -81,12 +81,9 @@ func (g *Group) Validate(validateTplFn ValidateTplFn, validateExpressions bool)
 	if g.Interval.Duration() < 0 {
 		return fmt.Errorf("interval shouldn't be lower than 0")
 	}
-	if g.EvalOffset.Duration() < 0 {
-		return fmt.Errorf("eval_offset shouldn't be lower than 0")
-	}
-	// if `eval_offset` is set, interval won't use global evaluationInterval flag and must bigger than offset.
-	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 `eval_offset` is set, the group interval must be specified explicitly(instead of inherited from global evaluationInterval flag) and must bigger than offset.
+	if g.EvalOffset.Duration().Abs() > g.Interval.Duration() {
+		return fmt.Errorf("the abs value of 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")

app/vmalert/config/config_test.go

@@ -176,11 +176,17 @@ func TestGroupValidate_Failure(t *testing.T) {
 	}, false, "interval shouldn't be lower than 0")
 
 	f(&Group{
-		Name:       "wrong eval_offset",
+		Name:       "too big eval_offset",
 		Interval:   promutil.NewDuration(time.Minute),
 		EvalOffset: promutil.NewDuration(2 * time.Minute),
 	}, false, "eval_offset should be smaller than interval")
 
+	f(&Group{
+		Name:       "too big negative eval_offset",
+		Interval:   promutil.NewDuration(time.Minute),
+		EvalOffset: promutil.NewDuration(-2 * time.Minute),
+	}, false, "eval_offset should be smaller than interval")
+
 	limit := -1
 	f(&Group{
 		Name:  "wrong limit",

app/vmalert/config/types.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	"slices"
 	"strings"
 
 	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
@@ -76,13 +77,12 @@ func (t *Type) ValidateExpr(expr string) error {
 		if 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)
-			}
+		labels, err := q.GetStatsLabels()
+		if err != nil {
+			return fmt.Errorf("cannot obtain labels from LogsQL expr: %q, err: %w", expr, err)
+		}
+		if slices.Contains(labels, "_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/datasource/client.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"maps"
 	"net/http"
 	"net/url"
 	"strings"
@@ -91,9 +92,7 @@ func (c *Client) Clone() *Client {
 		ns.extraHeaders = make([]keyValue, len(c.extraHeaders))
 		copy(ns.extraHeaders, c.extraHeaders)
 	}
-	for k, v := range c.extraParams {
-		ns.extraParams[k] = v
-	}
+	maps.Copy(ns.extraParams, c.extraParams)
 
 	return ns
 }

app/vmalert/datasource/client_prom.go

@@ -34,7 +34,7 @@ type promResponse struct {
 	// Stats supported by VictoriaMetrics since v1.90
 	Stats struct {
 		SeriesFetched *string `json:"seriesFetched,omitempty"`
-	} `json:"stats,omitempty"`
+	} `json:"stats"`
 	// IsPartial supported by VictoriaMetrics
 	IsPartial *bool `json:"isPartial,omitempty"`
 }

app/vmalert/datasource/datasource.go

@@ -134,7 +134,7 @@ func (ls Labels) String() string {
 func LabelCompare(a, b Labels) int {
 	l := min(len(b), len(a))
 
-	for i := 0; i < l; i++ {
+	for i := range l {
 		if a[i].Name != b[i].Name {
 			if a[i].Name < b[i].Name {
 				return -1

app/vmalert/datasource/vm_prom_api_timing_test.go

@@ -13,7 +13,7 @@ func BenchmarkPromInstantUnmarshal(b *testing.B) {
 
 	// BenchmarkParsePrometheusResponse/Instant_std+fastjson-10                    1760            668959 ns/op          280147 B/op       5781 allocs/op
 	b.Run("Instant std+fastjson", func(b *testing.B) {
-		for i := 0; i < b.N; i++ {
+		for range b.N {
 			var pi promInstant
 			err = pi.Unmarshal(data)
 			if err != nil {

app/vmalert/main.go

@@ -56,7 +56,7 @@ absolute path to all .tpl files in root.
  -rule.templates="dir/**/*.tpl". Includes all the .tpl files in "dir" subfolders recursively.
 `)
 
-	configCheckInterval = flag.Duration("configCheckInterval", 0, "Interval for checking for changes in '-rule' or '-notifier.config' files. "+
+	configCheckInterval = flag.Duration("configCheckInterval", 0, "Interval for checking for changes in '-rule', '-rule.templates' and '-notifier.config' files. "+
 		"By default, the checking is disabled. Send SIGHUP signal in order to force config check for changes.")
 
 	httpListenAddrs  = flagutil.NewArrayString("httpListenAddr", "Address to listen for incoming http requests. See also -tls and -httpListenAddr.useProxyProtocol")
@@ -81,9 +81,7 @@ absolute path to all .tpl files in root.
 	dryRun = flag.Bool("dryRun", false, "Whether to check only config files without running vmalert. The rules file are validated. The -rule flag must be specified.")
 )
 
-var (
-	extURL *url.URL
-)
+var extURL *url.URL
 
 func main() {
 	// Write flags and help message to stdout, since it is easier to grep or pipe.
@@ -161,7 +159,7 @@ func main() {
 	ctx, cancel := context.WithCancel(context.Background())
 	manager, err := newManager(ctx)
 	if err != nil {
-		logger.Fatalf("failed to init: %s", err)
+		logger.Fatalf("failed to create manager: %s", err)
 	}
 	logger.Infof("reading rules configuration file from %q", strings.Join(*rulePath, ";"))
 	groupsCfg, err := config.Parse(*rulePath, validateTplFn, *validateExpressions)

app/vmalert/manager_test.go

@@ -65,13 +65,11 @@ func TestManagerUpdateConcurrent(t *testing.T) {
 
 	const workers = 500
 	const iterations = 10
-	wg := sync.WaitGroup{}
-	wg.Add(workers)
-	for i := 0; i < workers; i++ {
-		go func(n int) {
-			defer wg.Done()
+	var wg sync.WaitGroup
+	for n := range workers {
+		wg.Go(func() {
 			r := rand.New(rand.NewSource(int64(n)))
-			for i := 0; i < iterations; i++ {
+			for range iterations {
 				rnd := r.Intn(len(paths))
 				cfg, err := config.Parse([]string{paths[rnd]}, notifier.ValidateTemplates, true)
 				if err != nil { // update can fail and this is expected
@@ -79,7 +77,7 @@ func TestManagerUpdateConcurrent(t *testing.T) {
 				}
 				_ = m.update(context.Background(), cfg, false)
 			}
-		}(i)
+		})
 	}
 	wg.Wait()
 }
@@ -261,7 +259,7 @@ func compareGroups(t *testing.T, a, b *rule.Group) {
 	for i, r := range a.Rules {
 		got, want := r, b.Rules[i]
 		if a.CreateID() != b.CreateID() {
-			t.Fatalf("expected to have rule %q; got %q", want.ID(), got.ID())
+			t.Fatalf("expected to have rule %d; got %d", want.ID(), got.ID())
 		}
 		if err := rule.CompareRules(t, want, got); err != nil {
 			t.Fatalf("comparison error: %s", err)

app/vmalert/notifier/alertmanager.go

@@ -14,7 +14,6 @@ import (
 	"github.com/VictoriaMetrics/metrics"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
@@ -172,11 +171,6 @@ const alertManagerPath = "/api/v2/alerts"
 func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, authCfg promauth.HTTPClientConfig,
 	relabelCfg *promrelabel.ParsedConfigs, timeout time.Duration,
 ) (*AlertManager, error) {
-
-	if err := httputil.CheckURL(alertManagerURL); err != nil {
-		return nil, fmt.Errorf("invalid alertmanager URL: %w", err)
-	}
-
 	tls := &promauth.TLSConfig{}
 	if authCfg.TLSConfig != nil {
 		tls = authCfg.TLSConfig

app/vmalert/notifier/config_watcher_test.go

@@ -212,18 +212,16 @@ consul_sd_configs:
 
 	const workers = 500
 	const iterations = 10
-	wg := sync.WaitGroup{}
-	wg.Add(workers)
-	for i := 0; i < workers; i++ {
-		go func(n int) {
-			defer wg.Done()
+	var wg sync.WaitGroup
+	for n := range workers {
+		wg.Go(func() {
 			r := rand.New(rand.NewSource(int64(n)))
-			for i := 0; i < iterations; i++ {
+			for range iterations {
 				rnd := r.Intn(len(paths))
 				_ = cw.reload(paths[rnd]) // update can fail and this is expected
 				_ = cw.notifiers()
 			}
-		}(i)
+		})
 	}
 	wg.Wait()
 }

app/vmalert/notifier/init.go

@@ -11,8 +11,8 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
@@ -229,6 +229,9 @@ func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
 			Headers: []string{headers.GetOptionalArg(i)},
 		}
 
+		if err := httputil.CheckURL(addr); err != nil {
+			return nil, fmt.Errorf("invalid notifier.url %q: %w", addr, err)
+		}
 		addr = strings.TrimSuffix(addr, "/")
 		am, err := NewAlertManager(addr+alertManagerPath, gen, authCfg, nil, sendTimeout.GetOptionalArg(i))
 		if err != nil {
@@ -266,7 +269,7 @@ func GetTargets() map[TargetType][]Target {
 	if getActiveNotifiers == nil {
 		return nil
 	}
-	var targets = make(map[TargetType][]Target)
+	targets := make(map[TargetType][]Target)
 	// use cached targets from configWatcher instead of getActiveNotifiers for the extra target labels
 	if cw != nil {
 		cw.targetsMu.RLock()
@@ -287,7 +290,7 @@ func GetTargets() map[TargetType][]Target {
 }
 
 // Send sends alerts to all active notifiers
-func Send(ctx context.Context, alerts []Alert, notifierHeaders map[string]string) *vmalertutil.ErrGroup {
+func Send(ctx context.Context, alerts []Alert, notifierHeaders map[string]string) chan error {
 	alertsToSend := make([]Alert, 0, len(alerts))
 	lblss := make([][]prompb.Label, 0, len(alerts))
 	// apply global relabel config first without modifying original alerts in alerts
@@ -300,17 +303,18 @@ func Send(ctx context.Context, alerts []Alert, notifierHeaders map[string]string
 		lblss = append(lblss, lbls)
 	}
 
-	errGr := new(vmalertutil.ErrGroup)
 	wg := sync.WaitGroup{}
 	activeNotifiers := getActiveNotifiers()
+	errCh := make(chan error, len(activeNotifiers))
+	defer close(errCh)
 	for i := range activeNotifiers {
 		nt := activeNotifiers[i]
 		wg.Go(func() {
 			if err := nt.Send(ctx, alertsToSend, lblss, notifierHeaders); err != nil {
-				errGr.Add(fmt.Errorf("failed to send alerts to addr %q: %w", nt.Addr(), err))
+				errCh <- fmt.Errorf("failed to send alerts to addr %q: %w", nt.Addr(), err)
 			}
 		})
 	}
 	wg.Wait()
-	return errGr
+	return errCh
 }

app/vmalert/notifier/init_test.go

@@ -55,9 +55,9 @@ func TestInitNegative(t *testing.T) {
 		*blackHole = oldBlackHole
 	}()
 
-	f := func(path, addr string, bh bool) {
+	f := func(path string, addr []string, bh bool) {
 		*configPath = path
-		*addrs = flagutil.ArrayString{addr}
+		*addrs = flagutil.ArrayString(addr)
 		*blackHole = bh
 		if err := Init(nil, ""); err == nil {
 			t.Fatalf("expected to get error; got nil instead")
@@ -65,9 +65,12 @@ func TestInitNegative(t *testing.T) {
 	}
 
 	// *configPath, *addrs and *blackhole are mutually exclusive
-	f("/dummy/path", "127.0.0.1", false)
-	f("/dummy/path", "", true)
-	f("", "127.0.0.1", true)
+	f("/dummy/path", []string{"127.0.0.1"}, false)
+	f("/dummy/path", []string{}, true)
+	f("", []string{"127.0.0.1"}, true)
+	// addr cannot be ""
+	f("", []string{""}, false)
+	f("", []string{"127.0.0.1", ""}, false)
 }
 
 func TestBlackHole(t *testing.T) {
@@ -202,7 +205,9 @@ alert_relabel_configs:
 		},
 	}
 	errG := Send(context.Background(), firingAlerts, nil)
-	if errG.Err() != nil {
-		t.Fatalf("unexpected error when sending alerts: %s", err)
+	for err := range errG {
+		if err != nil {
+			t.Errorf("unexpected error when sending alerts: %s", err)
+		}
 	}
 }

app/vmalert/remotewrite/client.go

@@ -113,7 +113,7 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 		input:         make(chan prompb.TimeSeries, cfg.MaxQueueSize),
 	}
 
-	for i := 0; i < cc; i++ {
+	for range cc {
 		c.run(ctx)
 	}
 	return c, nil
@@ -186,6 +186,11 @@ func (c *Client) run(ctx context.Context) {
 				return
 			case <-ticker.C:
 				c.flush(ctx, wr)
+				// drain the potential stale tick to avoid small or empty flushes after a slow flush.
+				select {
+				case <-ticker.C:
+				default:
+				}
 			case ts, ok := <-c.input:
 				if !ok {
 					continue
@@ -238,8 +243,10 @@ func (c *Client) flush(ctx context.Context, wr *prompb.WriteRequest) {
 	defer func() {
 		sendDuration.Add(time.Since(timeStart).Seconds())
 	}()
+
+	attempts := 0
 L:
-	for attempts := 0; ; attempts++ {
+	for {
 		err := c.send(ctx, b)
 		if err != nil && (errors.Is(err, io.EOF) || netutil.IsTrivialNetworkError(err)) {
 			// Something in the middle between client and destination might be closing
@@ -281,6 +288,7 @@ L:
 		time.Sleep(retryInterval)
 		retryInterval *= 2
 
+		attempts++
 	}
 
 	rwErrors.Inc()

app/vmalert/remotewrite/client_test.go

@@ -44,7 +44,7 @@ func TestClient_Push(t *testing.T) {
 
 	r := rand.New(rand.NewSource(1))
 	const rowsN = int(1e4)
-	for i := 0; i < rowsN; i++ {
+	for range rowsN {
 		s := prompb.TimeSeries{
 			Samples: []prompb.Sample{{
 				Value:     r.Float64(),
@@ -102,7 +102,7 @@ func TestClient_run_maxBatchSizeDuringShutdown(t *testing.T) {
 		}
 
 		// push time series to the client.
-		for i := 0; i < pushCnt; i++ {
+		for range pushCnt {
 			if err = rwClient.Push(prompb.TimeSeries{}); err != nil {
 				t.Fatalf("cannot time series to the client: %s", err)
 			}

app/vmalert/remotewrite/debug_client_test.go

@@ -22,7 +22,7 @@ func TestDebugClient_Push(t *testing.T) {
 
 	const rowsN = 100
 	var sent int
-	for i := 0; i < rowsN; i++ {
+	for i := range rowsN {
 		s := prompb.TimeSeries{
 			Samples: []prompb.Sample{{
 				Value:     float64(i),

app/vmalert/rule/alerting.go

@@ -818,7 +818,9 @@ func (ar *AlertingRule) restore(ctx context.Context, q datasource.Querier, ts ti
 	expr := fmt.Sprintf("default_rollup(%s{%s%s}[%ds])",
 		alertForStateMetricName, nameStr, labelsFilter, int(lookback.Seconds()))
 
-	res, _, err := q.Query(ctx, expr, ts)
+	// query ALERTS_FOR_STATE at `ts-1s` instead `ts` to avoid retrieving data written in the current run,
+	// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10335
+	res, _, err := q.Query(ctx, expr, ts.Add(-1*time.Second))
 	if err != nil {
 		return fmt.Errorf("failed to execute restore query %q: %w ", expr, err)
 	}

app/vmalert/rule/alerting_synctest_test.go

@@ -0,0 +1,106 @@
+//go:build synctest
+
+package rule
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
+)
+
+// TestAlertingRule_ActiveAtPreservedInAnnotations ensures that the fix for
+// https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9543 is preserved
+// while allowing query templates in labels (https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9783)
+func TestAlertingRule_ActiveAtPreservedInAnnotations(t *testing.T) {
+	// wrap into synctest because of time manipulations
+	synctest.Test(t, func(t *testing.T) {
+		fq := &datasource.FakeQuerier{}
+
+		ar := &AlertingRule{
+			Name: "TestActiveAtPreservation",
+			Labels: map[string]string{
+				"test_query_in_label": `{{ "static_value" }}`,
+			},
+			Annotations: map[string]string{
+				"description": "Alert active since {{ $activeAt }}",
+			},
+			alerts: make(map[uint64]*notifier.Alert),
+			q:      fq,
+			state: &ruleState{
+				entries: make([]StateEntry, 10),
+			},
+		}
+
+		// Mock query result - return empty result to make suppress_for_mass_alert = false
+		// (no need to add anything to fq for empty result)
+
+		// Add a metric that should trigger the alert
+		fq.Add(metricWithValueAndLabels(t, 1, "instance", "server1"))
+
+		// First execution - creates new alert
+		ts1 := time.Now()
+		_, err := ar.exec(context.TODO(), ts1, 0)
+		if err != nil {
+			t.Fatalf("unexpected error on first exec: %s", err)
+		}
+
+		if len(ar.alerts) != 1 {
+			t.Fatalf("expected 1 alert, got %d", len(ar.alerts))
+		}
+
+		firstAlert := ar.GetAlerts()[0]
+		// Verify first execution: activeAt should be ts1 and annotation should reflect it
+		if !firstAlert.ActiveAt.Equal(ts1) {
+			t.Fatalf("expected activeAt to be %v, got %v", ts1, firstAlert.ActiveAt)
+		}
+
+		// Extract time from annotation (format will be like "Alert active since 2025-09-30 08:55:13.638551611 -0400 EDT m=+0.002928464")
+		expectedTimeStr := ts1.Format("2006-01-02 15:04:05")
+		if !strings.Contains(firstAlert.Annotations["description"], expectedTimeStr) {
+			t.Fatalf("first exec annotation should contain time %s, got: %s", expectedTimeStr, firstAlert.Annotations["description"])
+		}
+
+		// Second execution - should preserve activeAt in annotation
+
+		// Ensure different timestamp with different seconds
+		// sleep is non-blocking thanks to synctest
+		time.Sleep(2 * time.Second)
+		ts2 := time.Now()
+		_, err = ar.exec(context.TODO(), ts2, 0)
+		if err != nil {
+			t.Fatalf("unexpected error on second exec: %s", err)
+		}
+
+		// Get the alert again (should be the same alert)
+		if len(ar.alerts) != 1 {
+			t.Fatalf("expected 1 alert, got %d", len(ar.alerts))
+		}
+		secondAlert := ar.GetAlerts()[0]
+
+		// Critical test: activeAt should still be ts1, not ts2
+		if !secondAlert.ActiveAt.Equal(ts1) {
+			t.Fatalf("activeAt should be preserved as %v, but got %v", ts1, secondAlert.ActiveAt)
+		}
+
+		// Critical test: annotation should still contain ts1 time, not ts2
+		if !strings.Contains(secondAlert.Annotations["description"], expectedTimeStr) {
+			t.Fatalf("second exec annotation should still contain original time %s, got: %s", expectedTimeStr, secondAlert.Annotations["description"])
+		}
+
+		// Additional verification: annotation should NOT contain ts2 time
+		ts2TimeStr := ts2.Format("2006-01-02 15:04:05")
+		if strings.Contains(secondAlert.Annotations["description"], ts2TimeStr) {
+			t.Fatalf("annotation should NOT contain new eval time %s, got: %s", ts2TimeStr, secondAlert.Annotations["description"])
+		}
+
+		// Verify query template in labels still works (this would fail if query templates were broken)
+		if firstAlert.Labels["test_query_in_label"] != "static_value" {
+			t.Fatalf("expected test_query_in_label=static_value, got %s", firstAlert.Labels["test_query_in_label"])
+		}
+	})
+}

app/vmalert/rule/alerting_test.go

@@ -10,7 +10,6 @@ import (
 	"strings"
 	"sync"
 	"testing"
-	"testing/synctest"
 	"time"
 
 	"github.com/VictoriaMetrics/metrics"
@@ -1479,95 +1478,3 @@ func TestAlertingRule_QueryTemplateInLabels(t *testing.T) {
 		t.Fatalf("expected 'suppress_for_mass_alert' label to be 'true' or 'false', got '%s'", suppressLabel)
 	}
 }
-
-// TestAlertingRule_ActiveAtPreservedInAnnotations ensures that the fix for
-// https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9543 is preserved
-// while allowing query templates in labels (https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9783)
-func TestAlertingRule_ActiveAtPreservedInAnnotations(t *testing.T) {
-	// wrap into synctest because of time manipulations
-	synctest.Test(t, func(t *testing.T) {
-		fq := &datasource.FakeQuerier{}
-
-		ar := &AlertingRule{
-			Name: "TestActiveAtPreservation",
-			Labels: map[string]string{
-				"test_query_in_label": `{{ "static_value" }}`,
-			},
-			Annotations: map[string]string{
-				"description": "Alert active since {{ $activeAt }}",
-			},
-			alerts: make(map[uint64]*notifier.Alert),
-			q:      fq,
-			state: &ruleState{
-				entries: make([]StateEntry, 10),
-			},
-		}
-
-		// Mock query result - return empty result to make suppress_for_mass_alert = false
-		// (no need to add anything to fq for empty result)
-
-		// Add a metric that should trigger the alert
-		fq.Add(metricWithValueAndLabels(t, 1, "instance", "server1"))
-
-		// First execution - creates new alert
-		ts1 := time.Now()
-		_, err := ar.exec(context.TODO(), ts1, 0)
-		if err != nil {
-			t.Fatalf("unexpected error on first exec: %s", err)
-		}
-
-		if len(ar.alerts) != 1 {
-			t.Fatalf("expected 1 alert, got %d", len(ar.alerts))
-		}
-
-		firstAlert := ar.GetAlerts()[0]
-		// Verify first execution: activeAt should be ts1 and annotation should reflect it
-		if !firstAlert.ActiveAt.Equal(ts1) {
-			t.Fatalf("expected activeAt to be %v, got %v", ts1, firstAlert.ActiveAt)
-		}
-
-		// Extract time from annotation (format will be like "Alert active since 2025-09-30 08:55:13.638551611 -0400 EDT m=+0.002928464")
-		expectedTimeStr := ts1.Format("2006-01-02 15:04:05")
-		if !strings.Contains(firstAlert.Annotations["description"], expectedTimeStr) {
-			t.Fatalf("first exec annotation should contain time %s, got: %s", expectedTimeStr, firstAlert.Annotations["description"])
-		}
-
-		// Second execution - should preserve activeAt in annotation
-
-		// Ensure different timestamp with different seconds
-		// sleep is non-blocking thanks to synctest
-		time.Sleep(2 * time.Second)
-		ts2 := time.Now()
-		_, err = ar.exec(context.TODO(), ts2, 0)
-		if err != nil {
-			t.Fatalf("unexpected error on second exec: %s", err)
-		}
-
-		// Get the alert again (should be the same alert)
-		if len(ar.alerts) != 1 {
-			t.Fatalf("expected 1 alert, got %d", len(ar.alerts))
-		}
-		secondAlert := ar.GetAlerts()[0]
-
-		// Critical test: activeAt should still be ts1, not ts2
-		if !secondAlert.ActiveAt.Equal(ts1) {
-			t.Fatalf("activeAt should be preserved as %v, but got %v", ts1, secondAlert.ActiveAt)
-		}
-
-		// Critical test: annotation should still contain ts1 time, not ts2
-		if !strings.Contains(secondAlert.Annotations["description"], expectedTimeStr) {
-			t.Fatalf("second exec annotation should still contain original time %s, got: %s", expectedTimeStr, secondAlert.Annotations["description"])
-		}
-
-		// Additional verification: annotation should NOT contain ts2 time
-		ts2TimeStr := ts2.Format("2006-01-02 15:04:05")
-		if strings.Contains(secondAlert.Annotations["description"], ts2TimeStr) {
-			t.Fatalf("annotation should NOT contain new eval time %s, got: %s", ts2TimeStr, secondAlert.Annotations["description"])
-		}
-
-		// Verify query template in labels still works (this would fail if query templates were broken)
-		if firstAlert.Labels["test_query_in_label"] != "static_value" {
-			t.Fatalf("expected test_query_in_label=static_value, got %s", firstAlert.Labels["test_query_in_label"])
-		}
-	})
-}

app/vmalert/rule/group.go

@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	"hash/fnv"
+	"maps"
 	"net/url"
 	"sync"
 	"time"
@@ -18,6 +19,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/remotewrite"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
@@ -29,8 +31,8 @@ var (
 		"0 means no limit.")
 	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.")
-	maxResolveDuration = flag.Duration("rule.maxResolveDuration", 0, "Limits the maxiMum duration for automatic alert expiration, "+
+	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). "+
@@ -96,9 +98,7 @@ type groupMetrics struct {
 // set2 has priority over set1.
 func mergeLabels(groupName, ruleName string, set1, set2 map[string]string) map[string]string {
 	r := map[string]string{}
-	for k, v := range set1 {
-		r[k] = v
-	}
+	maps.Copy(r, set1)
 	for k, v := range set2 {
 		if prevV, ok := r[k]; ok {
 			logger.Infof("label %q=%q for rule %q.%q overwritten with external label %q=%q",
@@ -374,7 +374,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 
 	g.infof("started")
 
-	eval := func(ctx context.Context, ts time.Time) {
+	eval := func(ctx context.Context, ts time.Time) time.Time {
 		g.metrics.iterationTotal.Inc()
 
 		start := time.Now()
@@ -382,7 +382,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 		if len(g.Rules) < 1 {
 			g.metrics.iterationDuration.UpdateDuration(start)
 			g.LastEvaluation = start
-			return
+			return ts
 		}
 
 		resolveDuration := getResolveDuration(g.Interval, *resendDelay, *maxResolveDuration)
@@ -396,6 +396,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 		}
 		g.metrics.iterationDuration.UpdateDuration(start)
 		g.LastEvaluation = start
+		return ts
 	}
 
 	evalCtx, cancel := context.WithCancel(ctx)
@@ -404,7 +405,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 	g.mu.Unlock()
 	defer g.evalCancel()
 
-	eval(evalCtx, evalTS)
+	realEvalTS := eval(evalCtx, evalTS)
 
 	t := time.NewTicker(g.Interval)
 	defer t.Stop()
@@ -412,7 +413,7 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 	// restore the rules state after the first evaluation
 	// so only active alerts can be restored.
 	if rr != nil {
-		err := g.restore(ctx, rr, evalTS, *remoteReadLookBack)
+		err := g.restore(ctx, rr, realEvalTS, *remoteReadLookBack)
 		if err != nil {
 			logger.Errorf("error while restoring ruleState for group %q: %s", g.Name, err)
 		}
@@ -483,8 +484,15 @@ func (g *Group) UpdateWith(newGroup *Group) {
 // delayBeforeStart calculates delay based on Group ID, so all groups will start at different moments of time.
 func (g *Group) delayBeforeStart(ts time.Time, maxDelay time.Duration) time.Duration {
 	if g.EvalOffset != nil {
+		offset := *g.EvalOffset
+		// adjust the offset for negative evalOffset, the rule is:
+		// `eval_offset: -x` is equivalent to `eval_offset: y` for `interval: x+y`.
+		// For example, `eval_offset: -6m` is equivalent to `eval_offset: 4m` for `interval: 10m`.
+		if offset < 0 {
+			offset += g.Interval
+		}
 		// if offset is specified, ignore the maxDelay and return a duration aligned with offset
-		currentOffsetPoint := ts.Truncate(g.Interval).Add(*g.EvalOffset)
+		currentOffsetPoint := ts.Truncate(g.Interval).Add(offset)
 		if currentOffsetPoint.Before(ts) {
 			// wait until the next offset point
 			return currentOffsetPoint.Add(g.Interval).Sub(ts)
@@ -493,11 +501,8 @@ func (g *Group) delayBeforeStart(ts time.Time, maxDelay time.Duration) time.Dura
 	}
 
 	// otherwise, return a random duration between [0..min(interval, maxDelay)] based on group ID
-	interval := g.Interval
-	if interval > maxDelay {
-		// artificially limit interval, so groups with big intervals could start sooner.
-		interval = maxDelay
-	}
+	// artificially limit interval, so groups with big intervals could start sooner.
+	interval := min(g.Interval, maxDelay)
 	var randSleep time.Duration
 	randSleep = time.Duration(float64(interval) * (float64(g.GetID()) / (1 << 64)))
 	sleepOffset := time.Duration(ts.UnixNano() % interval.Nanoseconds())
@@ -755,6 +760,7 @@ func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDurati
 		return fmt.Errorf("rule %q: failed to execute: %w", r, err)
 	}
 
+	var errG vmalertutil.ErrGroup
 	if e.Rw != nil {
 		pushToRW := func(tss []prompb.TimeSeries) error {
 			var lastErr error
@@ -766,20 +772,26 @@ func (e *executor) exec(ctx context.Context, r Rule, ts time.Time, resolveDurati
 			return lastErr
 		}
 		if err := pushToRW(tss); err != nil {
-			return err
+			errG.Add(err)
 		}
 	}
 
 	ar, ok := r.(*AlertingRule)
 	if !ok {
-		return nil
+		return errG.Err()
 	}
 
 	alerts := ar.alertsToSend(resolveDuration, *resendDelay)
 	if len(alerts) < 1 {
-		return nil
+		return errG.Err()
 	}
 
-	errGr := notifier.Send(ctx, alerts, e.notifierHeaders)
-	return errGr.Err()
+	notifierErr := notifier.Send(ctx, alerts, e.notifierHeaders)
+	for err := range notifierErr {
+		if err != nil {
+			errG.Add(fmt.Errorf("rule %q: notifier failure: %w", r, err))
+		}
+	}
+
+	return errG.Err()
 }

app/vmalert/rule/group_test.go

@@ -405,7 +405,8 @@ func TestGroupStart(t *testing.T) {
 
 		var cur uint64
 		prev := g.metrics.iterationTotal.Get()
-		for i := 0; ; i++ {
+		i := 0
+		for {
 			if i > 40 {
 				t.Fatalf("group wasn't able to perform %d evaluations during %d eval intervals", n, i)
 			}
@@ -414,6 +415,7 @@ func TestGroupStart(t *testing.T) {
 				return
 			}
 			time.Sleep(interval)
+			i++
 		}
 	}
 
@@ -604,6 +606,15 @@ func TestGroupStartDelay(t *testing.T) {
 	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")
 
+	// test group with negative offset -2min, which is equivalent to 3min offset for 5min interval
+	offset = -2 * time.Minute
+	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")
+
 	maxDelay = time.Minute * 1
 	g.EvalOffset = nil
 

app/vmalert/rule/rule.go

@@ -121,7 +121,7 @@ func (s *ruleState) add(e StateEntry) {
 func replayRule(r Rule, start, end time.Time, rw remotewrite.RWClient, replayRuleRetryAttempts int) (int, error) {
 	var err error
 	var tss []prompb.TimeSeries
-	for i := 0; i < replayRuleRetryAttempts; i++ {
+	for i := range replayRuleRetryAttempts {
 		tss, err = r.execRange(context.Background(), start, end)
 		if err == nil {
 			break

app/vmalert/rule/rule_test.go

@@ -40,7 +40,7 @@ func TestRule_state(t *testing.T) {
 	}
 
 	var last time.Time
-	for i := 0; i < stateEntriesN*2; i++ {
+	for range stateEntriesN * 2 {
 		last = time.Now()
 		r.state.add(StateEntry{At: last})
 	}
@@ -65,17 +65,15 @@ func TestRule_stateConcurrent(_ *testing.T) {
 	r := &AlertingRule{state: &ruleState{entries: make([]StateEntry, 20)}}
 	const workers = 50
 	const iterations = 100
-	wg := sync.WaitGroup{}
-	wg.Add(workers)
-	for i := 0; i < workers; i++ {
-		go func() {
-			defer wg.Done()
-			for i := 0; i < iterations; i++ {
+	var wg sync.WaitGroup
+	for range workers {
+		wg.Go(func() {
+			for range iterations {
 				r.state.add(StateEntry{At: time.Now()})
 				r.state.getAll()
 				r.state.getLast()
 			}
-		}()
+		})
 	}
 	wg.Wait()
 }

app/vmalert/rule/test_helpers.go

@@ -19,13 +19,13 @@ func CompareRules(t *testing.T, a, b Rule) error {
 	case *AlertingRule:
 		br, ok := b.(*AlertingRule)
 		if !ok {
-			return fmt.Errorf("rule %q supposed to be of type AlertingRule", b.ID())
+			return fmt.Errorf("rule %d supposed to be of type AlertingRule", b.ID())
 		}
 		return compareAlertingRules(t, v, br)
 	case *RecordingRule:
 		br, ok := b.(*RecordingRule)
 		if !ok {
-			return fmt.Errorf("rule %q supposed to be of type RecordingRule", b.ID())
+			return fmt.Errorf("rule %d supposed to be of type RecordingRule", b.ID())
 		}
 		return compareRecordingRules(t, v, br)
 	default:

app/vmalert/vmalertutil/err_group.go

@@ -45,7 +45,7 @@ func (eg *ErrGroup) Error() string {
 		return ""
 	}
 	var b strings.Builder
-	fmt.Fprintf(&b, "errors(%d): ", len(eg.errs))
+	fmt.Fprintf(&b, "errors(%d): \n", len(eg.errs))
 	for i, err := range eg.errs {
 		b.WriteString(err.Error())
 		if i != len(eg.errs)-1 {

app/vmalert/vmalertutil/err_group_test.go

@@ -30,8 +30,8 @@ func TestErrGroup(t *testing.T) {
 	}
 
 	f(nil, "")
-	f([]error{errors.New("timeout")}, "errors(1): timeout")
-	f([]error{errors.New("timeout"), errors.New("deadline")}, "errors(2): timeout\ndeadline")
+	f([]error{errors.New("timeout")}, "errors(1): \ntimeout")
+	f([]error{errors.New("timeout"), errors.New("deadline")}, "errors(2): \ntimeout\ndeadline")
 }
 
 // TestErrGroupConcurrent supposed to test concurrent
@@ -42,7 +42,7 @@ func TestErrGroupConcurrent(_ *testing.T) {
 
 	const writersN = 4
 	payload := make(chan error, writersN)
-	for i := 0; i < writersN; i++ {
+	for range writersN {
 		go func() {
 			for err := range payload {
 				eg.Add(err)
@@ -51,7 +51,7 @@ func TestErrGroupConcurrent(_ *testing.T) {
 	}
 
 	const iterations = 500
-	for i := 0; i < iterations; i++ {
+	for i := range iterations {
 		payload <- fmt.Errorf("error %d", i)
 		if i%10 == 0 {
 			_ = eg.Err()

app/vmalert/web.qtpl

@@ -9,6 +9,7 @@
     "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/vmalertutil"
     "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
     "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/rule"
+    "github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 ) %}
 
 {% func Controls(prefix, currentIcon, currentText string, icons, filters map[string]string, search bool) %}
@@ -78,6 +79,8 @@
 {% func Welcome(r *http.Request) %}
     {%= tpl.Header(r, navItems, "vmalert", getLastConfigError()) %}
     <p>
+        Version {%s buildinfo.Version %} <br>
+
         API:<br>
         {% for _, p := range apiLinks  %}
             {%code p, doc := p[0], p[1] %}

app/vmauth/auth_config.go

@@ -13,6 +13,7 @@ import (
 	"net/url"
 	"os"
 	"regexp"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -28,6 +29,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs/fscore"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
@@ -65,10 +67,11 @@ type AuthConfig struct {
 type UserInfo struct {
 	Name string `yaml:"name,omitempty"`
 
-	BearerToken string `yaml:"bearer_token,omitempty"`
-	AuthToken   string `yaml:"auth_token,omitempty"`
-	Username    string `yaml:"username,omitempty"`
-	Password    string `yaml:"password,omitempty"`
+	BearerToken string     `yaml:"bearer_token,omitempty"`
+	JWT         *JWTConfig `yaml:"jwt,omitempty"`
+	AuthToken   string     `yaml:"auth_token,omitempty"`
+	Username    string     `yaml:"username,omitempty"`
+	Password    string     `yaml:"password,omitempty"`
 
 	URLPrefix              *URLPrefix  `yaml:"url_prefix,omitempty"`
 	DiscoverBackendIPs     *bool       `yaml:"discover_backend_ips,omitempty"`
@@ -89,6 +92,8 @@ type UserInfo struct {
 
 	MetricLabels map[string]string `yaml:"metric_labels,omitempty"`
 
+	AccessLog *AccessLog `yaml:"access_log,omitempty"`
+
 	concurrencyLimitCh      chan struct{}
 	concurrencyLimitReached *metrics.Counter
 
@@ -101,11 +106,40 @@ type UserInfo struct {
 	requestsDuration *metrics.Summary
 }
 
+// AccessLog represents configuration for access log settings.
+type AccessLog struct {
+	Filters *AccessLogFilters `yaml:"filters"`
+}
+
+// AccessLogFilters represents list of filters for access logs printing
+type AccessLogFilters struct {
+	// SkipStatusCodes is a list of HTTP status codes for which access logs will be skipped
+	SkipStatusCodes []int `yaml:"skip_status_codes"`
+}
+
+func (ui *UserInfo) logRequest(r *http.Request, userName string, statusCode int, duration time.Duration) {
+	if ui.AccessLog == nil {
+		return
+	}
+	filters := ui.AccessLog.Filters
+	if filters != nil && len(filters.SkipStatusCodes) > 0 {
+		if slices.Contains(filters.SkipStatusCodes, statusCode) {
+			return
+		}
+	}
+
+	remoteAddr := httpserver.GetQuotedRemoteAddr(r)
+	requestURI := httpserver.GetRequestURI(r)
+	logger.Infof("access_log request_host=%q request_uri=%q status_code=%d remote_addr=%s user_agent=%q referer=%q duration_ms=%d username=%q",
+		r.Host, requestURI, statusCode, remoteAddr, r.UserAgent(), r.Referer(), duration.Milliseconds(), userName)
+}
+
 // HeadersConf represents config for request and response headers.
 type HeadersConf struct {
-	RequestHeaders   []*Header `yaml:"headers,omitempty"`
-	ResponseHeaders  []*Header `yaml:"response_headers,omitempty"`
-	KeepOriginalHost *bool     `yaml:"keep_original_host,omitempty"`
+	RequestHeaders     []*Header `yaml:"headers,omitempty"`
+	ResponseHeaders    []*Header `yaml:"response_headers,omitempty"`
+	KeepOriginalHost   *bool     `yaml:"keep_original_host,omitempty"`
+	hasAnyPlaceHolders bool
 }
 
 func (ui *UserInfo) beginConcurrencyLimit(ctx context.Context) error {
@@ -113,10 +147,8 @@ func (ui *UserInfo) beginConcurrencyLimit(ctx context.Context) error {
 	case ui.concurrencyLimitCh <- struct{}{}:
 		return nil
 	default:
-		ui.concurrencyLimitReached.Inc()
-
-		// The per-user limit for the number of concurrent requests is reached.
-		// Wait until the currently executed requests are finished, so the current request could be executed.
+		// The number of concurrently executed requests for the given user equals the limit.
+		// Wait until some of the currently executed requests are finished, so the current request could be executed.
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10078
 		select {
 		case ui.concurrencyLimitCh <- struct{}{}:
@@ -124,6 +156,8 @@ func (ui *UserInfo) beginConcurrencyLimit(ctx context.Context) error {
 		case <-ctx.Done():
 			err := ctx.Err()
 			if errors.Is(err, context.DeadlineExceeded) {
+				// The current request couldn't be executed until the request timeout.
+				ui.concurrencyLimitReached.Inc()
 				return fmt.Errorf("cannot start executing the request during -maxQueueDuration=%s because %d concurrent requests from the user %s are executed",
 					*maxQueueDuration, ui.getMaxConcurrentRequests(), ui.name())
 			}
@@ -150,12 +184,22 @@ func (ui *UserInfo) stopHealthChecks() {
 	if ui == nil {
 		return
 	}
-	if ui.URLPrefix == nil {
-		return
-	}
 
-	bus := ui.URLPrefix.bus.Load()
-	bus.stopHealthChecks()
+	if ui.URLPrefix != nil {
+		bus := ui.URLPrefix.bus.Load()
+		bus.stopHealthChecks()
+	}
+	if ui.DefaultURL != nil {
+		bus := ui.DefaultURL.bus.Load()
+		bus.stopHealthChecks()
+	}
+	for i := range ui.URLMaps {
+		um := &ui.URLMaps[i]
+		if um.URLPrefix != nil {
+			bus := um.URLPrefix.bus.Load()
+			bus.stopHealthChecks()
+		}
+	}
 }
 
 // Header is `Name: Value` http header, which must be added to the proxied request.
@@ -338,6 +382,7 @@ func (bus *backendURLs) add(u *url.URL) {
 		url:                u,
 		healthCheckContext: bus.healthChecksContext,
 		healthCheckWG:      &bus.healthChecksWG,
+		hasPlaceHolders:    hasAnyPlaceholders(u),
 	})
 }
 
@@ -355,6 +400,8 @@ type backendURL struct {
 	concurrentRequests atomic.Int32
 
 	url *url.URL
+
+	hasPlaceHolders bool
 }
 
 func (bu *backendURL) isBroken() bool {
@@ -363,12 +410,10 @@ func (bu *backendURL) isBroken() bool {
 
 func (bu *backendURL) setBroken() {
 	if bu.broken.CompareAndSwap(false, true) {
-		bu.healthCheckWG.Add(1)
-		go func() {
-			defer bu.healthCheckWG.Done()
+		bu.healthCheckWG.Go(func() {
 			bu.runHealthCheck()
 			bu.broken.Store(false)
-		}()
+		})
 	}
 }
 
@@ -580,7 +625,7 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 
 	// Slow path - select other backend urls.
 	n := atomicCounter.Add(1) - 1
-	for i := uint32(0); i < uint32(len(bus)); i++ {
+	for i := range uint32(len(bus)) {
 		idx := (n + i) % uint32(len(bus))
 		bu := bus[idx]
 		if bu.isBroken() {
@@ -590,7 +635,7 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 		// The Load() in front of CompareAndSwap() avoids CAS overhead for items with values bigger than 0.
 		if bu.concurrentRequests.Load() == 0 && bu.concurrentRequests.CompareAndSwap(0, 1) {
 			atomicCounter.CompareAndSwap(n+1, idx+1)
-			// There is no need in the call bu.get(), because we already incremented bu.concrrentRequests above.
+			// There is no need in the call bu.get(), because we already incremented bu.concurrentRequests above.
 			return bu
 		}
 	}
@@ -733,11 +778,9 @@ func initAuthConfig() {
 	configTimestamp.Set(fasttime.UnixTimestamp())
 
 	stopCh = make(chan struct{})
-	authConfigWG.Add(1)
-	go func() {
-		defer authConfigWG.Done()
+	authConfigWG.Go(func() {
 		authConfigReloader(sighupCh)
-	}()
+	})
 }
 
 func stopAuthConfig() {
@@ -793,6 +836,9 @@ var (
 	// authUsers contains the currently loaded auth users
 	authUsers atomic.Pointer[map[string]*UserInfo]
 
+	// jwt authentication cache
+	jwtAuthCache atomic.Pointer[jwtCache]
+
 	authConfigWG sync.WaitGroup
 	stopCh       chan struct{}
 )
@@ -832,6 +878,16 @@ func reloadAuthConfigData(data []byte) (bool, error) {
 		return false, fmt.Errorf("failed to parse auth config: %w", err)
 	}
 
+	jui, oidcDP, err := parseJWTUsers(ac)
+	if err != nil {
+		return false, fmt.Errorf("failed to parse JWT users from auth config: %w", err)
+	}
+	oidcDP.startDiscovery()
+	jwtc := &jwtCache{
+		users:  jui,
+		oidcDP: oidcDP,
+	}
+
 	m, err := parseAuthConfigUsers(ac)
 	if err != nil {
 		return false, fmt.Errorf("failed to parse users from auth config: %w", err)
@@ -848,9 +904,15 @@ func reloadAuthConfigData(data []byte) (bool, error) {
 	}
 	metrics.RegisterSet(ac.ms)
 
+	jwtcPrev := jwtAuthCache.Load()
+	if jwtcPrev != nil {
+		jwtcPrev.oidcDP.stopDiscovery()
+	}
+
 	authConfig.Store(ac)
 	authConfigData.Store(&data)
 	authUsers.Store(&m)
+	jwtAuthCache.Store(jwtc)
 
 	return true, nil
 }
@@ -875,12 +937,18 @@ func parseAuthConfig(data []byte) (*AuthConfig, error) {
 		if ui.BearerToken != "" {
 			return nil, fmt.Errorf("field bearer_token can't be specified for unauthorized_user section")
 		}
+		if ui.JWT != nil {
+			return nil, fmt.Errorf("field jwt can't be specified for unauthorized_user section")
+		}
 		if ui.AuthToken != "" {
 			return nil, fmt.Errorf("field auth_token can't be specified for unauthorized_user section")
 		}
 		if ui.Name != "" {
 			return nil, fmt.Errorf("field name can't be specified for unauthorized_user section")
 		}
+		if err := parseJWTPlaceholdersForUserInfo(ui, false); err != nil {
+			return nil, err
+		}
 		if err := ui.initURLs(); err != nil {
 			return nil, err
 		}
@@ -921,16 +989,27 @@ func parseAuthConfigUsers(ac *AuthConfig) (map[string]*UserInfo, error) {
 	}
 	for i := range uis {
 		ui := &uis[i]
+		// users with jwt tokens are parsed by parseJWTUsers function.
+		// the function also checks that users with jwt tokens do not have auth tokens, bearer tokens, usernames and passwords.
+		if ui.JWT != nil {
+			continue
+		}
+
 		ats, err := getAuthTokens(ui.AuthToken, ui.BearerToken, ui.Username, ui.Password)
 		if err != nil {
 			return nil, err
 		}
+
 		for _, at := range ats {
 			if uiOld := byAuthToken[at]; uiOld != nil {
 				return nil, fmt.Errorf("duplicate auth token=%q found for username=%q, name=%q; the previous one is set for username=%q, name=%q",
 					at, ui.Username, ui.Name, uiOld.Username, uiOld.Name)
 			}
 		}
+
+		if err := parseJWTPlaceholdersForUserInfo(ui, false); err != nil {
+			return nil, err
+		}
 		if err := ui.initURLs(); err != nil {
 			return nil, err
 		}
@@ -1030,6 +1109,7 @@ func (ui *UserInfo) initURLs() error {
 			return err
 		}
 	}
+
 	for _, e := range ui.URLMaps {
 		if len(e.SrcPaths) == 0 && len(e.SrcHosts) == 0 && len(e.SrcQueryArgs) == 0 && len(e.SrcHeaders) == 0 {
 			return fmt.Errorf("missing `src_paths`, `src_hosts`, `src_query_args` and `src_headers` in `url_map`")
@@ -1089,6 +1169,9 @@ func (ui *UserInfo) name() string {
 		h := xxhash.Sum64([]byte(ui.AuthToken))
 		return fmt.Sprintf("auth_token:hash:%016X", h)
 	}
+	if ui.JWT != nil {
+		return `jwt`
+	}
 	return ""
 }
 

app/vmauth/auth_config_test.go

@@ -4,8 +4,11 @@ import (
 	"bytes"
 	"fmt"
 	"net"
+	"net/http"
 	"net/url"
+	"strings"
 	"testing"
+	"time"
 
 	"gopkg.in/yaml.v2"
 
@@ -276,6 +279,50 @@ users:
   url_prefix: http://foo.bar
   metric_labels:
     not-prometheus-compatible: value
+`)
+	// placeholder in url_prefix
+	f(`
+users:
+- username: foo
+  password: bar
+  url_prefix: 'http://ahost/{{a_placeholder}}/foobar'
+`)
+	// placeholder in a header
+	f(`
+users:
+- username: foo
+  password: bar
+  headers:
+  - 'X-Foo: {{a_placeholder}}'
+  url_prefix: 'http://ahost'
+`)
+	// placeholder in url_prefix
+	f(`
+users:
+- username: foo
+  password: bar
+  url_prefix: 'http://ahost/{{a_placeholder}}/foobar'
+`)
+	// placeholder in a header in url_map
+	f(`
+users:
+- username: foo
+  password: bar
+  url_map:
+    - src_paths: ["/select/.*"]
+      headers:
+        - 'X-Foo: {{a_placeholder}}'
+      url_prefix: 'http://ahost'
+`)
+
+	// placeholder in a header in url_map
+	f(`
+users:
+- username: foo
+  password: bar
+  url_map:
+    - src_paths: ["/select/.*"]
+      url_prefix: 'http://ahost/{{a_placeholder}}/foobar'
 `)
 }
 
@@ -378,7 +425,7 @@ users:
 			RetryStatusCodes:       []int{500, 501},
 			LoadBalancingPolicy:    "first_available",
 			MergeQueryArgs:         []string{"foo", "bar"},
-			DropSrcPathPrefixParts: intp(1),
+			DropSrcPathPrefixParts: new(1),
 			DiscoverBackendIPs:     &discoverBackendIPsTrue,
 		},
 	}, nil)
@@ -621,6 +668,47 @@ unauthorized_user:
 			},
 		},
 	})
+
+	// skip user info with jwt, it is parsed by parseJWTUsers
+	f(`
+users:
+- username: foo
+  password: bar
+  url_prefix: http://aaa:343/bbb
+- jwt: {skip_verify: true}
+  url_prefix: http://aaa:343/bbb
+`, map[string]*UserInfo{
+		getHTTPAuthBasicToken("foo", "bar"): {
+			Username:  "foo",
+			Password:  "bar",
+			URLPrefix: mustParseURL("http://aaa:343/bbb"),
+		},
+	}, nil)
+
+	// Multiple users with access logs enabled
+	f(`
+users:
+- username: foo
+  url_prefix: http://foo
+  access_log: {}
+- username: bar
+  url_prefix: https://bar/x/
+  access_log:
+    filters:
+      skip_status_codes: [404]
+`, map[string]*UserInfo{
+		getHTTPAuthBasicToken("foo", ""): {
+			Username:  "foo",
+			URLPrefix: mustParseURL("http://foo"),
+			AccessLog: &AccessLog{},
+		},
+		getHTTPAuthBasicToken("bar", ""): {
+			Username:  "bar",
+			URLPrefix: mustParseURL("https://bar/x/"),
+			AccessLog: &AccessLog{Filters: &AccessLogFilters{SkipStatusCodes: []int{404}}},
+		},
+	}, nil)
+
 }
 
 func TestParseAuthConfigPassesTLSVerificationConfig(t *testing.T) {
@@ -831,7 +919,7 @@ func TestBrokenBackend(t *testing.T) {
 	bus[1].setBroken()
 
 	// broken backend should never return while there are healthy backends
-	for i := 0; i < 1e3; i++ {
+	for range int(1e3) {
 		b := up.getBackendURL()
 		if b.isBroken() {
 			t.Fatalf("unexpected broken backend %q", b.url)
@@ -908,6 +996,41 @@ func TestDiscoverBackendIPsWithIPV6(t *testing.T) {
 
 }
 
+func TestLogRequest(t *testing.T) {
+	ui := &UserInfo{AccessLog: &AccessLog{}}
+
+	testOutput := &bytes.Buffer{}
+	logger.SetOutputForTests(testOutput)
+	defer logger.ResetOutputForTest()
+
+	req, err := http.NewRequest("GET", "http://localhost:8080/select/0/prometheus", nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+
+	f := func(user string, status int, duration time.Duration, expectedLog string) {
+		t.Helper()
+
+		testOutput.Reset()
+		ui.logRequest(req, user, status, duration)
+
+		got := testOutput.String()
+		if expectedLog == "" && got != "" {
+			t.Fatalf("expected empty log, got %q", got)
+		}
+		if !strings.Contains(got, expectedLog) {
+			t.Fatalf("output \n%q \nshould contain \n%q", testOutput.String(), expectedLog)
+		}
+	}
+
+	f("foo", 200, 10*time.Millisecond, `access_log request_host="localhost:8080" request_uri="" status_code=200 remote_addr="" user_agent="" referer="" duration_ms=10 username="foo"`)
+	f("foo", 404, time.Second, `access_log request_host="localhost:8080" request_uri="" status_code=404 remote_addr="" user_agent="" referer="" duration_ms=1000 username="foo"`)
+
+	ui.AccessLog.Filters = &AccessLogFilters{SkipStatusCodes: []int{200}}
+	f("foo", 200, 10*time.Millisecond, ``)
+	f("foo", 404, 10*time.Millisecond, `access_log request_host="localhost:8080" request_uri="" status_code=404 remote_addr="" user_agent="" referer="" duration_ms=10 username="foo"`)
+}
+
 func getRegexs(paths []string) []*Regex {
 	var sps []*Regex
 	for _, path := range paths {
@@ -963,10 +1086,6 @@ func mustParseURLs(us []string) *URLPrefix {
 	return up
 }
 
-func intp(n int) *int {
-	return &n
-}
-
 func mustNewRegex(s string) *Regex {
 	var re Regex
 	if err := yaml.Unmarshal([]byte(s), &re); err != nil {

app/vmauth/example_config.yml

@@ -116,6 +116,20 @@ users:
   - "http://default1:8888/unsupported_url_handler"
   - "http://default2:8888/unsupported_url_handler"
 
+# A JWT token based routing:
+# - Requests with JWT token that has the following structure:
+# {"team": "ops", "security": {"read_access": "1"}, "vm_access": {"metrics_account_id": 1000,"metrics_project_id":5}}
+# is routed to vmselect nodes and request url placeholder replaced with metrics tenant identificators
+- name: jwt-opts-team
+  jwt:
+    match_claims:
+     team: ops
+     security.read_access: "1"
+    skip_verify: true
+  url_prefix:
+  - "http://vmselect1:8481/select/{{.MetricsTenant}}/prometheus"
+  - "http://vmselect2:8481/select/{{.MetricsTenant}}/prometheus"
+
 # Requests without Authorization header are proxied according to `unauthorized_user` section.
 # Requests are proxied in round-robin fashion between `url_prefix` backends.
 # The deny_partial_response query arg is added to all the proxied requests.
@@ -125,3 +139,8 @@ unauthorized_user:
   - http://vmselect-az1/?deny_partial_response=1
   - http://vmselect-az2/?deny_partial_response=1
   retry_status_codes: [503, 500]
+  # log access for requests routed to this user
+  access_log:
+    filters:
+      # except requests with Status Codes below
+      skip_status_codes: [200, 202]

app/vmauth/jwt.go

@@ -0,0 +1,486 @@
+package main
+
+import (
+	"fmt"
+	"net/url"
+	"os"
+	"slices"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/jwt"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+const (
+	metricsTenantPlaceholder       = `{{.MetricsTenant}}`
+	metricsExtraLabelsPlaceholder  = `{{.MetricsExtraLabels}}`
+	metricsExtraFiltersPlaceholder = `{{.MetricsExtraFilters}}`
+
+	logsAccountIDPlaceholder          = `{{.LogsAccountID}}`
+	logsProjectIDPlaceholder          = `{{.LogsProjectID}}`
+	logsExtraFiltersPlaceholder       = `{{.LogsExtraFilters}}`
+	logsExtraStreamFiltersPlaceholder = `{{.LogsExtraStreamFilters}}`
+
+	placeholderPrefix = `{{`
+)
+
+var allPlaceholders = []string{
+	metricsTenantPlaceholder,
+	metricsExtraLabelsPlaceholder,
+	metricsExtraFiltersPlaceholder,
+	logsAccountIDPlaceholder,
+	logsProjectIDPlaceholder,
+	logsExtraFiltersPlaceholder,
+	logsExtraStreamFiltersPlaceholder,
+}
+
+var urlPathPlaceHolders = []string{
+	metricsTenantPlaceholder,
+	logsAccountIDPlaceholder,
+	logsProjectIDPlaceholder,
+}
+
+type jwtCache struct {
+	// users contain UserInfo`s from AuthConfig with JWTConfig set
+	users []*UserInfo
+
+	oidcDP *oidcDiscovererPool
+}
+
+type JWTConfig struct {
+	PublicKeys        []string          `yaml:"public_keys,omitempty"`
+	PublicKeyFiles    []string          `yaml:"public_key_files,omitempty"`
+	SkipVerify        bool              `yaml:"skip_verify,omitempty"`
+	OIDC              *oidcConfig       `yaml:"oidc,omitempty"`
+	MatchClaims       map[string]string `yaml:"match_claims,omitempty"`
+	parsedMatchClaims []*jwt.Claim
+
+	// verifierPool is used to verify JWT tokens.
+	// It is initialized from PublicKeys and/or PublicKeyFiles.
+	// In this case, it is initialized once at config reload and never updated until next reload
+	// In case of OIDC, it is initialized on config reload and periodically updated by discovery process.
+	verifierPool atomic.Pointer[jwt.VerifierPool]
+}
+
+func parseJWTUsers(ac *AuthConfig) ([]*UserInfo, *oidcDiscovererPool, error) {
+	jui := make([]*UserInfo, 0, len(ac.Users))
+	oidcDP := &oidcDiscovererPool{}
+
+	uniqClaims := make(map[string]*UserInfo)
+	var sortedClaims []string
+	for idx, ui := range ac.Users {
+		jwtToken := ui.JWT
+		if jwtToken == nil {
+			continue
+		}
+
+		if ui.AuthToken != "" || ui.BearerToken != "" || ui.Username != "" || ui.Password != "" {
+			return nil, nil, fmt.Errorf("auth_token, bearer_token, username and password cannot be specified if jwt is set")
+		}
+		if len(jwtToken.PublicKeys) == 0 && len(jwtToken.PublicKeyFiles) == 0 && !jwtToken.SkipVerify && jwtToken.OIDC == nil {
+			return nil, nil, fmt.Errorf("jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true")
+		}
+		var claimsString string
+		sortedClaims = sortedClaims[:0]
+		parsedClaims := make([]*jwt.Claim, 0, len(jwtToken.MatchClaims))
+		for ck, cv := range jwtToken.MatchClaims {
+			sortedClaims = append(sortedClaims, fmt.Sprintf("%s=%s", ck, cv))
+			pc, err := jwt.NewClaim(ck, cv)
+			if err != nil {
+				return nil, nil, fmt.Errorf("incorrect match claim, key=%q, value regex=%q: %w", ck, cv, err)
+			}
+			parsedClaims = append(parsedClaims, pc)
+		}
+		ui.JWT.parsedMatchClaims = parsedClaims
+		sort.Strings(sortedClaims)
+		claimsString = strings.Join(sortedClaims, ",")
+
+		if oldUI, ok := uniqClaims[claimsString]; ok {
+			return nil, nil, fmt.Errorf("duplicate match claims=%q found for name=%q at idx=%d; the previous one is set for name=%q", claimsString, ui.Name, idx, oldUI.Name)
+		}
+		uniqClaims[claimsString] = &ui
+		if len(jwtToken.PublicKeys) > 0 || len(jwtToken.PublicKeyFiles) > 0 {
+			keys := make([]any, 0, len(jwtToken.PublicKeys)+len(jwtToken.PublicKeyFiles))
+
+			for i := range jwtToken.PublicKeys {
+				k, err := jwt.ParseKey([]byte(jwtToken.PublicKeys[i]))
+				if err != nil {
+					return nil, nil, err
+				}
+				keys = append(keys, k)
+			}
+
+			for _, filePath := range jwtToken.PublicKeyFiles {
+				keyData, err := os.ReadFile(filePath)
+				if err != nil {
+					return nil, nil, fmt.Errorf("cannot read public key from file %q: %w", filePath, err)
+				}
+				k, err := jwt.ParseKey(keyData)
+				if err != nil {
+					return nil, nil, fmt.Errorf("cannot parse public key from file %q: %w", filePath, err)
+				}
+				keys = append(keys, k)
+			}
+
+			vp, err := jwt.NewVerifierPool(keys)
+			if err != nil {
+				return nil, nil, err
+			}
+
+			jwtToken.verifierPool.Store(vp)
+		}
+		if jwtToken.OIDC != nil {
+			if len(jwtToken.PublicKeys) > 0 || len(jwtToken.PublicKeyFiles) > 0 || jwtToken.SkipVerify {
+				return nil, nil, fmt.Errorf("jwt with oidc cannot contain public keys or have skip_verify=true")
+			}
+
+			if jwtToken.OIDC.Issuer == "" {
+				return nil, nil, fmt.Errorf("oidc issuer cannot be empty")
+			}
+			isserURL, err := url.Parse(jwtToken.OIDC.Issuer)
+			if err != nil {
+				return nil, nil, fmt.Errorf("oidc issuer %q must be a valid URL", jwtToken.OIDC.Issuer)
+			}
+			if isserURL.Scheme != "https" && isserURL.Scheme != "http" {
+				return nil, nil, fmt.Errorf("oidc issuer %q must have http or https scheme", jwtToken.OIDC.Issuer)
+			}
+
+			oidcDP.createOrAdd(ui.JWT.OIDC.Issuer, &ui.JWT.verifierPool)
+		}
+
+		if err := parseJWTPlaceholdersForUserInfo(&ui, true); err != nil {
+			return nil, nil, err
+		}
+
+		if err := ui.initURLs(); err != nil {
+			return nil, nil, err
+		}
+
+		metricLabels, err := ui.getMetricLabels()
+		if err != nil {
+			return nil, nil, fmt.Errorf("cannot parse metric_labels: %w", err)
+		}
+		ui.requests = ac.ms.GetOrCreateCounter(`vmauth_user_requests_total` + metricLabels)
+		ui.requestErrors = ac.ms.GetOrCreateCounter(`vmauth_user_request_errors_total` + metricLabels)
+		ui.backendRequests = ac.ms.GetOrCreateCounter(`vmauth_user_request_backend_requests_total` + metricLabels)
+		ui.backendErrors = ac.ms.GetOrCreateCounter(`vmauth_user_request_backend_errors_total` + metricLabels)
+		ui.requestsDuration = ac.ms.GetOrCreateSummary(`vmauth_user_request_duration_seconds` + metricLabels)
+		mcr := ui.getMaxConcurrentRequests()
+		ui.concurrencyLimitCh = make(chan struct{}, mcr)
+		ui.concurrencyLimitReached = ac.ms.GetOrCreateCounter(`vmauth_user_concurrent_requests_limit_reached_total` + metricLabels)
+		_ = ac.ms.GetOrCreateGauge(`vmauth_user_concurrent_requests_capacity`+metricLabels, func() float64 {
+			return float64(cap(ui.concurrencyLimitCh))
+		})
+		_ = ac.ms.GetOrCreateGauge(`vmauth_user_concurrent_requests_current`+metricLabels, func() float64 {
+			return float64(len(ui.concurrencyLimitCh))
+		})
+
+		rt, err := newRoundTripper(ui.TLSCAFile, ui.TLSCertFile, ui.TLSKeyFile, ui.TLSServerName, ui.TLSInsecureSkipVerify)
+		if err != nil {
+			return nil, nil, fmt.Errorf("cannot initialize HTTP RoundTripper: %w", err)
+		}
+		ui.rt = rt
+
+		jui = append(jui, &ui)
+	}
+
+	// sort by amount of matching claims
+	// it allows to more specific claim win in case of clash
+	sort.SliceStable(jui, func(i, j int) bool {
+		return len(jui[i].JWT.MatchClaims) > len(jui[j].JWT.MatchClaims)
+	})
+
+	return jui, oidcDP, nil
+}
+
+var tokenPool sync.Pool
+
+func getToken() *jwt.Token {
+	tkn := tokenPool.Get()
+	if tkn == nil {
+		return &jwt.Token{}
+	}
+	return tkn.(*jwt.Token)
+}
+
+func putToken(tkn *jwt.Token) {
+	tkn.Reset()
+	tokenPool.Put(tkn)
+}
+
+func getJWTUserInfo(ats []string) (*UserInfo, *jwt.Token) {
+	js := *jwtAuthCache.Load()
+	if len(js.users) == 0 {
+		return nil, nil
+	}
+
+	tkn := getToken()
+
+	for _, at := range ats {
+		if strings.Count(at, ".") != 2 {
+			continue
+		}
+
+		at, _ = strings.CutPrefix(at, `http_auth:`)
+		tkn.Reset()
+		if err := tkn.Parse(at, true); err != nil {
+			if *logInvalidAuthTokens {
+				logger.Infof("cannot parse jwt token: %s", err)
+			}
+			continue
+		}
+		if tkn.IsExpired(time.Now()) {
+			if *logInvalidAuthTokens {
+				// TODO: add more context:
+				// token claims with issuer
+				logger.Infof("jwt token is expired")
+			}
+			continue
+		}
+
+		if ui := getUserInfoByJWTToken(tkn, js.users); ui != nil {
+			return ui, tkn
+		}
+	}
+
+	putToken(tkn)
+	return nil, nil
+}
+
+func getUserInfoByJWTToken(tkn *jwt.Token, users []*UserInfo) *UserInfo {
+	for _, ui := range users {
+		if !tkn.MatchClaims(ui.JWT.parsedMatchClaims) {
+			continue
+		}
+
+		if ui.JWT.SkipVerify {
+			return ui
+		}
+
+		if ui.JWT.OIDC != nil {
+			// OIDC requires iss claim.
+			// It must match the discovery issuer URL set in OIDC config.
+			// https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata
+			if tkn.Issuer() == "" {
+				if *logInvalidAuthTokens {
+					logger.Infof("jwt token must have issuer filed")
+				}
+				return nil
+			}
+			if tkn.Issuer() != ui.JWT.OIDC.Issuer {
+				if *logInvalidAuthTokens {
+					logger.Infof("jwt token issuer: %q does not match oidc issuer: %q", tkn.Issuer(), ui.JWT.OIDC.Issuer)
+				}
+				return nil
+			}
+		}
+
+		vp := ui.JWT.verifierPool.Load()
+		if vp == nil {
+			if *logInvalidAuthTokens {
+				logger.Infof("jwt verifier not initialed")
+			}
+			return nil
+		}
+
+		if err := vp.Verify(tkn); err != nil {
+			if *logInvalidAuthTokens {
+				logger.Infof("cannot verify jwt token: %s", err)
+			}
+			return nil
+		}
+
+		return ui
+	}
+
+	if *logInvalidAuthTokens {
+		logger.Infof("no user match jwt token")
+	}
+
+	return nil
+}
+
+func replaceJWTPlaceholders(bu *backendURL, hc HeadersConf, vma *jwt.VMAccessClaim) (*url.URL, HeadersConf) {
+	if !bu.hasPlaceHolders && !hc.hasAnyPlaceHolders {
+		return bu.url, hc
+	}
+	targetURL := bu.url
+	data := jwtClaimsData(vma)
+	if bu.hasPlaceHolders {
+		// template url params and request path
+		// make a copy of url
+		uCopy := *bu.url
+		for _, uph := range urlPathPlaceHolders {
+			replacement := data[uph]
+			uCopy.Path = strings.ReplaceAll(uCopy.Path, uph, replacement[0])
+		}
+		query := uCopy.Query()
+		var foundAnyQueryPlaceholder bool
+		var templatedValues []string
+		for param, values := range query {
+			templatedValues = templatedValues[:0]
+			// filter in-place values with placeholders
+			// and accumulate replacements
+			// it will change the order of param values
+			// but it's not guaranteed
+			// and will be changed in any way with multiple arg templates
+			var cnt int
+			for _, value := range values {
+				if dv, ok := data[value]; ok {
+					foundAnyQueryPlaceholder = true
+					templatedValues = append(templatedValues, dv...)
+					continue
+				}
+				values[cnt] = value
+				cnt++
+			}
+			values = values[:cnt]
+			values = append(values, templatedValues...)
+			query[param] = values
+		}
+		if foundAnyQueryPlaceholder {
+			uCopy.RawQuery = query.Encode()
+		}
+		targetURL = &uCopy
+	}
+	if hc.hasAnyPlaceHolders {
+		// make a copy of headers and update only values with placeholder
+		rhs := make([]*Header, 0, len(hc.RequestHeaders))
+		for _, rh := range hc.RequestHeaders {
+			if dv, ok := data[rh.Value]; ok {
+				rh := &Header{
+					Name:  rh.Name,
+					Value: strings.Join(dv, ","),
+				}
+				rhs = append(rhs, rh)
+				continue
+			}
+			rhs = append(rhs, rh)
+		}
+		hc.RequestHeaders = rhs
+	}
+
+	return targetURL, hc
+}
+
+func jwtClaimsData(vma *jwt.VMAccessClaim) map[string][]string {
+	data := map[string][]string{
+		// TODO: optimize at parsing stage
+		metricsTenantPlaceholder:       {fmt.Sprintf("%d:%d", vma.MetricsAccountID, vma.MetricsProjectID)},
+		metricsExtraLabelsPlaceholder:  vma.MetricsExtraLabels,
+		metricsExtraFiltersPlaceholder: vma.MetricsExtraFilters,
+
+		// TODO: optimize at parsing stage
+		logsAccountIDPlaceholder:          {fmt.Sprintf("%d", vma.LogsAccountID)},
+		logsProjectIDPlaceholder:          {fmt.Sprintf("%d", vma.LogsProjectID)},
+		logsExtraFiltersPlaceholder:       vma.LogsExtraFilters,
+		logsExtraStreamFiltersPlaceholder: vma.LogsExtraStreamFilters,
+	}
+	return data
+}
+
+func parseJWTPlaceholdersForUserInfo(ui *UserInfo, isAllowed bool) error {
+	if ui.URLPrefix != nil {
+		if err := validateJWTPlaceholdersForURL(ui.URLPrefix, isAllowed); err != nil {
+			return err
+		}
+	}
+	if err := parsePlaceholdersForHC(&ui.HeadersConf, isAllowed); err != nil {
+		return err
+	}
+	if ui.DefaultURL != nil {
+		if err := validateJWTPlaceholdersForURL(ui.DefaultURL, isAllowed); err != nil {
+			return fmt.Errorf("invalid `default_url` placeholders: %w", err)
+		}
+	}
+	for i := range ui.URLMaps {
+		e := &ui.URLMaps[i]
+		if e.URLPrefix != nil {
+			if err := validateJWTPlaceholdersForURL(e.URLPrefix, isAllowed); err != nil {
+				return fmt.Errorf("invalid `url_map` `url_prefix` placeholders: %w", err)
+			}
+		}
+		if err := parsePlaceholdersForHC(&e.HeadersConf, isAllowed); err != nil {
+			return fmt.Errorf("invalid `url_map` headers placeholders: %w", err)
+		}
+
+	}
+	return nil
+}
+
+func validateJWTPlaceholdersForURL(up *URLPrefix, isAllowed bool) error {
+	for _, bu := range up.busOriginal {
+		ok := strings.Contains(bu.Path, placeholderPrefix)
+		if ok && !isAllowed {
+			return fmt.Errorf("placeholder: %q is only allowed at JWT token context", bu.Path)
+		}
+		if ok {
+			p := bu.Path
+			for _, ph := range allPlaceholders {
+				p = strings.ReplaceAll(p, ph, ``)
+			}
+			if strings.Contains(p, placeholderPrefix) {
+				return fmt.Errorf("invalid placeholder found in URL request path: %q, supported values are: %s", bu.Path, strings.Join(allPlaceholders, ", "))
+
+			}
+		}
+		for param, values := range bu.Query() {
+			for _, value := range values {
+				ok := strings.Contains(value, placeholderPrefix)
+				if ok && !isAllowed {
+					return fmt.Errorf("query param: %q with placeholder: %q is only allowed at JWT token context", param, value)
+				}
+				if ok {
+					// possible placeholder
+					if !slices.Contains(allPlaceholders, value) {
+						return fmt.Errorf("query param: %q has unsupported placeholder string: %q, supported values are: %s", param, value, strings.Join(allPlaceholders, ", "))
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func parsePlaceholdersForHC(hc *HeadersConf, isAllowed bool) error {
+	for _, rhs := range hc.RequestHeaders {
+		ok := strings.Contains(rhs.Value, placeholderPrefix)
+		if ok && !isAllowed {
+			return fmt.Errorf("request header: %q placeholder: %q is only supported at JWT context", rhs.Name, rhs.Value)
+		}
+		if ok {
+			if !slices.Contains(allPlaceholders, rhs.Value) {
+				return fmt.Errorf("request header: %q has unsupported placeholder: %q, supported values are: %s", rhs.Name, rhs.Value, strings.Join(allPlaceholders, ", "))
+			}
+			hc.hasAnyPlaceHolders = true
+		}
+	}
+	for _, rhs := range hc.ResponseHeaders {
+		if strings.Contains(rhs.Value, placeholderPrefix) {
+			return fmt.Errorf("response header placeholders are not supported; found placeholder prefix at header: %q with value: %q", rhs.Name, rhs.Value)
+		}
+	}
+	return nil
+}
+
+func hasAnyPlaceholders(u *url.URL) bool {
+	if strings.Contains(u.Path, placeholderPrefix) {
+		return true
+	}
+	if len(u.Query()) == 0 {
+		return false
+	}
+	for _, values := range u.Query() {
+		for _, value := range values {
+			if strings.HasPrefix(value, placeholderPrefix) {
+				return true
+			}
+		}
+
+	}
+	return false
+}

app/vmauth/jwt_test.go

@@ -0,0 +1,503 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestJWTParseAuthConfigFailure(t *testing.T) {
+	validRSAPublicKey := `-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAiX7oPWKOWRQsGFEWvwZO
+mL2PYsdYUsu9nr0qtPCjxQHUJgLfT3rdKlvKpPFYv7ZmKnqTncg36Wz9uiYmWJ7e
+IB5Z+fko8kVIMzarCqVvpAJDzYF/pUii68xvuYoK3L9TIOAeyCXv+prwnr2IH+Mw
+9AONzWbRrYoO74XyTE9vMU5qmI/L1VPk+PR8lqPOSptLvzsfoaIk2ED4yK2nRB+6
+st+k4nccPqbErqHc8aiXnXfugfnr6b+NPFYUzKsDqkymGOokVijrI8B3jNw6c6Do
+zphk+D3wgLsXYHfMcZbXIMqffqm/aB8Qg88OpFOkQ3rd2p6R9+hacnZkfkn3Phiw
+yQIDAQAB
+-----END PUBLIC KEY-----
+`
+	// ECDSA with the P-521 curve
+	validECDSAPublicKey := `-----BEGIN PUBLIC KEY-----
+MIGbMBAGByqGSM49AgEGBSuBBAAjA4GGAAQAU9RmtkCRuYTKCyvLlDn5DtBZOHSe
+QTa5j9q/oQVpCKqcXVFrH5dgh0GL+P/ZhkeuowPzCZqntGf0+7wPt9OxSJcADVJm
+dv92m540MXss8zdHf5qtE0gsu2Ved0R7Z8a8QwGZ/1mYZ+kFGGbdQTlSvRqDySTq
+XOtclIk1uhc03oL9nOQ=
+-----END PUBLIC KEY-----
+`
+
+	f := func(s string, expErr string) {
+		t.Helper()
+		ac, err := parseAuthConfig([]byte(s))
+		if err != nil {
+			if expErr != err.Error() {
+				t.Fatalf("unexpected error; got\n%q\nwant\n%q", err.Error(), expErr)
+			}
+			return
+		}
+		users, oidcDP, err := parseJWTUsers(ac)
+		if err == nil {
+			t.Fatalf("expecting non-nil error; got %v", users)
+		}
+		if expErr != err.Error() {
+			t.Fatalf("unexpected error; got\n%q\nwant \n%q", err.Error(), expErr)
+		}
+		if oidcDP != nil {
+			t.Fatalf("expecting nil oidcDP; got %v", oidcDP)
+		}
+	}
+
+	// unauthorized_user cannot be used with jwt
+	f(`
+unauthorized_user:
+  jwt: {skip_verify: true}
+  url_prefix: http://foo.bar
+`, `field jwt can't be specified for unauthorized_user section`)
+
+	// username and jwt in a single config
+	f(`
+users:
+- username: foo
+  jwt: {skip_verify: true}
+  url_prefix: http://foo.bar
+`, `auth_token, bearer_token, username and password cannot be specified if jwt is set`)
+	// bearer_token and jwt in a single config
+	f(`
+users:
+- bearer_token: foo
+  jwt: {skip_verify: true}
+  url_prefix: http://foo.bar
+`, `auth_token, bearer_token, username and password cannot be specified if jwt is set`)
+	// bearer_token and jwt in a single config
+	f(`
+users:
+- auth_token: "Foo token"
+  jwt: {skip_verify: true}
+  url_prefix: http://foo.bar
+`, `auth_token, bearer_token, username and password cannot be specified if jwt is set`)
+
+	// jwt public_keys or skip_verify must be set, part 1
+	f(`
+users:
+- jwt: {}
+  url_prefix: http://foo.bar
+`, `jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true`)
+
+	// jwt public_keys or skip_verify must be set, part 2
+	f(`
+users:
+- jwt: {public_keys: null}
+  url_prefix: http://foo.bar
+`, `jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true`)
+
+	// jwt public_keys or skip_verify must be set, part 3
+	f(`
+users:
+- jwt: {public_keys: []}
+  url_prefix: http://foo.bar
+`, `jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true`)
+
+	// jwt public_keys, public_key_files or skip_verify must be set
+	f(`
+users:
+- jwt: {public_key_files: []}
+  url_prefix: http://foo.bar
+`, `jwt must contain at least a single public key, public_key_files, oidc or have skip_verify=true`)
+
+	// invalid public key, part 1
+	f(`
+users:
+- jwt: {public_keys: [""]}
+  url_prefix: http://foo.bar
+`, `failed to parse key "": failed to decode PEM block containing public key`)
+
+	// invalid public key, part 2
+	f(`
+users:
+- jwt: {public_keys: ["invalid"]}
+  url_prefix: http://foo.bar
+`, `failed to parse key "invalid": failed to decode PEM block containing public key`)
+
+	// invalid public key, part 2
+	f(fmt.Sprintf(`
+users:
+- jwt: 
+    public_keys:
+    - %q
+    - %q
+    - "invalid"
+  url_prefix: http://foo.bar
+`, validRSAPublicKey, validECDSAPublicKey), `failed to parse key "invalid": failed to decode PEM block containing public key`)
+
+	// several jwt users
+	// invalid public key, part 2
+	f(fmt.Sprintf(`
+users:
+- jwt: 
+    public_keys:
+    - %q
+  url_prefix: http://foo.bar
+- jwt:
+    public_keys:
+    - %q
+  url_prefix: http://foo.bar
+`, validRSAPublicKey, validECDSAPublicKey), `duplicate match claims="" found for name="" at idx=1; the previous one is set for name=""`)
+
+	// public key file doesn't exist
+	f(`
+users:
+- jwt: 
+    public_key_files: 
+    - /path/to/nonexistent/file.pem
+  url_prefix: http://foo.bar
+`, "cannot read public key from file \"/path/to/nonexistent/file.pem\": open /path/to/nonexistent/file.pem: no such file or directory")
+
+	// public key file invalid
+	// auth with key from file
+	publicKeyFile := filepath.Join(t.TempDir(), "a_public_key.pem")
+	if err := os.WriteFile(publicKeyFile, []byte(`invalidPEM`), 0o644); err != nil {
+		t.Fatalf("failed to write public key file: %s", err)
+	}
+	f(`
+users:
+- jwt: 
+    public_key_files: 
+    - `+publicKeyFile+`
+  url_prefix: http://foo.bar
+`, "cannot parse public key from file \""+publicKeyFile+"\": failed to parse key \"invalidPEM\": failed to decode PEM block containing public key")
+
+	// unsupported placeholder in a header
+	f(`
+users:
+- jwt: 
+    skip_verify: true
+  url_prefix: http://foo.bar/{{.UnsupportedPlaceholder}}/foo`,
+		"invalid placeholder found in URL request path: \"/{{.UnsupportedPlaceholder}}/foo\", supported values are: {{.MetricsTenant}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
+	)
+	// unsupported placeholder in a header
+	f(`
+users:
+- jwt: 
+    skip_verify: true
+  headers:
+    - "AccountID: {{.UnsupportedPlaceholder}}"
+  url_prefix: http://foo.bar
+`,
+		"request header: \"AccountID\" has unsupported placeholder: \"{{.UnsupportedPlaceholder}}\", supported values are: {{.MetricsTenant}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
+	)
+
+	// spaces in templating not allowed
+	f(`
+users:
+- jwt: 
+    skip_verify: true
+  headers:
+    - "AccountID: {{ .LogsAccountID }}"
+  url_prefix: http://foo.bar
+`,
+		"request header: \"AccountID\" has unsupported placeholder: \"{{ .LogsAccountID }}\", supported values are: {{.MetricsTenant}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
+	)
+
+	// oidc is not an object
+	f(`
+users:
+- jwt: 
+    oidc: "not an object"
+  url_prefix: http://foo.bar
+`,
+		"cannot unmarshal AuthConfig data: yaml: unmarshal errors:\n  line 4: cannot unmarshal !!str `not an ...` into main.oidcConfig",
+	)
+
+	// oidc issuer empty
+	f(`
+users:
+- jwt: 
+    oidc: {}
+  url_prefix: http://foo.bar
+`,
+		"oidc issuer cannot be empty",
+	)
+
+	// oidc issuer invalid urls
+	f(`
+users:
+- jwt: 
+    oidc:
+      issuer: "::invalid-url"
+  url_prefix: http://foo.bar
+`,
+		"oidc issuer \"::invalid-url\" must be a valid URL",
+	)
+
+	// oidc issuer invalid urls
+	f(`
+users:
+- jwt: 
+    oidc:
+      issuer: "invalid-url"
+  url_prefix: http://foo.bar
+`,
+		"oidc issuer \"invalid-url\" must have http or https scheme",
+	)
+
+	// oidc and public_keys are not allowed
+	f(fmt.Sprintf(`
+users:
+- jwt: 
+    public_keys:
+    - %q
+    oidc: 
+      issuer: https://example.com
+  url_prefix: http://foo.bar
+`, validRSAPublicKey),
+		"jwt with oidc cannot contain public keys or have skip_verify=true",
+	)
+
+	// oidc and skip_verify are not allowed
+	f(`
+users:
+- jwt: 
+    skip_verify: true
+    oidc: 
+      issuer: https://example.com
+  url_prefix: http://foo.bar
+`,
+		"jwt with oidc cannot contain public keys or have skip_verify=true",
+	)
+	// duplicate claims
+	f(`
+users:
+- jwt:
+    skip_verify: true
+    match_claims:
+     team: ops
+  name: user-1
+  url_prefix: http://foo.bar
+- jwt:
+    skip_verify: true
+    match_claims:
+     team: ops
+  name: user-2
+  url_prefix: http://foo.bar`,
+		"duplicate match claims=\"team=ops\" found for name=\"user-2\" at idx=1; the previous one is set for name=\"user-1\"",
+	)
+}
+
+func TestJWTParseAuthConfigSuccess(t *testing.T) {
+	validRSAPublicKey := `-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAiX7oPWKOWRQsGFEWvwZO
+mL2PYsdYUsu9nr0qtPCjxQHUJgLfT3rdKlvKpPFYv7ZmKnqTncg36Wz9uiYmWJ7e
+IB5Z+fko8kVIMzarCqVvpAJDzYF/pUii68xvuYoK3L9TIOAeyCXv+prwnr2IH+Mw
+9AONzWbRrYoO74XyTE9vMU5qmI/L1VPk+PR8lqPOSptLvzsfoaIk2ED4yK2nRB+6
+st+k4nccPqbErqHc8aiXnXfugfnr6b+NPFYUzKsDqkymGOokVijrI8B3jNw6c6Do
+zphk+D3wgLsXYHfMcZbXIMqffqm/aB8Qg88OpFOkQ3rd2p6R9+hacnZkfkn3Phiw
+yQIDAQAB
+-----END PUBLIC KEY-----
+`
+	// ECDSA with the P-521 curve
+	validECDSAPublicKey := `-----BEGIN PUBLIC KEY-----
+MIGbMBAGByqGSM49AgEGBSuBBAAjA4GGAAQAU9RmtkCRuYTKCyvLlDn5DtBZOHSe
+QTa5j9q/oQVpCKqcXVFrH5dgh0GL+P/ZhkeuowPzCZqntGf0+7wPt9OxSJcADVJm
+dv92m540MXss8zdHf5qtE0gsu2Ved0R7Z8a8QwGZ/1mYZ+kFGGbdQTlSvRqDySTq
+XOtclIk1uhc03oL9nOQ=
+-----END PUBLIC KEY-----
+`
+
+	f := func(s string) {
+		t.Helper()
+		ac, err := parseAuthConfig([]byte(s))
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+
+		jui, oidcDP, err := parseJWTUsers(ac)
+		if err != nil {
+			t.Fatalf("unexpected error: %s", err)
+		}
+		oidcDP.startDiscovery()
+		defer oidcDP.stopDiscovery()
+
+		for _, ui := range jui {
+			if ui.JWT == nil {
+				t.Fatalf("unexpected nil JWTConfig")
+			}
+
+			if ui.JWT.SkipVerify {
+				if ui.JWT.verifierPool.Load() != nil {
+					t.Fatalf("unexpected non-nil verifier pool for skip_verify=true")
+				}
+				continue
+			}
+
+			if ui.JWT.verifierPool.Load() == nil {
+				t.Fatalf("unexpected nil verifier pool for non-empty public keys")
+			}
+		}
+	}
+
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+  url_prefix: http://foo.bar
+`, validRSAPublicKey))
+
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+  url_prefix: http://foo.bar
+`, validECDSAPublicKey))
+
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+    - %q
+  url_prefix: http://foo.bar
+`, validRSAPublicKey, validECDSAPublicKey))
+
+	f(`
+users:
+- jwt:
+    skip_verify: true
+  url_prefix: http://foo.bar
+`)
+
+	// combined with other auth methods
+	f(`
+users:
+- username: foo
+  password: bar
+  url_prefix: http://foo.bar
+
+- jwt:
+    skip_verify: true
+  url_prefix: http://foo.bar
+
+- bearer_token: foo
+  url_prefix: http://foo.bar
+`)
+
+	rsaKeyFile := filepath.Join(t.TempDir(), "rsa_public_key.pem")
+	if err := os.WriteFile(rsaKeyFile, []byte(validRSAPublicKey), 0o644); err != nil {
+		t.Fatalf("failed to write RSA key file: %s", err)
+	}
+	ecdsaKeyFile := filepath.Join(t.TempDir(), "ecdsa_public_key.pem")
+	if err := os.WriteFile(ecdsaKeyFile, []byte(validECDSAPublicKey), 0o644); err != nil {
+		t.Fatalf("failed to write ECDSA key file: %s", err)
+	}
+
+	// Test single public key file
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_key_files:
+    - %q
+  url_prefix: http://foo.bar
+`, rsaKeyFile))
+
+	// Test multiple public key files
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_key_files:
+    - %q
+    - %q
+  url_prefix: http://foo.bar
+`, rsaKeyFile, ecdsaKeyFile))
+
+	// Test combined inline keys and files
+	f(fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+    public_key_files:
+    - %q
+  url_prefix: http://foo.bar
+`, validECDSAPublicKey, rsaKeyFile))
+
+	// oidc stub server
+	var ipSrv *httptest.Server
+	ipSrv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/.well-known/openid-configuration" {
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(map[string]string{
+				"issuer":   ipSrv.URL,
+				"jwks_uri": fmt.Sprintf("%s/jwks", ipSrv.URL),
+			})
+			return
+		}
+		if r.URL.Path == "/jwks" {
+			// resp generated by https://jwkset.com/generate
+			w.Header().Set("Content-Type", "application/json")
+			w.Write([]byte(`
+{
+  "keys": [
+    {
+      "kty": "RSA",
+      "kid": "f13eee91-f566-4829-80fa-fca847c21f0e",
+      "d": "Ua1llEFz3LZ05CrK5a2JxKMUEWJGXhBPPF20hHQjzxd1w0IEJK_mhPZQG8dNtBROBNIi1FC9l6QRw-RTnVIVat5Xy4yDFNKXXL3ZLXejOHY8SXrNEIDqQ-cSwIpK9cK7Umib0PcPeEeeAED5mqDH75D8_YssWFF18kLbNB5Z9pZmn6Fshiht7l2Sh4GN-KcReOW6eiQQwckDte3OGmZCRbtEriLWJt5TUGUvfZVIlcclqNMycNB6jGa9E1pO5Up7Ki3ZbI_-6XmRgZPtqnR9oLJ1zn3fj3hYpCXo-zcqLuOu3qxcslsq5igsfBzgGtfIJHY9LfWmHUsaDEa5cAX1gQ",
+      "n": "xbLXXBTNREk70UCMiqZ53_mTzYh89W-UaPU61GZ-RZ5lYcLgyWOb5mdyRbvJpcgfZpsOeGAUWbk3GkQ4vqn8kUMnnWhUum2Qk9kGubOJGLW6yaURd00j3E-ilQ5xO2R_Hzz8bAojxV8GKdGTQ-iTf8z8nsSHH8kR2SERbNJCFFtwtFU7vyFWyoH4Lmvu2UpICTHFCR9RqwQVjyoKB1JjJ6Dh1L4zPTlsvQEnqoeFQHPYr0QcQSMYXdfPvlt_FiLOAOE89fX_9T2r9WbFAoda3uTRE5_aal0jxUU2cFyeVSIgauNtF07fp422XFb4XPkWQWrdNx0KX53laSIYQ9HOpw",
+      "e": "AQAB",
+      "p": "2JT57AD-Q2lamgjgyn0wL7DgYZ3OoCTTrDm5_NHg6h13uDvyIlXSukuUeWm4tzPSDedpstbS7dgXkLw5eQXBHwPYtByTcEZS8Z37CBnhMOOhfo_U1aNIPPanJACvWBgz47-TxHsxW1YhztZqghRoicBZPSSBAj49MgANJ4jF0zc",
+      "q": "6a4MkeSXJI-ZzQ-bgP8hwJqpLFr0AiNGQcjZMH4Nn4CPGdnGiqqe6flhfLimgbNhbb67B0-8fLIji8zGhGKDL_JSIpAAdmfs2vzeEsY2hScrqVbd1VbfRcRh0J6lsn7obxkbvQthp9sX2DQbeDcEeaFEvd9gDKQSATYEqWo7eBE",
+      "dp": "haL2yu6Z9RJuuxi7S3YPY33qFZF_y0St71j3L854zzw7gMxMTW9TRWwZQwk-1pv9AmNFzvnK0MNDVyUs-UXZsb932TrApshdqYRnPsppLvdl0GgDVYcYrbUr0IUzrFHSwraVAOlavRbaaXvX4EejcUvkRFvf1nh83fs2Iqy8E-U",
+      "dq": "Cnf5qC-Ndd3ZDg688LJ9WJuVKJ-Kfu4Fn7zXvgxnn9Wqk4XmFyA9rk21yFidXQIkQz5gMpun3g48-W5bFmMzbVp1w4af_q35NnZNnJm0p5Jxqkxx87TIm9-IYkg5NB3rW87MJ1PzNAnkr5LmCCSu1qQa6Eaxjt9qzxMUcmKH94E",
+      "qi": "saAeU11iaKHmye3cwCAYkegcyWbXV3xIXEVJtS9Af_yM19UhspwY2VhuwRaajcwYZwtvR9_ITmX9M-ea7uLdd7aDYO1fujC8NGbopeC4Hkr7yb5vTly3pfKf4h-3LwGGUucJUetdz1lmMIYiyuG4_gSf1yIEtPDLKzXiedgEMdI"
+    }
+  ]
+}
+`))
+			return
+		}
+
+		http.NotFound(w, r)
+	}))
+	defer ipSrv.Close()
+
+	f(`
+users:
+- jwt:
+    oidc:
+      issuer: ` + ipSrv.URL + `
+  url_prefix: http://foo.bar
+`)
+	// multiple match claims
+	f(fmt.Sprintf(`
+users:
+- jwt:
+   match_claims:
+    role: ro
+    team: dev
+   public_keys:
+    - %q
+  url_prefix: http://foo.bar
+- jwt:
+   match_claims:
+      role: admin
+      team: dev
+   public_key_files:
+    - %q
+    - %q
+  url_prefix: http://foo.bar
+- jwt:
+   match_claims:
+     role: viewer
+     team: dev
+     department: ceo
+   skip_verify: true
+  url_prefix: http://foo.bar
+
+
+`, validRSAPublicKey, rsaKeyFile, ecdsaKeyFile))
+
+}

app/vmauth/main.go

@@ -16,6 +16,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/jwt"
 	"github.com/VictoriaMetrics/metrics"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
@@ -24,6 +25,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/ioutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
@@ -40,27 +42,38 @@ var (
 	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")
-	maxIdleConnsPerBackend = flag.Int("maxIdleConnsPerBackend", 100, "The maximum number of idle connections vmauth can open per each backend host. "+
-		"See also -maxConcurrentRequests")
-	idleConnTimeout = flag.Duration("idleConnTimeout", 50*time.Second, "The timeout for HTTP keep-alive connections to backend services. "+
+	maxIdleConnsPerBackend = flag.Int("maxIdleConnsPerBackend", 100, "The maximum number of idle connections vmauth can open per each backend host")
+	idleConnTimeout        = flag.Duration("idleConnTimeout", 50*time.Second, "The timeout for HTTP keep-alive connections to backend services. "+
 		"It is recommended setting this value to values smaller than -http.idleConnTimeout set at backend services")
 	responseTimeout = flag.Duration("responseTimeout", 5*time.Minute, "The timeout for receiving a response from backend")
 
-	maxConcurrentRequests = flag.Int("maxConcurrentRequests", 1000, "The maximum number of concurrent requests vmauth can process. Other requests are rejected with "+
-		"'429 Too Many Requests' http status code. See also -maxQueueDuration, -maxConcurrentPerUserRequests and -maxIdleConnsPerBackend command-line options")
-	maxConcurrentPerUserRequests = flag.Int("maxConcurrentPerUserRequests", 300, "The maximum number of concurrent requests vmauth can process per each configured user. "+
-		"Other requests are rejected with '429 Too Many Requests' http status code. See also -maxQueueDuration and -maxConcurrentRequests command-line options "+
-		"and max_concurrent_requests option in per-user config")
-	maxQueueDuration = flag.Duration("maxQueueDuration", 10*time.Second, "The maximum duration the request waits for execution when the number of concurrently executed "+
-		"requests reach -maxConcurrentRequests or -maxConcurrentPerUserRequests before returning '429 Too Many Requests' error. "+
-		"This allows graceful handling of short spikes in the number of concurrent requests")
+	requestBufferSize = flagutil.NewBytes("requestBufferSize", 32*1024, "The size of the buffer for reading the request body before proxying the request to backends. "+
+		"This allows reducing the consumption of backend resources when processing requests from clients connected via slow networks. "+
+		"Set to 0 to disable request buffering. See https://docs.victoriametrics.com/victoriametrics/vmauth/#request-body-buffering")
+	maxRequestBodySizeToRetry = flagutil.NewBytes("maxRequestBodySizeToRetry", 16*1024, "The maximum request body size to buffer in memory for potential retries at other backends. "+
+		"Request bodies larger than this size cannot be retried if the backend fails. Zero or negative value disables request body buffering and retries. "+
+		"See also -requestBufferSize")
+
+	maxConcurrentRequests = flag.Int("maxConcurrentRequests", 1000, "The maximum number of concurrent requests vmauth can process simultaneously. "+
+		"Requests exceeding this limit are queued for up to -maxQueueDuration and then rejected with '429 Too Many Requests' http status code if the limit is still reached. "+
+		"This protects vmauth itself from overloading and out-of-memory (OOM) failures. See also -maxConcurrentPerUserRequests "+
+		"and https://docs.victoriametrics.com/victoriametrics/vmauth/#concurrency-limiting")
+	maxConcurrentPerUserRequests = flag.Int("maxConcurrentPerUserRequests", 100, "The maximum number of concurrent requests vmauth can process per each configured user. "+
+		"Requests exceeding this limit are queued for up to -maxQueueDuration and then rejected with '429 Too Many Requests' http status code if the limit is still reached. "+
+		"This provides fairness and isolation between users, preventing a single user from consuming all the available resources. "+
+		"It works in conjunction with -maxConcurrentRequests, which sets the global limit across all users. "+
+		"This default can be overridden for individual users via max_concurrent_requests option in per-user config. "+
+		"See https://docs.victoriametrics.com/victoriametrics/vmauth/#concurrency-limiting")
+	maxQueueDuration = flag.Duration("maxQueueDuration", 10*time.Second, "The maximum duration to wait before rejecting incoming requests if concurrency limit "+
+		"specified via -maxConcurrentRequests or -maxConcurrentPerUserRequests command-line flags is reached. "+
+		"Requests are rejected with '429 Too Many Requests' http status code if the limit is still reached after the -maxQueueDuration duration. "+
+		"This allows graceful handling of short spikes in concurrent requests. See https://docs.victoriametrics.com/victoriametrics/vmauth/#concurrency-limiting")
 
 	reloadAuthKey        = flagutil.NewPassword("reloadAuthKey", "Auth key for /-/reload http endpoint. It must be passed via authKey query arg. It overrides -httpAuth.*")
 	logInvalidAuthTokens = flag.Bool("logInvalidAuthTokens", false, "Whether to log requests with invalid auth tokens. "+
 		`Such requests are always counted at vmauth_http_request_errors_total{reason="invalid_auth_token"} metric, which is exposed at /metrics page`)
-	failTimeout               = flag.Duration("failTimeout", 3*time.Second, "Sets a delay period for load balancing to skip a malfunctioning backend")
-	maxRequestBodySizeToRetry = flagutil.NewBytes("maxRequestBodySizeToRetry", 16*1024, "The maximum request body size, which can be cached and re-tried at other backends. "+
-		"Bigger values may require more memory. Zero or negative value disables caching of request body. This may be useful when proxying data ingestion requests")
+	failTimeout = flag.Duration("failTimeout", 3*time.Second, "Sets a delay period for load balancing to skip a malfunctioning backend")
+
 	backendTLSInsecureSkipVerify = flag.Bool("backend.tlsInsecureSkipVerify", false, "Whether to skip TLS verification when connecting to backends over HTTPS. "+
 		"See https://docs.victoriametrics.com/victoriametrics/vmauth/#backend-tls-setup")
 	backendTLSCAFile = flag.String("backend.TLSCAFile", "", "Optional path to TLS root CA file, which is used for TLS verification when connecting to backends over HTTPS. "+
@@ -161,7 +174,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		// Process requests for unauthorized users
 		ui := authConfig.Load().UnauthorizedUser
 		if ui != nil {
-			processUserRequest(w, r, ui)
+			processUserRequest(w, r, ui, nil)
 			return true
 		}
 
@@ -169,29 +182,36 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	}
 
-	ui := getUserInfoByAuthTokens(ats)
-	if ui == nil {
-		uu := authConfig.Load().UnauthorizedUser
-		if uu != nil {
-			processUserRequest(w, r, uu)
-			return true
-		}
-
-		invalidAuthTokenRequests.Inc()
-		if *logInvalidAuthTokens {
-			err := fmt.Errorf("cannot authorize request with auth tokens %q", ats)
-			err = &httpserver.ErrorWithStatusCode{
-				Err:        err,
-				StatusCode: http.StatusUnauthorized,
-			}
-			httpserver.Errorf(w, r, "%s", err)
-		} else {
-			http.Error(w, "Unauthorized", http.StatusUnauthorized)
+	if ui := getUserInfoByAuthTokens(ats); ui != nil {
+		processUserRequest(w, r, ui, nil)
+		return true
+	}
+	if ui, tkn := getJWTUserInfo(ats); ui != nil {
+		if tkn == nil {
+			logger.Panicf("BUG: unexpected nil jwt token for user %q", ui.name())
 		}
+		defer putToken(tkn)
+		processUserRequest(w, r, ui, tkn)
 		return true
 	}
 
-	processUserRequest(w, r, ui)
+	uu := authConfig.Load().UnauthorizedUser
+	if uu != nil {
+		processUserRequest(w, r, uu, nil)
+		return true
+	}
+
+	invalidAuthTokenRequests.Inc()
+	if *logInvalidAuthTokens {
+		err := fmt.Errorf("cannot authorize request with auth tokens %q", ats)
+		err = &httpserver.ErrorWithStatusCode{
+			Err:        err,
+			StatusCode: http.StatusUnauthorized,
+		}
+		httpserver.Errorf(w, r, "%s", err)
+	} else {
+		http.Error(w, "Unauthorized", http.StatusUnauthorized)
+	}
 	return true
 }
 
@@ -206,7 +226,37 @@ func getUserInfoByAuthTokens(ats []string) *UserInfo {
 	return nil
 }
 
-func processUserRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
+// responseWriterWithStatus is a wrapper around http.ResponseWriter that captures the status code written to the response.
+type responseWriterWithStatus struct {
+	http.ResponseWriter
+	status int
+}
+
+// WriteHeader records the status so it can be easily retrieved later
+func (rws *responseWriterWithStatus) WriteHeader(status int) {
+	rws.status = status
+	rws.ResponseWriter.WriteHeader(status)
+}
+
+// Flush implements net/http.Flusher interface
+//
+// This is needed for the copyStreamToClient()
+func (rws *responseWriterWithStatus) Flush() {
+	flusher, ok := rws.ResponseWriter.(http.Flusher)
+	if !ok {
+		logger.Panicf("BUG: it is expected http.ResponseWriter (%T) supports http.Flusher interface", rws.ResponseWriter)
+	}
+	flusher.Flush()
+}
+
+// Unwrap returns the original ResponseWriter wrapped by rws.
+//
+// This is needed for the net/http.ResponseController - see https://pkg.go.dev/net/http#NewResponseController
+func (rws *responseWriterWithStatus) Unwrap() http.ResponseWriter {
+	return rws.ResponseWriter
+}
+
+func processUserRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo, tkn *jwt.Token) {
 	startTime := time.Now()
 	defer ui.requestsDuration.UpdateDuration(startTime)
 
@@ -215,49 +265,132 @@ func processUserRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 	ctx, cancel := context.WithTimeout(r.Context(), *maxQueueDuration)
 	defer cancel()
 
-	// Limit the concurrency of requests to backends
+	userName := ui.name()
+	if userName == "" {
+		userName = "unauthorized"
+	}
+
+	if ui.AccessLog != nil {
+		w = &responseWriterWithStatus{ResponseWriter: w}
+		defer func() {
+			rws := w.(*responseWriterWithStatus)
+			duration := time.Since(startTime)
+			ui.logRequest(r, userName, rws.status, duration)
+		}()
+	}
+
+	// Acquire global concurrency limit.
+	if err := beginConcurrencyLimit(ctx); err != nil {
+		handleConcurrencyLimitError(w, r, err)
+		return
+	}
+	defer endConcurrencyLimit()
+
+	// Set read deadline for reading the initial chunk for the request body.
+	rc := http.NewResponseController(w)
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		logger.Panicf("BUG: expecting valid deadline for the context")
+	}
+	if err := rc.SetReadDeadline(deadline); err != nil {
+		logger.Panicf("BUG: cannot set read deadline: %s", err)
+	}
+
+	// Read the initial chunk for the request body.
+	bb, err := bufferRequestBody(ctx, r.Body, userName)
+	if err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return
+	}
+	r.Body = bb
+
+	// Disable the read deadline for the rest of the request body.
+	if err := rc.SetReadDeadline(time.Time{}); err != nil {
+		logger.Panicf("BUG: cannot reset read deadline: %s", err)
+	}
+
+	// Acquire concurrency limit for the given user.
+	if err := ui.beginConcurrencyLimit(ctx); err != nil {
+		handleConcurrencyLimitError(w, r, err)
+		return
+	}
+	defer ui.endConcurrencyLimit()
+
+	// Process the request.
+	processRequest(w, r, ui, tkn)
+}
+
+func beginConcurrencyLimit(ctx context.Context) error {
 	concurrencyLimitOnce.Do(concurrencyLimitInit)
 	select {
 	case concurrencyLimitCh <- struct{}{}:
-		if err := ui.beginConcurrencyLimit(ctx); err != nil {
-			handleConcurrencyLimitError(w, r, err)
-			<-concurrencyLimitCh
-			return
-		}
+		return nil
 	default:
 		// The -maxConcurrentRequests are executed. Wait until some of the requests are finished,
 		// so the current request could be executed.
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10078
 		select {
 		case concurrencyLimitCh <- struct{}{}:
-			if err := ui.beginConcurrencyLimit(ctx); err != nil {
-				handleConcurrencyLimitError(w, r, err)
-				<-concurrencyLimitCh
-				return
-			}
+			return nil
 		case <-ctx.Done():
 			err := ctx.Err()
-
-			concurrentRequestsLimitReached.Inc()
-
 			if errors.Is(err, context.DeadlineExceeded) {
-				err = fmt.Errorf("cannot start executing the request during -maxQueueDuration=%s because -maxConcurrentRequests=%d concurrent requests are executed",
+				// The current request couldn't be executed until the request timeout.
+				concurrentRequestsLimitReached.Inc()
+				return fmt.Errorf("cannot start executing the request during -maxQueueDuration=%s because -maxConcurrentRequests=%d concurrent requests are executed",
 					*maxQueueDuration, cap(concurrencyLimitCh))
-				handleConcurrencyLimitError(w, r, err)
-				return
 			}
-
-			err = fmt.Errorf("cannot start executing the request because -maxConcurrentRequests=%d concurrent requests are executed: %w", cap(concurrencyLimitCh), err)
-			handleConcurrencyLimitError(w, r, err)
-			return
+			return fmt.Errorf("cannot start executing the request because -maxConcurrentRequests=%d concurrent requests are executed: %w", cap(concurrencyLimitCh), err)
 		}
 	}
-	processRequest(w, r, ui)
-	ui.endConcurrencyLimit()
+}
+
+func endConcurrencyLimit() {
 	<-concurrencyLimitCh
 }
 
-func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
+func bufferRequestBody(ctx context.Context, r io.ReadCloser, userName string) (io.ReadCloser, error) {
+	if r == nil {
+		// This is a GET request with nil reader.
+		return nil, nil
+	}
+
+	maxBufSize := max(requestBufferSize.IntN(), maxRequestBodySizeToRetry.IntN())
+	if maxBufSize <= 0 {
+		return r, nil
+	}
+
+	lr := ioutil.GetLimitedReader(r, int64(maxBufSize))
+	defer ioutil.PutLimitedReader(lr)
+
+	start := time.Now()
+	buf, err := io.ReadAll(lr)
+	bufferRequestBodyDuration.UpdateDuration(start)
+
+	if err != nil {
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			rejectSlowClientRequests.Inc()
+
+			d := time.Since(start)
+
+			return nil, &httpserver.ErrorWithStatusCode{
+				Err: fmt.Errorf("reject request from the user %s because the request body couldn't be read in -maxQueueDuration=%s; read %d bytes in %s",
+					userName, *maxQueueDuration, len(buf), d.Truncate(time.Second)),
+				StatusCode: http.StatusBadRequest,
+			}
+		}
+
+		return nil, &httpserver.ErrorWithStatusCode{
+			Err:        fmt.Errorf("cannot read request body: %w", err),
+			StatusCode: http.StatusBadRequest,
+		}
+	}
+
+	bb := newBufferedBody(r, buf, maxBufSize)
+	return bb, nil
+}
+
+func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo, tkn *jwt.Token) {
 	u := normalizeURL(r.URL)
 	up, hc := ui.getURLPrefixAndHeaders(u, r.Host, r.Header)
 	isDefault := false
@@ -282,28 +415,31 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 		isDefault = true
 	}
 
-	rtb := newReadTrackingBody(r.Body, maxRequestBodySizeToRetry.IntN())
-	r.Body = rtb
-
 	maxAttempts := up.getBackendsCount()
-	for i := 0; i < maxAttempts; i++ {
+	for range maxAttempts {
 		bu := up.getBackendURL()
 		if bu == nil {
 			break
 		}
 		targetURL := bu.url
-		// Don't change path and add request_path query param for default route.
+		if tkn != nil {
+			// for security reasons allow templating only for configured url values and headers
+			targetURL, hc = replaceJWTPlaceholders(bu, hc, tkn.VMAccess())
+		}
 		if isDefault {
+			// Don't change path and add request_path query param for default route.
+			targetURLCopy := *targetURL
 			query := targetURL.Query()
 			query.Set("request_path", u.String())
-			targetURL.RawQuery = query.Encode()
-		} else { // Update path for regular routes.
+			targetURLCopy.RawQuery = query.Encode()
+			targetURL = &targetURLCopy
+		} else {
+			// Update path for regular routes.
 			targetURL = mergeURLs(targetURL, u, up.dropSrcPathPrefixParts, up.mergeQueryArgs)
 		}
-
 		wasLocalRetry := false
 	again:
-		ok, needLocalRetry := tryProcessingRequest(w, r, targetURL, hc, up.retryStatusCodes, ui)
+		ok, needLocalRetry := tryProcessingRequest(w, r, targetURL, hc, up.retryStatusCodes, ui, bu)
 		if needLocalRetry && !wasLocalRetry {
 			wasLocalRetry = true
 			goto again
@@ -313,18 +449,19 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo) {
 		if ok {
 			return
 		}
+
 		bu.setBroken()
 		ui.backendErrors.Inc()
 	}
 	err := &httpserver.ErrorWithStatusCode{
-		Err:        fmt.Errorf("all the %d backends for the user %q are unavailable", up.getBackendsCount(), ui.name()),
+		Err:        fmt.Errorf("all the %d backends for the user %q are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend", up.getBackendsCount(), ui.name()),
 		StatusCode: http.StatusBadGateway,
 	}
 	httpserver.Errorf(w, r, "%s", err)
 	ui.requestErrors.Inc()
 }
 
-func tryProcessingRequest(w http.ResponseWriter, r *http.Request, targetURL *url.URL, hc HeadersConf, retryStatusCodes []int, ui *UserInfo) (bool, bool) {
+func tryProcessingRequest(w http.ResponseWriter, r *http.Request, targetURL *url.URL, hc HeadersConf, retryStatusCodes []int, ui *UserInfo, bu *backendURL) (bool, bool) {
 	ui.backendRequests.Inc()
 	req := sanitizeRequestHeaders(r)
 
@@ -339,30 +476,19 @@ func tryProcessingRequest(w http.ResponseWriter, r *http.Request, targetURL *url
 		}
 	}
 
-	rtb, rtbOK := req.Body.(*readTrackingBody)
+	bb, bbOK := req.Body.(*bufferedBody)
+	canRetry := !bbOK || bb.canRetry()
+
 	res, err := ui.rt.RoundTrip(req)
 
-	if ctxErr := r.Context().Err(); ctxErr != nil {
-		// Override the error returned by the RoundTrip with the context error if it isn't non-nil
-		// This makes sure the proper logging for canceled and timed out requests - log the real cause of the error
-		// instead of the random error, which could be returned from RoundTrip because of canceled or timed out request.
-		err = ctxErr
+	if errors.Is(r.Context().Err(), context.Canceled) {
+		// Do not retry canceled requests.
+		clientCanceledRequests.Inc()
+		return true, false
 	}
+
 	if err != nil {
-		// Do not retry canceled
-		if errors.Is(err, context.Canceled) {
-			clientCanceledRequests.Inc()
-			return true, false
-		}
-		// Do not retry timed out requests
-		if errors.Is(err, context.DeadlineExceeded) {
-			remoteAddr := httpserver.GetQuotedRemoteAddr(r)
-			requestURI := httpserver.GetRequestURI(r)
-			// Timed out request must be counted as errors, since this usually means that the backend is slow.
-			logger.Warnf("remoteAddr: %s; requestURI: %s; timeout while proxying the response from %s: %s", remoteAddr, requestURI, targetURL, err)
-			return false, false
-		}
-		if !rtbOK || !rtb.canRetry() {
+		if !canRetry {
 			// Request body cannot be re-sent to another backend. Return the error to the client then.
 			err = &httpserver.ErrorWithStatusCode{
 				Err:        fmt.Errorf("cannot proxy the request to %s: %w", targetURL, err),
@@ -371,27 +497,32 @@ func tryProcessingRequest(w http.ResponseWriter, r *http.Request, targetURL *url
 			httpserver.Errorf(w, r, "%s", err)
 			ui.backendErrors.Inc()
 			ui.requestErrors.Inc()
+			bu.setBroken()
 			return true, false
 		}
 		if netutil.IsTrivialNetworkError(err) {
 			// Retry request at the same backend on trivial network errors, such as proxy idle timeout misconfiguration or socket close by OS
+			if bbOK {
+				bb.resetReader()
+			}
 			return false, true
 		}
 
-		// Request body wasn't read yet, this usually means that the backend isn't reachable; retry the request at another backend
+		// Retry the request at another backend
 		remoteAddr := httpserver.GetQuotedRemoteAddr(r)
-		// NOTE: do not use httpserver.GetRequestURI
-		// it explicitly reads request body, which may fail retries.
-		logger.Warnf("remoteAddr: %s; requestURI: %s; request to %s failed: %s, retrying the request at another backend", remoteAddr, req.URL, targetURL, err)
+		requestURI := httpserver.GetRequestURI(r)
+		logger.Warnf("remoteAddr: %s; requestURI: %s; request to %s failed: %s, retrying the request at another backend", remoteAddr, requestURI, targetURL, err)
+		if bbOK {
+			bb.resetReader()
+		}
 		return false, false
 	}
 	if slices.Contains(retryStatusCodes, res.StatusCode) {
-		_ = res.Body.Close()
-		if !rtbOK || !rtb.canRetry() {
+		if !canRetry {
 			// If we get an error from the retry_status_codes list, but cannot execute retry,
 			// we consider such a request an error as well.
 			err := &httpserver.ErrorWithStatusCode{
-				Err: fmt.Errorf("got response status code=%d from %s, but cannot retry the request at another backend, because the request has been already consumed",
+				Err: fmt.Errorf("got response status code=%d from %s, but cannot retry the request at another backend, because the request body has been already consumed",
 					res.StatusCode, targetURL),
 				StatusCode: http.StatusServiceUnavailable,
 			}
@@ -400,13 +531,16 @@ func tryProcessingRequest(w http.ResponseWriter, r *http.Request, targetURL *url
 			ui.requestErrors.Inc()
 			return true, false
 		}
+
 		// Retry requests at other backends if it matches retryStatusCodes.
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4893
 		remoteAddr := httpserver.GetQuotedRemoteAddr(r)
-		// NOTE: do not use httpserver.GetRequestURI
-		// it explicitly reads request body, which may fail retries.
+		requestURI := httpserver.GetRequestURI(r)
 		logger.Warnf("remoteAddr: %s; requestURI: %s; request to %s failed, retrying the request at another backend because response status code=%d belongs to retry_status_codes=%d",
-			remoteAddr, req.URL, targetURL, res.StatusCode, retryStatusCodes)
+			remoteAddr, requestURI, targetURL, res.StatusCode, retryStatusCodes)
+		if bbOK {
+			bb.resetReader()
+		}
 		return false, false
 	}
 	removeHopHeaders(res.Header)
@@ -416,13 +550,16 @@ func tryProcessingRequest(w http.ResponseWriter, r *http.Request, targetURL *url
 
 	err = copyStreamToClient(w, res.Body)
 	_ = res.Body.Close()
-	if errors.Is(err, context.Canceled) {
+
+	if errors.Is(r.Context().Err(), context.Canceled) {
+		// Do not retry canceled requests.
 		clientCanceledRequests.Inc()
 		return true, false
-	} else if err != nil && !netutil.IsTrivialNetworkError(err) {
+	}
+
+	if err != nil && !netutil.IsTrivialNetworkError(err) {
 		remoteAddr := httpserver.GetQuotedRemoteAddr(r)
 		requestURI := httpserver.GetRequestURI(r)
-
 		logger.Warnf("remoteAddr: %s; requestURI: %s; error when proxying response body from %s: %s", remoteAddr, requestURI, targetURL, err)
 		ui.requestErrors.Inc()
 		return true, false
@@ -553,6 +690,9 @@ var (
 	invalidAuthTokenRequests = metrics.NewCounter(`vmauth_http_request_errors_total{reason="invalid_auth_token"}`)
 	missingRouteRequests     = metrics.NewCounter(`vmauth_http_request_errors_total{reason="missing_route"}`)
 	clientCanceledRequests   = metrics.NewCounter(`vmauth_http_request_errors_total{reason="client_canceled"}`)
+	rejectSlowClientRequests = metrics.NewCounter(`vmauth_http_request_errors_total{reason="reject_slow_client"}`)
+
+	bufferRequestBodyDuration = metrics.NewSummary(`vmauth_buffer_request_body_duration_seconds`)
 )
 
 func newRoundTripper(caFileOpt, certFileOpt, keyFileOpt, serverNameOpt string, insecureSkipVerifyP *bool) (http.RoundTripper, error) {
@@ -636,8 +776,7 @@ func handleMissingAuthorizationError(w http.ResponseWriter) {
 }
 
 func handleConcurrencyLimitError(w http.ResponseWriter, r *http.Request, err error) {
-	ctx := r.Context()
-	if errors.Is(ctx.Err(), context.Canceled) {
+	if errors.Is(r.Context().Err(), context.Canceled) {
 		// Do not return any response for the request canceled by the client,
 		// since the connection to the client is already closed.
 		clientCanceledRequests.Inc()
@@ -652,123 +791,78 @@ func handleConcurrencyLimitError(w http.ResponseWriter, r *http.Request, err err
 	httpserver.Errorf(w, r, "%s", err)
 }
 
-// readTrackingBody must be obtained via getReadTrackingBody()
-type readTrackingBody struct {
-	// maxBodySize is the maximum body size to cache in buf.
+// bufferedBody serves two purposes:
+//  1. Enables request retries when the body size does not exceed maxBodySize
+//     by fully buffering the body in memory.
+//  2. Prevents slow clients from reducing effective server capacity by
+//     buffering the request body before acquiring a per-user concurrency slot.
+//
+// See bufferRequestBody for details on how bufferedBody is used.
+type bufferedBody struct {
+	// r contains reader for reading the data after buf is read.
 	//
-	// Bigger bodies cannot be retried.
-	maxBodySize int
-
-	// r contains reader for initial data reading
+	// r is nil if buf contains all the data.
 	r io.ReadCloser
 
-	// buf is a buffer for data read from r. Buf size is limited by maxBodySize.
-	// If more than maxBodySize is read from r, then cannotRetry is set to true.
+	// buf contains the initial buffer read from r.
 	buf []byte
 
-	// readBuf points to the cached data at buf, which must be read in the next call to Read().
-	readBuf []byte
+	// bufOffset is the offset at buf for already read bytes.
+	bufOffset int
 
-	// cannotRetry is set to true when more than maxBodySize bytes are read from r.
-	// In this case the read data cannot fit buf, so it cannot be re-read from buf.
+	// cannotRetry is set to true after Close() call on non-nil r.
 	cannotRetry bool
-
-	// bufComplete is set to true when buf contains complete request body read from r.
-	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{}
-	if maxBodySize < 0 {
-		maxBodySize = 0
+func newBufferedBody(r io.ReadCloser, buf []byte, maxBufSize int) *bufferedBody {
+	// Do not use sync.Pool here, since http.RoundTrip may still use request body after return.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8051
+
+	if len(buf) < maxBufSize {
+		// Read the full request body into buf.
+		r = nil
 	}
-	rtb.maxBodySize = maxBodySize
 
-	if r == nil {
-		// This is GET request without request body
-		r = (*zeroReader)(nil)
+	return &bufferedBody{
+		r:   r,
+		buf: buf,
 	}
-	rtb.r = r
-	return rtb
-}
-
-type zeroReader struct{}
-
-func (r *zeroReader) Read(_ []byte) (int, error) {
-	return 0, io.EOF
-}
-
-func (r *zeroReader) Close() error {
-	return nil
 }
 
 // Read implements io.Reader interface.
-func (rtb *readTrackingBody) Read(p []byte) (int, error) {
-	if len(rtb.readBuf) > 0 {
-		n := copy(p, rtb.readBuf)
-		rtb.readBuf = rtb.readBuf[n:]
+func (bb *bufferedBody) Read(p []byte) (int, error) {
+	if bb.cannotRetry {
+		return 0, fmt.Errorf("cannot read already closed body")
+	}
+	if bb.bufOffset < len(bb.buf) {
+		n := copy(p, bb.buf[bb.bufOffset:])
+		bb.bufOffset += n
 		return n, nil
 	}
-
-	if rtb.r == nil {
-		if rtb.bufComplete {
-			return 0, io.EOF
-		}
-		return 0, fmt.Errorf("cannot read client request body after closing client reader")
+	if bb.r == nil {
+		return 0, io.EOF
 	}
-
-	n, err := rtb.r.Read(p)
-	if rtb.cannotRetry {
-		return n, err
-	}
-
-	if len(rtb.buf)+n > rtb.maxBodySize {
-		rtb.cannotRetry = true
-		return n, err
-	}
-	rtb.buf = append(rtb.buf, p[:n]...)
-	if err == io.EOF {
-		rtb.bufComplete = true
-	}
-	return n, err
+	return bb.r.Read(p)
 }
 
-func (rtb *readTrackingBody) canRetry() bool {
-	if rtb.cannotRetry {
-		return false
-	}
-	if rtb.bufComplete {
-		return true
-	}
-	return rtb.r != nil
+func (bb *bufferedBody) canRetry() bool {
+	return bb.r == nil
 }
 
 // Close implements io.Closer interface.
-func (rtb *readTrackingBody) Close() error {
-	if !rtb.cannotRetry {
-		rtb.readBuf = rtb.buf
-	} else {
-		rtb.readBuf = nil
+func (bb *bufferedBody) Close() error {
+	bb.resetReader()
+	if bb.r != nil {
+		bb.cannotRetry = true
+		return bb.r.Close()
 	}
-
-	// Close rtb.r only if the request body is completely read or if it is too big.
-	// http.Roundtrip performs body.Close call even without any Read calls,
-	// so this hack allows us to reuse request body.
-	if rtb.bufComplete || rtb.cannotRetry {
-		if rtb.r == nil {
-			return nil
-		}
-		err := rtb.r.Close()
-		rtb.r = nil
-		return err
-	}
-
 	return nil
 }
 
+func (bb *bufferedBody) resetReader() {
+	bb.bufOffset = 0
+}
+
 func debugInfo(u *url.URL, r *http.Request) string {
 	s := &strings.Builder{}
 	fmt.Fprintf(s, " (host: %q; ", r.Host)

app/vmauth/main_timing_test.go

@@ -0,0 +1,194 @@
+package main
+
+import (
+	"crypto"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"encoding/pem"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func BenchmarkJWTRequestHandler(b *testing.B) {
+	// Generate RSA key pair for testing
+	privateKey, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		b.Fatalf("cannot generate RSA key: %s", err)
+	}
+
+	// Generate public key PEM
+	publicKeyBytes, err := x509.MarshalPKIXPublicKey(&privateKey.PublicKey)
+	if err != nil {
+		b.Fatalf("cannot marshal public key: %s", err)
+	}
+	publicKeyPEM := pem.EncodeToMemory(&pem.Block{
+		Type:  "PUBLIC KEY",
+		Bytes: publicKeyBytes,
+	})
+
+	genToken := func(t *testing.B, body map[string]any, valid bool) string {
+		t.Helper()
+
+		headerJSON, err := json.Marshal(map[string]any{
+			"alg": "RS256",
+			"typ": "JWT",
+		})
+		if err != nil {
+			t.Fatalf("cannot marshal header: %s", err)
+		}
+		headerB64 := base64.RawURLEncoding.EncodeToString(headerJSON)
+
+		bodyJSON, err := json.Marshal(body)
+		if err != nil {
+			t.Fatalf("cannot marshal body: %s", err)
+		}
+		bodyB64 := base64.RawURLEncoding.EncodeToString(bodyJSON)
+
+		payload := headerB64 + "." + bodyB64
+
+		var signatureB64 string
+		if valid {
+			// Create real RSA signature
+			hash := crypto.SHA256
+			h := hash.New()
+			h.Write([]byte(payload))
+			digest := h.Sum(nil)
+
+			signature, err := rsa.SignPKCS1v15(rand.Reader, privateKey, hash, digest)
+			if err != nil {
+				t.Fatalf("cannot sign token: %s", err)
+			}
+			signatureB64 = base64.RawURLEncoding.EncodeToString(signature)
+		} else {
+			signatureB64 = base64.RawURLEncoding.EncodeToString([]byte("invalid_signature"))
+		}
+
+		return payload + "." + signatureB64
+	}
+
+	f := func(name string, cfgStr string, r *http.Request, statusCodeExpected int) {
+		b.Helper()
+
+		b.ReportAllocs()
+		b.ResetTimer()
+		ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+			if _, err := w.Write([]byte("path: " + r.URL.Path + "\n")); err != nil {
+				panic(fmt.Errorf("cannot write response: %w", err))
+			}
+		}))
+		defer ts.Close()
+
+		cfgStr = strings.ReplaceAll(cfgStr, "{BACKEND}", ts.URL)
+
+		cfgOrigP := authConfigData.Load()
+		if _, err := reloadAuthConfigData([]byte(cfgStr)); err != nil {
+			b.Fatalf("cannot load config data: %s", err)
+		}
+		defer func() {
+			cfgOrig := []byte("unauthorized_user:\n  url_prefix: http://foo/bar")
+			if cfgOrigP != nil {
+				cfgOrig = *cfgOrigP
+			}
+			_, err := reloadAuthConfigData(cfgOrig)
+			if err != nil {
+				b.Fatalf("cannot load the original config: %s", err)
+			}
+		}()
+
+		b.Run(name, func(b *testing.B) {
+			b.ResetTimer()
+			b.ReportAllocs()
+			b.RunParallel(func(pb *testing.PB) {
+				w := &fakeResponseWriter{}
+				for pb.Next() {
+					w.reset()
+					if !requestHandlerWithInternalRoutes(w, r) {
+						b.Fatalf("unexpected false is returned from requestHandler")
+					}
+					if w.statusCode != statusCodeExpected {
+						b.Fatalf("unexpected response code (-%d;+%d)", statusCodeExpected, w.statusCode)
+					}
+
+				}
+			})
+		})
+	}
+
+	simpleCfgStr := fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+  url_prefix: {BACKEND}/foo`, string(publicKeyPEM))
+	noVMAccessClaimToken := genToken(b, nil, true)
+	expiredToken := genToken(b, map[string]any{
+		"exp":       10,
+		"vm_access": map[string]any{},
+	}, true)
+
+	fullToken := genToken(b, map[string]any{
+		"exp":   time.Now().Add(10 * time.Minute).Unix(),
+		"scope": "email id",
+		"vm_access": map[string]any{
+			"extra_labels": map[string]string{
+				"label":  "value1",
+				"label2": "value3",
+			},
+			"extra_filters":      []string{"stream_filter1", "stream_filter2"},
+			"metrics_account_id": 123,
+			"metrics_project_id": 234,
+			"metrics_extra_labels": []string{
+				"label1=value1",
+				"label2=value2",
+			},
+			"metrics_extra_filters": []string{
+				`{label3="value3"}`,
+				`{label4="value4"}`,
+			},
+			"logs_account_id": 345,
+			"logs_project_id": 456,
+			"logs_extra_filters": []string{
+				`{"namespace":"my-app","env":"prod"}`,
+			},
+			"logs_extra_stream_filters": []string{
+				`{"team":"dev"}`,
+			},
+		},
+	}, true)
+
+	// tenant headers are overwritten if set as placeholders
+	// extra_filters extra_stream_filters from vm_access claim merged with statically defined
+	request := httptest.NewRequest(`GET`, "http://some-host.com/query", nil)
+	request.Header.Set(`Authorization`, `Bearer `+fullToken)
+	f("full_template",
+		fmt.Sprintf(`
+users:
+- jwt:
+    public_keys:
+    - %q
+  headers:
+    - "AccountID: {{.LogsAccountID}}"
+    - "ProjectID: {{.LogsProjectID}}"
+  url_prefix: {BACKEND}/select/logsql/?extra_filters=aStaticFilter&extra_stream_filters=aStaticStreamFilter&extra_filters={{.LogsExtraFilters}}&extra_stream_filters={{.LogsExtraStreamFilters}}`, string(publicKeyPEM)),
+		request,
+		http.StatusOK,
+	)
+
+	// token without vm_access claim
+	request = httptest.NewRequest(`GET`, "http://some-host.com/abc", nil)
+	request.Header.Set(`Authorization`, `Bearer `+noVMAccessClaimToken)
+	f("token_without_claim", simpleCfgStr, request, http.StatusUnauthorized)
+
+	// expired token
+	request = httptest.NewRequest(`GET`, "http://some-host.com/abc", nil)
+	request.Header.Set(`Authorization`, `Bearer `+expiredToken)
+	f("expired_token", simpleCfgStr, request, http.StatusUnauthorized)
+}

app/vmauth/oidc.go

@@ -0,0 +1,195 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/jwt"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+)
+
+type oidcConfig struct {
+	Issuer string `yaml:"issuer"`
+}
+
+type oidcDiscovererPool struct {
+	ds map[string]*oidcDiscoverer
+
+	context context.Context
+	cancel  func()
+	wg      *sync.WaitGroup
+}
+
+func (dp *oidcDiscovererPool) createOrAdd(issuer string, vp *atomic.Pointer[jwt.VerifierPool]) {
+	if dp.ds == nil {
+		dp.ds = make(map[string]*oidcDiscoverer)
+		dp.context, dp.cancel = context.WithCancel(context.Background())
+		dp.wg = &sync.WaitGroup{}
+	}
+
+	ds, found := dp.ds[issuer]
+	if !found {
+		ds = &oidcDiscoverer{
+			issuer: issuer,
+		}
+		dp.ds[issuer] = ds
+	}
+
+	ds.vps = append(ds.vps, vp)
+}
+
+func (dp *oidcDiscovererPool) startDiscovery() {
+	if len(dp.ds) == 0 {
+		return
+	}
+
+	for _, d := range dp.ds {
+		dp.wg.Go(func() {
+			if err := d.refreshVerifierPools(dp.context); err != nil {
+				logger.Errorf("failed to initialize OIDC verifier pool at start for issuer %q: %s", d.issuer, err)
+			}
+		})
+	}
+	dp.wg.Wait()
+
+	for _, d := range dp.ds {
+		dp.wg.Go(func() {
+			d.run(dp.context)
+		})
+	}
+}
+
+func (dp *oidcDiscovererPool) stopDiscovery() {
+	if len(dp.ds) == 0 {
+		return
+	}
+
+	dp.cancel()
+	dp.wg.Wait()
+}
+
+type oidcDiscoverer struct {
+	issuer string
+	vps    []*atomic.Pointer[jwt.VerifierPool]
+}
+
+func (d *oidcDiscoverer) run(ctx context.Context) {
+	t := time.NewTimer(timeutil.AddJitterToDuration(time.Second * 10))
+	defer t.Stop()
+
+	for {
+		select {
+		case <-t.C:
+			if err := d.refreshVerifierPools(ctx); errors.Is(err, context.Canceled) {
+				return
+			} else if err != nil {
+				t.Reset(timeutil.AddJitterToDuration(time.Second * 10))
+				logger.Errorf("failed to refresh OIDC verifier pool for issuer %q: %v", d.issuer, err)
+				continue
+			}
+			// OIDC may return Cache-Control header with max-age directive.
+			// It could be used as time range for next refresh.
+			// https://openid.net/specs/openid-connect-core-1_0.html#RotateEncKeys
+			t.Reset(timeutil.AddJitterToDuration(time.Minute * 5))
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (d *oidcDiscoverer) refreshVerifierPools(ctx context.Context) error {
+	cfg, err := getOpenIDConfiguration(ctx, d.issuer)
+	if err != nil {
+		return err
+	}
+	// The issuer in the OIDC configuration must match the expected issuer.
+	// https://openid.net/specs/openid-connect-core-1_0.html#RotateEncKeys
+	if cfg.Issuer != d.issuer {
+		return fmt.Errorf("openid configuration issuer %q does not match expected issuer %q", cfg.Issuer, d.issuer)
+	}
+
+	verifierPool, err := fetchAndParseJWKs(ctx, cfg.JWKsURI)
+	if err != nil {
+		return err
+	}
+
+	for _, vp := range d.vps {
+		vp.Store(verifierPool)
+	}
+	return nil
+}
+
+// See https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata for details.
+type openidConfig struct {
+	Issuer  string `json:"issuer"`
+	JWKsURI string `json:"jwks_uri"`
+}
+
+var oidcHTTPClient = &http.Client{
+	Timeout: time.Second * 5,
+}
+
+func fetchAndParseJWKs(ctx context.Context, jwksURI string) (*jwt.VerifierPool, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, jwksURI, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request for fetching jwks keys from %q: %w", jwksURI, err)
+	}
+
+	resp, err := oidcHTTPClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch jwks keys from %q: %w", jwksURI, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status code %d when fetching jwks keys from %q", resp.StatusCode, jwksURI)
+	}
+
+	b, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read response body from %q: %w", jwksURI, err)
+	}
+
+	vp, err := jwt.ParseJWKs(b)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse jwks keys from %q: %v", jwksURI, err)
+	}
+
+	return vp, nil
+}
+
+func getOpenIDConfiguration(ctx context.Context, issuer string) (openidConfig, error) {
+	issuer, _ = strings.CutSuffix(issuer, "/")
+	configURL := fmt.Sprintf("%s/.well-known/openid-configuration", issuer)
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, configURL, nil)
+	if err != nil {
+		return openidConfig{}, fmt.Errorf("failed to create request for fetching openid config from %q: %w", configURL, err)
+	}
+
+	resp, err := oidcHTTPClient.Do(req)
+	if err != nil {
+		return openidConfig{}, fmt.Errorf("failed to fetch openid config from %q: %w", configURL, err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return openidConfig{}, fmt.Errorf("unexpected status code %d when fetching openid config from %q", resp.StatusCode, configURL)
+	}
+
+	var cfg openidConfig
+	if err := json.NewDecoder(resp.Body).Decode(&cfg); err != nil {
+		return openidConfig{}, fmt.Errorf("failed to decode openid config from %q: %s", configURL, err)
+	}
+
+	return cfg, nil
+}

app/vmauth/target_url_test.go

@@ -174,7 +174,7 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 		},
 		RetryStatusCodes:       []int{503, 501},
 		LoadBalancingPolicy:    "first_available",
-		DropSrcPathPrefixParts: intp(2),
+		DropSrcPathPrefixParts: new(2),
 	}, "/a/b/c", "http://foo.bar/c", `bb: aaa`, `x: y`, []int{503, 501}, "first_available", 2)
 	f(&UserInfo{
 		URLPrefix: mustParseURL("http://foo.bar/federate"),
@@ -219,13 +219,13 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 				},
 				RetryStatusCodes:       []int{503, 500, 501},
 				LoadBalancingPolicy:    "first_available",
-				DropSrcPathPrefixParts: intp(1),
+				DropSrcPathPrefixParts: new(1),
 			},
 			{
 				SrcPaths:               getRegexs([]string{"/api/v1/write"}),
 				URLPrefix:              mustParseURL("http://vminsert/0/prometheus"),
 				RetryStatusCodes:       []int{},
-				DropSrcPathPrefixParts: intp(0),
+				DropSrcPathPrefixParts: new(0),
 			},
 			{
 				SrcPaths:  getRegexs([]string{"/metrics"}),
@@ -242,7 +242,7 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 			},
 		},
 		RetryStatusCodes:       []int{502},
-		DropSrcPathPrefixParts: intp(2),
+		DropSrcPathPrefixParts: new(2),
 	}
 	f(ui, "http://host42/vmsingle/api/v1/query?query=up&db=foo", "http://vmselect/0/prometheus/api/v1/query?db=foo&query=up",
 		"xx: aa\nyy: asdf", "qwe: rty", []int{503, 500, 501}, "first_available", 1)
@@ -259,7 +259,7 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 				SrcPaths:               getRegexs([]string{"/api/v1/write"}),
 				URLPrefix:              mustParseURL("http://vminsert/0/prometheus"),
 				RetryStatusCodes:       []int{},
-				DropSrcPathPrefixParts: intp(0),
+				DropSrcPathPrefixParts: new(0),
 			},
 			{
 				SrcPaths:  getRegexs([]string{"/metrics/a/b"}),
@@ -275,7 +275,7 @@ func TestCreateTargetURLSuccess(t *testing.T) {
 			},
 		},
 		RetryStatusCodes:       []int{502},
-		DropSrcPathPrefixParts: intp(2),
+		DropSrcPathPrefixParts: new(2),
 	}
 	f(ui, "https://foo-host/api/v1/write", "http://vminsert/0/prometheus/api/v1/write", "", "", []int{}, "least_loaded", 0)
 	f(ui, "https://foo-host/metrics/a/b", "http://metrics-server/b", "", "", []int{502}, "least_loaded", 2)

app/vmctl/backoff/backoff.go

@@ -7,6 +7,8 @@ import (
 	"math"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 )
 
@@ -45,7 +47,7 @@ func New(retries int, factor float64, minDuration time.Duration) (*Backoff, erro
 // Retry process retries until all attempts are completed
 func (b *Backoff) Retry(ctx context.Context, cb retryableFunc) (uint64, error) {
 	var attempt uint64
-	for i := 0; i < b.retries; i++ {
+	for i := range b.retries {
 		err := cb()
 		if err == nil {
 			return attempt, nil
@@ -55,6 +57,7 @@ func (b *Backoff) Retry(ctx context.Context, cb retryableFunc) (uint64, error) {
 			return attempt, err // fail fast if not recoverable
 		}
 		attempt++
+		retriesTotal.Inc()
 		backoff := float64(b.minDuration) * math.Pow(b.factor, float64(i))
 		dur := time.Duration(backoff)
 		logger.Errorf("got error: %s on attempt: %d; will retry in %v", err, attempt, dur)
@@ -74,3 +77,7 @@ func (b *Backoff) Retry(ctx context.Context, cb retryableFunc) (uint64, error) {
 	}
 	return attempt, fmt.Errorf("execution failed after %d retry attempts", b.retries)
 }
+
+var (
+	retriesTotal = metrics.NewCounter(`vmctl_backoff_retries_total`)
+)

app/vmctl/flags.go

@@ -14,6 +14,12 @@ const (
 	globalSilent             = "s"
 	globalVerbose            = "verbose"
 	globalDisableProgressBar = "disable-progress-bar"
+
+	globalPushMetricsURL         = "pushmetrics.url"
+	globalPushMetricsInterval    = "pushmetrics.interval"
+	globalPushExtraLabels        = "pushmetrics.extraLabel"
+	globalPushHeaders            = "pushmetrics.header"
+	globalPushDisableCompression = "pushmetrics.disableCompression"
 )
 
 var (
@@ -33,6 +39,29 @@ var (
 			Value: false,
 			Usage: "Whether to disable progress bar during the import.",
 		},
+		&cli.StringSliceFlag{
+			Name:  globalPushMetricsURL,
+			Usage: "Optional URL to push metrics. See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#push-metrics",
+		},
+		&cli.DurationFlag{
+			Name:  globalPushMetricsInterval,
+			Value: 10 * time.Second,
+			Usage: "Interval for pushing metrics to every -pushmetrics.url",
+		},
+		&cli.StringSliceFlag{
+			Name: globalPushExtraLabels,
+			Usage: "Extra labels to add to pushed metrics. In case of collision, label value defined by flag will have priority. " +
+				"Flag can be set multiple times, to add few additional labels. " +
+				"For example, -pushmetrics.extraLabel='instance=\"foo\"' adds instance=\"foo\" label to all the metrics pushed to every -pushmetrics.url",
+		},
+		&cli.StringSliceFlag{
+			Name:  globalPushHeaders,
+			Usage: "Optional HTTP headers to add to pushed metrics. Flag can be set multiple times, to add few additional headers.",
+		},
+		&cli.BoolFlag{
+			Name:  globalPushDisableCompression,
+			Usage: "Whether to disable compression when pushing metrics.",
+		},
 	}
 )
 
@@ -123,32 +152,32 @@ var (
 			Name:  vmExtraLabel,
 			Value: nil,
 			Usage: "Extra labels, that will be added to imported timeseries. In case of collision, label value defined by flag" +
-				"will have priority. Flag can be set multiple times, to add few additional labels.",
+				" will have priority. Flag can be set multiple times, to add few additional labels.",
 		},
 		&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 configured via '--vmAddr' destination.",
+				"By default, the rate limit is disabled. It can be useful for limiting load on configured via '--vm-addr' destination.",
 		},
 		&cli.StringFlag{
 			Name:  vmCertFile,
-			Usage: "Optional path to client-side TLS certificate file to use when connecting to '--vmAddr'",
+			Usage: "Optional path to client-side TLS certificate file to use when connecting to '--vm-addr'",
 		},
 		&cli.StringFlag{
 			Name:  vmKeyFile,
-			Usage: "Optional path to client-side TLS key to use when connecting to '--vmAddr'",
+			Usage: "Optional path to client-side TLS key to use when connecting to '--vm-addr'",
 		},
 		&cli.StringFlag{
 			Name:  vmCAFile,
-			Usage: "Optional path to TLS CA file to use for verifying connections to '--vmAddr'. By default, system CA is used",
+			Usage: "Optional path to TLS CA file to use for verifying connections to '--vm-addr'. By default, system CA is used",
 		},
 		&cli.StringFlag{
 			Name:  vmServerName,
-			Usage: "Optional TLS server name to use for connections to '--vmAddr'. By default, the server name from '--vmAddr' is used",
+			Usage: "Optional TLS server name to use for connections to '--vm-addr'. By default, the server name from '--vm-addr' is used",
 		},
 		&cli.BoolFlag{
 			Name:  vmInsecureSkipVerify,
-			Usage: "Whether to skip tls verification when connecting to '--vmAddr'",
+			Usage: "Whether to skip tls verification when connecting to '--vm-addr'",
 			Value: false,
 		},
 		&cli.IntFlag{
@@ -598,7 +627,7 @@ var (
 			Name:  vmExtraLabel,
 			Value: nil,
 			Usage: "Extra labels, that will be added to imported timeseries. In case of collision, label value defined by flag" +
-				"will have priority. Flag can be set multiple times, to add few additional labels.",
+				" will have priority. Flag can be set multiple times, to add few additional labels.",
 		},
 		&cli.Int64Flag{
 			Name: vmRateLimit,
@@ -625,8 +654,8 @@ var (
 		&cli.BoolFlag{
 			Name: vmNativeDisableBinaryProtocol,
 			Usage: "Whether to use https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-export-data-in-json-line-format " +
-				"instead of https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-export-data-in-native-format API." +
-				"Binary export/import API protocol implies less network and resource usage, as it transfers compressed binary data blocks." +
+				"instead of https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#how-to-export-data-in-native-format API. " +
+				"Binary export/import API protocol implies less network and resource usage, as it transfers compressed binary data blocks. " +
 				"Non-binary export/import API is less efficient, but supports deduplication if it is configured on vm-native-src-addr side.",
 			Value: false,
 		},

app/vmctl/influx.go

@@ -7,6 +7,8 @@ import (
 	"log"
 	"sync"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
@@ -52,6 +54,7 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 		return nil
 	}
 
+	influxSeriesTotal.Add(len(series))
 	bar := barpool.AddWithTemplate(fmt.Sprintf(barTpl, "Processing series"), len(series))
 	if err := barpool.Start(); err != nil {
 		return err
@@ -63,18 +66,18 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 	ip.im.ResetStats()
 
 	var wg sync.WaitGroup
-	wg.Add(ip.cc)
-	for i := 0; i < ip.cc; i++ {
-		go func() {
-			defer wg.Done()
+	for range ip.cc {
+		wg.Go(func() {
 			for s := range seriesCh {
 				if err := ip.do(s); err != nil {
+					influxErrorsTotal.Inc()
 					errCh <- fmt.Errorf("request failed for %q.%q: %s", s.Measurement, s.Field, err)
 					return
 				}
+				influxSeriesProcessed.Inc()
 				bar.Increment()
 			}
-		}()
+		})
 	}
 
 	// any error breaks the import
@@ -83,6 +86,7 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 		case infErr := <-errCh:
 			return fmt.Errorf("influx error: %s", infErr)
 		case vmErr := <-ip.im.Errors():
+			influxErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, ip.isVerbose))
 		case seriesCh <- s:
 		}
@@ -95,6 +99,7 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 	// drain import errors channel
 	for vmErr := range ip.im.Errors() {
 		if vmErr.Err != nil {
+			influxErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, ip.isVerbose))
 		}
 	}
@@ -169,3 +174,9 @@ func (ip *influxProcessor) do(s *influx.Series) error {
 		}
 	}
 }
+
+var (
+	influxSeriesTotal     = metrics.NewCounter(`vmctl_influx_migration_series_total`)
+	influxSeriesProcessed = metrics.NewCounter(`vmctl_influx_migration_series_processed`)
+	influxErrorsTotal     = metrics.NewCounter(`vmctl_influx_migration_errors_total`)
+)

app/vmctl/limiter/limiter.go

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 )
 
@@ -45,9 +47,16 @@ func (l *Limiter) Register(dataLen int) {
 			t := timerpool.Get(d)
 			<-t.C
 			timerpool.Put(t)
+			limiterThrottleEventsTotal.Inc()
 		}
 		l.budget += limit
 		l.deadline = time.Now().Add(time.Second)
 	}
 	l.budget -= int64(dataLen)
+	limiterBytesProcessed.Add(dataLen)
 }
+
+var (
+	limiterBytesProcessed      = metrics.NewCounter(`vmctl_limiter_bytes_processed_total`)
+	limiterThrottleEventsTotal = metrics.NewCounter(`vmctl_limiter_throttle_events_total`)
+)

app/vmctl/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"flag"
 	"fmt"
 	"log"
 	"net/http"
@@ -19,7 +20,9 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/native"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/remoteread"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/opentsdb"
@@ -41,11 +44,20 @@ func main() {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	start := time.Now()
 	beforeFn := func(c *cli.Context) error {
+		flag.Parse()
+		logger.Init()
 		isSilent = c.Bool(globalSilent)
 		if c.Bool(globalDisableProgressBar) {
 			barpool.Disable(true)
 		}
 		netutil.EnableIPv6()
+		pushmetrics.InitWith(&pushmetrics.Config{
+			URLs:               c.StringSlice(globalPushMetricsURL),
+			Interval:           c.Duration(globalPushMetricsInterval),
+			ExtraLabels:        c.StringSlice(globalPushExtraLabels),
+			DisableCompression: c.Bool(globalPushDisableCompression),
+			Headers:            c.StringSlice(globalPushHeaders),
+		})
 		return nil
 	}
 	app := &cli.App{
@@ -451,6 +463,7 @@ func main() {
 		log.Fatalln(err)
 	}
 	log.Printf("Total time: %v", time.Since(start))
+	pushmetrics.StopAndPush()
 }
 
 func initConfigVM(c *cli.Context) (vm.Config, error) {

app/vmctl/native/client.go

@@ -8,6 +8,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/auth"
 )
 
@@ -36,12 +38,15 @@ type Response struct {
 
 // Explore finds metric names by provided filter from api/v1/label/__name__/values
 func (c *Client) Explore(ctx context.Context, f Filter, tenantID string, start, end time.Time) ([]string, error) {
+	startTime := time.Now()
+	exploreRequestsTotal.Inc()
 	url := fmt.Sprintf("%s/%s", c.Addr, nativeMetricNamesAddr)
 	if tenantID != "" {
 		url = fmt.Sprintf("%s/select/%s/prometheus/%s", c.Addr, tenantID, nativeMetricNamesAddr)
 	}
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
+		exploreRequestsErrorsTotal.Inc()
 		return nil, fmt.Errorf("cannot create request to %q: %s", url, err)
 	}
 
@@ -53,37 +58,53 @@ func (c *Client) Explore(ctx context.Context, f Filter, tenantID string, start,
 
 	resp, err := c.do(req, http.StatusOK)
 	if err != nil {
+		exploreRequestsErrorsTotal.Inc()
+		exploreDuration.UpdateDuration(startTime)
 		return nil, fmt.Errorf("series request failed: %s", err)
 	}
 
 	var response Response
 	if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
+		exploreRequestsErrorsTotal.Inc()
+		exploreDuration.UpdateDuration(startTime)
 		return nil, fmt.Errorf("cannot decode series response: %s", err)
 	}
+	exploreDuration.UpdateDuration(startTime)
 	return response.MetricNames, resp.Body.Close()
 }
 
 // ImportPipe uses pipe reader in request to process data
 func (c *Client) ImportPipe(ctx context.Context, dstURL string, pr *io.PipeReader) error {
+	startTime := time.Now()
+	importRequestsTotal.Inc()
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, dstURL, pr)
 	if err != nil {
+		importRequestsErrorsTotal.Inc()
 		return fmt.Errorf("cannot create import request to %q: %s", c.Addr, err)
 	}
 
 	importResp, err := c.do(req, http.StatusNoContent)
 	if err != nil {
+		importRequestsErrorsTotal.Inc()
+		importDuration.UpdateDuration(startTime)
 		return fmt.Errorf("import request failed: %s", err)
 	}
 	if err := importResp.Body.Close(); err != nil {
+		importRequestsErrorsTotal.Inc()
+		importDuration.UpdateDuration(startTime)
 		return fmt.Errorf("cannot close import response body: %s", err)
 	}
+	importDuration.UpdateDuration(startTime)
 	return nil
 }
 
 // ExportPipe makes request by provided filter and return io.ReadCloser which can be used to get data
 func (c *Client) ExportPipe(ctx context.Context, url string, f Filter) (io.ReadCloser, error) {
+	startTime := time.Now()
+	exportRequestsTotal.Inc()
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
+		exportRequestsErrorsTotal.Inc()
 		return nil, fmt.Errorf("cannot create request to %q: %s", c.Addr, err)
 	}
 
@@ -102,8 +123,11 @@ func (c *Client) ExportPipe(ctx context.Context, url string, f Filter) (io.ReadC
 
 	resp, err := c.do(req, http.StatusOK)
 	if err != nil {
+		exportRequestsErrorsTotal.Inc()
+		exportDuration.UpdateDuration(startTime)
 		return nil, fmt.Errorf("export request failed: %w", err)
 	}
+	exportDuration.UpdateDuration(startTime)
 	return resp.Body, nil
 }
 
@@ -162,3 +186,16 @@ func (c *Client) do(req *http.Request, expSC int) (*http.Response, error) {
 	}
 	return resp, err
 }
+
+var (
+	importRequestsTotal        = metrics.NewCounter(`vmctl_vm_native_requests_total{type="import"}`)
+	exportRequestsTotal        = metrics.NewCounter(`vmctl_vm_native_requests_total{type="export"}`)
+	exploreRequestsTotal       = metrics.NewCounter(`vmctl_vm_native_requests_total{type="explore"}`)
+	importRequestsErrorsTotal  = metrics.NewCounter(`vmctl_vm_native_request_errors_total{type="import"}`)
+	exportRequestsErrorsTotal  = metrics.NewCounter(`vmctl_vm_native_request_errors_total{type="export"}`)
+	exploreRequestsErrorsTotal = metrics.NewCounter(`vmctl_vm_native_request_errors_total{type="explore"}`)
+
+	importDuration  = metrics.NewHistogram(`vmctl_vm_native_import_duration_seconds`)
+	exportDuration  = metrics.NewHistogram(`vmctl_vm_native_export_duration_seconds`)
+	exploreDuration = metrics.NewHistogram(`vmctl_vm_native_explore_duration_seconds`)
+)

app/vmctl/opentsdb.go

@@ -7,6 +7,8 @@ import (
 	"sync"
 	"time"
 
+	vmetrics "github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/opentsdb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
 	"github.com/cheggaaa/pb/v3"
@@ -57,6 +59,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 	if !prompt(ctx, question) {
 		return nil
 	}
+
 	op.im.ResetStats()
 	var startTime int64
 	if op.oc.HardTS != 0 {
@@ -84,23 +87,24 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 		seriesCh := make(chan queryObj, op.otsdbcc)
 		errCh := make(chan error)
 		// we're going to make serieslist * queryRanges queries, so we should represent that in the progress bar
+		otsdbSeriesTotal.Add(len(serieslist) * queryRanges)
 		bar := pb.StartNew(len(serieslist) * queryRanges)
 		defer func(bar *pb.ProgressBar) {
 			bar.Finish()
 		}(bar)
 		var wg sync.WaitGroup
-		wg.Add(op.otsdbcc)
-		for i := 0; i < op.otsdbcc; i++ {
-			go func() {
-				defer wg.Done()
+		for range op.otsdbcc {
+			wg.Go(func() {
 				for s := range seriesCh {
 					if err := op.do(s); err != nil {
+						otsdbErrorsTotal.Inc()
 						errCh <- fmt.Errorf("couldn't retrieve series for %s : %s", metric, err)
 						return
 					}
+					otsdbSeriesProcessed.Inc()
 					bar.Increment()
 				}
-			}()
+			})
 		}
 		/*
 			Loop through all series for this metric, processing all retentions and time ranges
@@ -117,6 +121,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 					case otsdbErr := <-errCh:
 						return fmt.Errorf("opentsdb error: %s", otsdbErr)
 					case vmErr := <-op.im.Errors():
+						otsdbErrorsTotal.Inc()
 						return fmt.Errorf("import process failed: %s", wrapErr(vmErr, op.isVerbose))
 					case seriesCh <- queryObj{
 						Tr: tr, StartTime: startTime,
@@ -141,6 +146,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 	op.im.Close()
 	for vmErr := range op.im.Errors() {
 		if vmErr.Err != nil {
+			otsdbErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, op.isVerbose))
 		}
 	}
@@ -171,3 +177,9 @@ func (op *otsdbProcessor) do(s queryObj) error {
 	}
 	return op.im.Input(&ts)
 }
+
+var (
+	otsdbSeriesTotal     = vmetrics.NewCounter(`vmctl_opentsdb_migration_series_total`)
+	otsdbSeriesProcessed = vmetrics.NewCounter(`vmctl_opentsdb_migration_series_processed`)
+	otsdbErrorsTotal     = vmetrics.NewCounter(`vmctl_opentsdb_migration_errors_total`)
+)

app/vmctl/opentsdb/opentsdb.go

@@ -109,7 +109,7 @@ func (c Client) FindMetrics(q string) ([]string, error) {
 		return nil, fmt.Errorf("failed to send GET request to %q: %s", q, err)
 	}
 	if resp.StatusCode != 200 {
-		return nil, fmt.Errorf("bad return from OpenTSDB: %q: %v", resp.StatusCode, resp)
+		return nil, fmt.Errorf("bad return from OpenTSDB: %d: %v", resp.StatusCode, resp)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	body, err := io.ReadAll(resp.Body)
@@ -133,7 +133,7 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
 		return nil, fmt.Errorf("failed to set GET request to %q: %s", q, err)
 	}
 	if resp.StatusCode != 200 {
-		return nil, fmt.Errorf("bad return from OpenTSDB: %q: %v", resp.StatusCode, resp)
+		return nil, fmt.Errorf("bad return from OpenTSDB: %d: %v", resp.StatusCode, resp)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	body, err := io.ReadAll(resp.Body)

app/vmctl/prometheus.go

@@ -11,6 +11,8 @@ import (
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/prometheus/prometheus/tsdb/chunkenc"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/prometheus"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
@@ -113,6 +115,7 @@ func (pp *prometheusProcessor) do(b tsdb.BlockReader) error {
 }
 
 func (pp *prometheusProcessor) processBlocks(blocks []tsdb.BlockReader) error {
+	promBlocksTotal.Add(len(blocks))
 	bar := barpool.AddWithTemplate(fmt.Sprintf(barTpl, "Processing blocks"), len(blocks))
 	if err := barpool.Start(); err != nil {
 		return err
@@ -124,18 +127,18 @@ func (pp *prometheusProcessor) processBlocks(blocks []tsdb.BlockReader) error {
 	pp.im.ResetStats()
 
 	var wg sync.WaitGroup
-	wg.Add(pp.cc)
-	for i := 0; i < pp.cc; i++ {
-		go func() {
-			defer wg.Done()
+	for range pp.cc {
+		wg.Go(func() {
 			for br := range blockReadersCh {
 				if err := pp.do(br); err != nil {
+					promErrorsTotal.Inc()
 					errCh <- fmt.Errorf("read failed for block %q: %s", br.Meta().ULID, err)
 					return
 				}
+				promBlocksProcessed.Inc()
 				bar.Increment()
 			}
-		}()
+		})
 	}
 	// any error breaks the import
 	for _, br := range blocks {
@@ -145,6 +148,7 @@ func (pp *prometheusProcessor) processBlocks(blocks []tsdb.BlockReader) error {
 			return fmt.Errorf("prometheus error: %s", promErr)
 		case vmErr := <-pp.im.Errors():
 			close(blockReadersCh)
+			promErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
 		case blockReadersCh <- br:
 		}
@@ -158,6 +162,7 @@ func (pp *prometheusProcessor) processBlocks(blocks []tsdb.BlockReader) error {
 	// drain import errors channel
 	for vmErr := range pp.im.Errors() {
 		if vmErr.Err != nil {
+			promErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
 		}
 	}
@@ -167,3 +172,9 @@ func (pp *prometheusProcessor) processBlocks(blocks []tsdb.BlockReader) error {
 
 	return nil
 }
+
+var (
+	promBlocksTotal     = metrics.NewCounter(`vmctl_prometheus_migration_blocks_total`)
+	promBlocksProcessed = metrics.NewCounter(`vmctl_prometheus_migration_blocks_processed`)
+	promErrorsTotal     = metrics.NewCounter(`vmctl_prometheus_migration_errors_total`)
+)

app/vmctl/remoteread.go

@@ -7,6 +7,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/remoteread"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
@@ -51,6 +53,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 		return nil
 	}
 
+	remoteReadRangesTotal.Add(len(ranges))
 	bar := barpool.AddWithTemplate(fmt.Sprintf(barTpl, "Processing ranges"), len(ranges))
 	if err := barpool.Start(); err != nil {
 		return err
@@ -66,18 +69,18 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 	errCh := make(chan error)
 
 	var wg sync.WaitGroup
-	wg.Add(rrp.cc)
-	for i := 0; i < rrp.cc; i++ {
-		go func() {
-			defer wg.Done()
+	for range rrp.cc {
+		wg.Go(func() {
 			for r := range rangeC {
 				if err := rrp.do(ctx, r); err != nil {
+					remoteReadErrorsTotal.Inc()
 					errCh <- fmt.Errorf("request failed for: %s", err)
 					return
 				}
+				remoteReadRangesProcessed.Inc()
 				bar.Increment()
 			}
-		}()
+		})
 	}
 
 	for _, r := range ranges {
@@ -85,6 +88,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 		case infErr := <-errCh:
 			return fmt.Errorf("remote read error: %s", infErr)
 		case vmErr := <-rrp.dst.Errors():
+			remoteReadErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
 		case rangeC <- &remoteread.Filter{
 			StartTimestampMs: r[0].UnixMilli(),
@@ -100,6 +104,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 	// drain import errors channel
 	for vmErr := range rrp.dst.Errors() {
 		if vmErr.Err != nil {
+			remoteReadErrorsTotal.Inc()
 			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
 		}
 	}
@@ -120,3 +125,9 @@ func (rrp *remoteReadProcessor) do(ctx context.Context, filter *remoteread.Filte
 		return nil
 	})
 }
+
+var (
+	remoteReadRangesTotal     = metrics.NewCounter(`vmctl_remote_read_migration_ranges_total`)
+	remoteReadRangesProcessed = metrics.NewCounter(`vmctl_remote_read_migration_ranges_processed`)
+	remoteReadErrorsTotal     = metrics.NewCounter(`vmctl_remote_read_migration_errors_total`)
+)

app/vmctl/vm/timeseries.go

@@ -76,11 +76,11 @@ func (ts *TimeSeries) write(w io.Writer) (int, error) {
 
 		pointsCount := len(timestampsBatch)
 		cw.printf(`},"timestamps":[`)
-		for i := 0; i < pointsCount-1; i++ {
+		for i := range pointsCount - 1 {
 			cw.printf(`%d,`, timestampsBatch[i])
 		}
 		cw.printf(`%d],"values":[`, timestampsBatch[pointsCount-1])
-		for i := 0; i < pointsCount-1; i++ {
+		for i := range pointsCount - 1 {
 			cw.printf(`%v,`, valuesBatch[i])
 		}
 		cw.printf("%v]}\n", valuesBatch[pointsCount-1])

app/vmctl/vm/vm.go

@@ -12,6 +12,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/backoff"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/limiter"
@@ -80,6 +82,12 @@ type Importer struct {
 
 	s       *stats
 	backoff *backoff.Backoff
+
+	importRequestsTotal       *metrics.Counter
+	importRequestsErrorsTotal *metrics.Counter
+	importSamplesTotal        *metrics.Counter
+	importBytesTotal          *metrics.Counter
+	importDuration            *metrics.Histogram
 }
 
 // ResetStats resets im stats.
@@ -147,6 +155,12 @@ func NewImporter(ctx context.Context, cfg Config) (*Importer, error) {
 		input:      make(chan *TimeSeries, cfg.Concurrency*4),
 		errors:     make(chan *ImportError, cfg.Concurrency),
 		backoff:    cfg.Backoff,
+
+		importRequestsTotal:       metrics.GetOrCreateCounter(`vmctl_importer_requests_total`),
+		importRequestsErrorsTotal: metrics.GetOrCreateCounter(`vmctl_importer_request_errors_total`),
+		importSamplesTotal:        metrics.GetOrCreateCounter(`vmctl_importer_samples_total`),
+		importBytesTotal:          metrics.GetOrCreateCounter(`vmctl_importer_bytes_total`),
+		importDuration:            metrics.GetOrCreateHistogram(`vmctl_importer_request_duration_seconds`),
 	}
 	if err := im.Ping(); err != nil {
 		return nil, fmt.Errorf("ping to %q failed: %s", addr, err)
@@ -156,15 +170,13 @@ func NewImporter(ctx context.Context, cfg Config) (*Importer, error) {
 		cfg.BatchSize = 1e5
 	}
 
-	im.wg.Add(int(cfg.Concurrency))
-	for i := 0; i < int(cfg.Concurrency); i++ {
+	for i := range int(cfg.Concurrency) {
 		pbPrefix := fmt.Sprintf(`{{ green "VM worker %d:" }}`, i)
 		bar := barpool.AddWithTemplate(pbPrefix+pbTpl, 0)
 
-		go func(bar barpool.Bar) {
-			defer im.wg.Done()
+		im.wg.Go(func() {
 			im.startWorker(ctx, bar, cfg.BatchSize, cfg.SignificantFigures, cfg.RoundDigits)
-		}(bar)
+		})
 	}
 	im.ResetStats()
 	return im, nil
@@ -313,9 +325,13 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
 		return nil
 	}
 
+	startTime := time.Now()
+	im.importRequestsTotal.Inc()
+
 	pr, pw := io.Pipe()
 	req, err := http.NewRequest(http.MethodPost, im.importPath, pr)
 	if err != nil {
+		im.importRequestsErrorsTotal.Inc()
 		return fmt.Errorf("cannot create request to %q: %s", im.addr, err)
 	}
 	if im.user != "" {
@@ -335,6 +351,7 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
 	if im.compress {
 		zw, err := gzip.NewWriterLevel(w, 1)
 		if err != nil {
+			im.importRequestsErrorsTotal.Inc()
 			return fmt.Errorf("unexpected error when creating gzip writer: %s", err)
 		}
 		w = zw
@@ -346,29 +363,39 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
 	for _, ts := range tsBatch {
 		n, err := ts.write(bw)
 		if err != nil {
+			im.importRequestsErrorsTotal.Inc()
 			return fmt.Errorf("write err: %w", err)
 		}
 		totalBytes += n
 		totalSamples += len(ts.Values)
 	}
 	if err := bw.Flush(); err != nil {
+		im.importRequestsErrorsTotal.Inc()
 		return err
 	}
 	if closer, ok := w.(io.Closer); ok {
 		err := closer.Close()
 		if err != nil {
+			im.importRequestsErrorsTotal.Inc()
 			return err
 		}
 	}
 	if err := pw.Close(); err != nil {
+		im.importRequestsErrorsTotal.Inc()
 		return err
 	}
 
 	requestErr := <-errCh
 	if requestErr != nil {
+		im.importRequestsErrorsTotal.Inc()
+		im.importDuration.UpdateDuration(startTime)
 		return fmt.Errorf("import request error for %q: %w", im.addr, requestErr)
 	}
 
+	im.importSamplesTotal.Add(totalSamples)
+	im.importBytesTotal.Add(totalBytes)
+	im.importDuration.UpdateDuration(startTime)
+
 	im.s.Lock()
 	im.s.bytes += uint64(totalBytes)
 	im.s.samples += uint64(totalSamples)

app/vmctl/vm_native.go

@@ -9,6 +9,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/metrics"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/backoff"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/limiter"
@@ -82,13 +84,19 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
 		if !prompt(ctx, question) {
 			return nil
 		}
+		migrationTenantsTotal.Set(uint64(len(tenants)))
 	}
 
 	for _, tenantID := range tenants {
 		err := p.runBackfilling(ctx, tenantID, ranges)
 		if err != nil {
+			migrationErrorsTotal.Inc()
 			return fmt.Errorf("migration failed: %s", err)
 		}
+
+		if p.interCluster {
+			migrationTenantsProcessed.Inc()
+		}
 	}
 
 	log.Println("Import finished!")
@@ -156,6 +164,7 @@ func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcU
 	p.s.bytes += uint64(written)
 	p.s.requests++
 	p.s.Unlock()
+	migrationBytesTransferredTotal.AddInt64(written)
 
 	if err := pw.Close(); err != nil {
 		return err
@@ -199,7 +208,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 
 	var foundSeriesMsg string
 	var requestsToMake int
-	var metrics = map[string][][]time.Time{
+	var metricsMap = map[string][][]time.Time{
 		"": ranges,
 	}
 
@@ -211,11 +220,11 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 
 	if !p.disablePerMetricRequests {
 		format = fmt.Sprintf(nativeWithBackoffTpl, barPrefix)
-		metrics, err = p.explore(ctx, p.src, tenantID, ranges)
+		metricsMap, err = p.explore(ctx, p.src, tenantID, ranges)
 		if err != nil {
 			return fmt.Errorf("failed to explore metric names: %s", err)
 		}
-		if len(metrics) == 0 {
+		if len(metricsMap) == 0 {
 			errMsg := "no metrics found"
 			if tenantID != "" {
 				errMsg = fmt.Sprintf("%s for tenant id: %s", errMsg, tenantID)
@@ -223,10 +232,14 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 			log.Println(errMsg)
 			return nil
 		}
-		for _, m := range metrics {
+		for _, m := range metricsMap {
 			requestsToMake += len(m)
 		}
-		foundSeriesMsg = fmt.Sprintf("Found %d unique metric names to import. Total import/export requests to make %d", len(metrics), requestsToMake)
+		foundSeriesMsg = fmt.Sprintf("Found %d unique metric names to import. Total import/export requests to make %d", len(metricsMap), requestsToMake)
+
+		migrationMetricsTotal.Add(len(metricsMap))
+	} else {
+		requestsToMake = len(ranges)
 	}
 
 	if !p.interCluster {
@@ -240,6 +253,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 		log.Print(foundSeriesMsg)
 	}
 
+	migrationRequestsPlanned.Add(requestsToMake)
 	bar := barpool.NewSingleProgress(format, requestsToMake)
 	bar.Start()
 	defer bar.Finish()
@@ -248,10 +262,8 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 	errCh := make(chan error, p.cc)
 
 	var wg sync.WaitGroup
-	for i := 0; i < p.cc; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+	for range p.cc {
+		wg.Go(func() {
 			for f := range filterCh {
 				if !p.disablePerMetricRequests {
 					if err := p.do(ctx, f, srcURL, dstURL, nil); err != nil {
@@ -265,12 +277,13 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 						return
 					}
 				}
+				migrationRequestsCompleted.Inc()
 			}
-		}()
+		})
 	}
 
 	// any error breaks the import
-	for mName, mRanges := range metrics {
+	for mName, mRanges := range metricsMap {
 		match, err := buildMatchWithFilter(p.filter.Match, mName)
 		if err != nil {
 			logger.Errorf("failed to build filter %q for metric name %q: %s", p.filter.Match, mName, err)
@@ -290,6 +303,9 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 			}:
 			}
 		}
+		if !p.disablePerMetricRequests {
+			migrationMetricsProcessed.Inc()
+		}
 	}
 
 	close(filterCh)
@@ -398,3 +414,18 @@ func buildMatchWithFilter(filter string, metricName string) (string, error) {
 	match := "{" + strings.Join(filters, " or ") + "}"
 	return match, nil
 }
+
+var (
+	migrationMetricsTotal     = metrics.NewCounter(`vmctl_vm_native_migration_metrics_total`)
+	migrationMetricsProcessed = metrics.NewCounter(`vmctl_vm_native_migration_metrics_processed`)
+
+	migrationRequestsPlanned   = metrics.NewCounter(`vmctl_vm_native_migration_requests_planned`)
+	migrationRequestsCompleted = metrics.NewCounter(`vmctl_vm_native_migration_requests_completed`)
+
+	migrationErrorsTotal = metrics.NewCounter(`vmctl_vm_native_migration_errors_total`)
+
+	migrationTenantsTotal     = metrics.NewCounter(`vmctl_vm_native_migration_tenants_total`)
+	migrationTenantsProcessed = metrics.NewCounter(`vmctl_vm_native_migration_tenants_processed`)
+
+	migrationBytesTransferredTotal = metrics.NewCounter(`vmctl_vm_native_migration_bytes_transferred_total`)
+)

app/vminsert/datadogsketches/request_handler.go

@@ -46,15 +46,14 @@ func insertRows(at *auth.Token, sketches []*datadogsketches.Sketch, extraLabels
 		ms := sketch.ToSummary()
 		for _, m := range ms {
 			ctx.Labels = ctx.Labels[:0]
+			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10557
+			ctx.AddLabel("host", sketch.Host) // newly added
 			ctx.AddLabel("", m.Name)
 			for _, label := range m.Labels {
 				ctx.AddLabel(label.Name, label.Value)
 			}
 			for _, tag := range sketch.Tags {
 				name, value := datadogutil.SplitTag(tag)
-				if name == "host" {
-					name = "exported_host"
-				}
 				ctx.AddLabel(name, value)
 			}
 			for j := range extraLabels {

app/vminsert/main.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -152,7 +153,9 @@ func main() {
 	if len(listenAddrs) == 0 {
 		listenAddrs = []string{":8480"}
 	}
-	go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{UseProxyProtocol: useProxyProtocol})
+	go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
+		UseProxyProtocol: useProxyProtocol,
+	})
 
 	pushmetrics.Init()
 	sig := procutil.WaitForSigterm()
@@ -487,10 +490,5 @@ func checkDuplicates(arr []string) string {
 }
 
 func hasEmptyValues(arr []string) bool {
-	for _, s := range arr {
-		if s == "" {
-			return true
-		}
-	}
-	return false
+	return slices.Contains(arr, "")
 }

app/vminsert/netstorage/netstorage.go

@@ -150,11 +150,7 @@ func (sn *storageNode) run(snb *storageNodesBucket, snIdx int) {
 		replicas = len(sns)
 	}
 
-	sn.readOnlyCheckerWG.Add(1)
-	go func() {
-		defer sn.readOnlyCheckerWG.Done()
-		sn.readOnlyChecker()
-	}()
+	sn.readOnlyCheckerWG.Go(sn.readOnlyChecker)
 	defer sn.readOnlyCheckerWG.Wait()
 
 	d := timeutil.AddJitterToDuration(time.Millisecond * 200)
@@ -214,7 +210,7 @@ func (sn *storageNode) run(snb *storageNodesBucket, snIdx int) {
 func sendBufToReplicasNonblocking(snb *storageNodesBucket, br *bufRows, snIdx, replicas int) bool {
 	usedStorageNodes := make(map[*storageNode]struct{}, replicas)
 	sns := snb.sns
-	for i := 0; i < replicas; i++ {
+	for i := range replicas {
 		idx := snIdx + i
 		attempts := 0
 		for {
@@ -605,10 +601,7 @@ func initStorageNodes(unsortedAddrs []string, rpcCall vminsertapi.RPCCall, hashS
 		sns = append(sns, sn)
 	}
 
-	maxBufSizePerStorageNode = memory.Allowed() / 8 / len(sns)
-	if maxBufSizePerStorageNode > consts.MaxInsertPacketSizeForVMInsert {
-		maxBufSizePerStorageNode = consts.MaxInsertPacketSizeForVMInsert
-	}
+	maxBufSizePerStorageNode = min(memory.Allowed()/8/len(sns), consts.MaxInsertPacketSizeForVMInsert)
 
 	metrics.RegisterSet(ms)
 
@@ -622,11 +615,9 @@ func initStorageNodes(unsortedAddrs []string, rpcCall vminsertapi.RPCCall, hashS
 	}
 
 	for idx, sn := range sns {
-		wg.Add(1)
-		go func(sn *storageNode, idx int) {
+		wg.Go(func() {
 			sn.run(snb, idx)
-			wg.Done()
-		}(sn, idx)
+		})
 	}
 
 	return snb

app/vminsert/netstorage/netstorage_test.go

@@ -14,21 +14,21 @@ func TestInitStopNodes(t *testing.T) {
 	if err := flag.Set("vmstorageDialTimeout", "1ms"); err != nil {
 		t.Fatalf("cannot set vmstorageDialTimeout flag: %s", err)
 	}
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		Init([]string{"host1", "host2"}, 0)
 		runtime.Gosched()
 		MustStop()
 	}
 
 	// Try initializing the netstorage with bigger number of nodes
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		Init([]string{"host1", "host2", "host3"}, 0)
 		runtime.Gosched()
 		MustStop()
 	}
 
 	// Try initializing the netstorage with smaller number of nodes
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		Init([]string{"host1"}, 0)
 		runtime.Gosched()
 		MustStop()

app/vmselect/clusternative/vmselect.go

@@ -146,14 +146,13 @@ type workItem struct {
 
 func newBlockIterator(qt *querytracer.Tracer, denyPartialResponse bool, sq *storage.SearchQuery, deadline searchutil.Deadline) *blockIterator {
 	bi := getBlockIterator()
-	bi.wg.Add(1)
 	workers, processBlocks := netstorage.PrepareProcessRawBlocks(qt, denyPartialResponse, sq, deadline)
 	bi.workCh = make(chan workItem, workers)
 	bi.wis = slicesutil.SetLength(bi.wis, workers)
 	for i := range bi.wis {
 		bi.wis[i].doneCh = make(chan struct{})
 	}
-	go func() {
+	bi.wg.Go(func() {
 		_, err := processBlocks(func(mb []byte, workerID uint) error {
 			wi := bi.wis[workerID]
 			wi.rawMetricBlock = mb
@@ -163,8 +162,7 @@ func newBlockIterator(qt *querytracer.Tracer, denyPartialResponse bool, sq *stor
 		})
 		close(bi.workCh)
 		bi.err = err
-		bi.wg.Done()
-	}()
+	})
 	return bi
 }
 

app/vmselect/graphite/aggr_state.go

@@ -142,7 +142,7 @@ type aggrStatePercentile struct {
 
 func newAggrStatePercentile(pointsLen int, n float64) aggrState {
 	hs := make([]*histogram.Fast, pointsLen)
-	for i := 0; i < pointsLen; i++ {
+	for i := range pointsLen {
 		hs[i] = histogram.NewFast()
 	}
 	return &aggrStatePercentile{

app/vmselect/graphite/eval.go

@@ -10,6 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/searchutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
@@ -52,7 +53,7 @@ func (ec *evalConfig) newTimestamps(step int64) []int64 {
 	pointsLen := ec.pointsLen(step)
 	timestamps := make([]int64, pointsLen)
 	ts := ec.startTime
-	for i := 0; i < pointsLen; i++ {
+	for i := range pointsLen {
 		timestamps[i] = ts
 		ts += step
 	}
@@ -199,12 +200,17 @@ func newNextSeriesForSearchQuery(ec *evalConfig, sq *storage.SearchQuery, expr g
 				pathExpression: safePathExpression(expr),
 			}
 			s.summarize(aggrAvg, ec.startTime, ec.endTime, ec.storageStep, 0)
-			t := timerpool.Get(30 * time.Second)
+
+			// A negative or zero duration will cause timer.C to return immediately
+			remainingTimeout := ec.deadline.Deadline() - fasttime.UnixTimestamp()
+			t := timerpool.Get(time.Duration(remainingTimeout) * time.Second)
 			defer timerpool.Put(t)
+
 			select {
 			case seriesCh <- s:
 			case <-t.C:
-				logger.Errorf("resource leak when processing the %s (full query: %s); please report this error to VictoriaMetrics developers",
+				logger.Errorf("reached timeout when processing the %s (full query: %s), it can be due to the amount of storageNodes configured in vmselect is more than vmselect’s available CPU count "+
+					"or vmselect is heavy loaded. Consider adding resources or increasing `-search.maxQueryDuration` or `timeout` parameter in the query.",
 					expr.AppendString(nil), ec.originalQuery)
 			}
 			return nil

app/vmselect/graphite/natural_compare.go

@@ -25,7 +25,7 @@ func naturalLess(a, b string) bool {
 }
 
 func getNonNumPrefix(s string) (prefix string, tail string) {
-	for i := 0; i < len(s); i++ {
+	for i := range len(s) {
 		ch := s[i]
 		if ch >= '0' && ch <= '9' {
 			return s[:i], s[i:]

app/vmselect/graphite/render_api.go

@@ -84,7 +84,7 @@ func RenderHandler(startTime time.Time, at *auth.Token, w http.ResponseWriter, r
 	if s := r.FormValue("maxDataPoints"); len(s) > 0 {
 		n, err := strconv.ParseFloat(s, 64)
 		if err != nil {
-			return fmt.Errorf("cannot parse maxDataPoints=%q: %w", maxDataPoints, err)
+			return fmt.Errorf("cannot parse maxDataPoints=%d: %w", maxDataPoints, err)
 		}
 		if n <= 0 {
 			return fmt.Errorf("maxDataPoints must be greater than 0; got %f", n)
@@ -214,7 +214,7 @@ func parseInterval(s string) (int64, error) {
 	s = strings.TrimSpace(s)
 	prefix := s
 	var suffix string
-	for i := 0; i < len(s); i++ {
+	for i := range len(s) {
 		ch := s[i]
 		if ch != '-' && ch != '+' && ch != '.' && (ch < '0' || ch > '9') {
 			prefix = s[:i]

app/vmselect/graphite/transform.go

@@ -1228,7 +1228,7 @@ func transformDelay(ec *evalConfig, fe *graphiteql.FuncExpr) (nextSeriesFunc, er
 				stepsLocal = len(values)
 			}
 			copy(values[stepsLocal:], values[:len(values)-stepsLocal])
-			for i := 0; i < stepsLocal; i++ {
+			for i := range stepsLocal {
 				values[i] = nan
 			}
 		}
@@ -1740,7 +1740,7 @@ func transformGroup(ec *evalConfig, fe *graphiteql.FuncExpr) (nextSeriesFunc, er
 
 func groupSeriesLists(ec *evalConfig, args []*graphiteql.ArgExpr, expr graphiteql.Expr) (nextSeriesFunc, error) {
 	var nextSeriess []nextSeriesFunc
-	for i := 0; i < len(args); i++ {
+	for i := range args {
 		nextSeries, err := evalSeriesList(ec, args, "seriesList", i)
 		if err != nil {
 			for _, f := range nextSeriess {
@@ -3233,7 +3233,7 @@ func transformSeriesByTag(ec *evalConfig, fe *graphiteql.FuncExpr) (nextSeriesFu
 		return nil, fmt.Errorf("at least one tagExpression must be passed to seriesByTag")
 	}
 	var tagExpressions []string
-	for i := 0; i < len(args); i++ {
+	for i := range args {
 		te, err := getString(args, "tagExpressions", i)
 		if err != nil {
 			return nil, err
@@ -3633,7 +3633,7 @@ var graphiteToGolangRe = regexp.MustCompile(`\\(\d+)`)
 
 func getNodes(args []*graphiteql.ArgExpr) ([]graphiteql.Expr, error) {
 	var nodes []graphiteql.Expr
-	for i := 0; i < len(args); i++ {
+	for i := range args {
 		expr := args[i].Expr
 		switch expr.(type) {
 		case *graphiteql.NumberExpr, *graphiteql.StringExpr:
@@ -3896,27 +3896,9 @@ func nextSeriesConcurrentWrapper(nextSeries nextSeriesFunc, f func(s *series) (*
 	seriesCh := make(chan *series, goroutines)
 	errCh := make(chan error, 1)
 	var wg sync.WaitGroup
-	wg.Add(goroutines)
-	go func() {
-		var err error
-		for {
-			s, e := nextSeries()
-			if e != nil || s == nil {
-				err = e
-				break
-			}
-			seriesCh <- s
-		}
-		close(seriesCh)
-		wg.Wait()
-		close(resultCh)
-		errCh <- err
-		close(errCh)
-	}()
 	var skipProcessing atomic.Bool
-	for i := 0; i < goroutines; i++ {
-		go func() {
-			defer wg.Done()
+	for range goroutines {
+		wg.Go(func() {
 			for s := range seriesCh {
 				if skipProcessing.Load() {
 					continue
@@ -3934,8 +3916,24 @@ func nextSeriesConcurrentWrapper(nextSeries nextSeriesFunc, f func(s *series) (*
 					}
 				}
 			}
-		}()
+		})
 	}
+	go func() {
+		var err error
+		for {
+			s, e := nextSeries()
+			if e != nil || s == nil {
+				err = e
+				break
+			}
+			seriesCh <- s
+		}
+		close(seriesCh)
+		wg.Wait()
+		close(resultCh)
+		errCh <- err
+		close(errCh)
+	}()
 	wrapper := func() (*series, error) {
 		r := <-resultCh
 		if r == nil {
@@ -4054,7 +4052,7 @@ func formatPathsFromSeriesExpressions(seriesExpressions []string, sortPaths bool
 
 func newNaNSeries(ec *evalConfig, step int64) *series {
 	values := make([]float64, ec.pointsLen(step))
-	for i := 0; i < len(values); i++ {
+	for i := range values {
 		values[i] = nan
 	}
 	return &series{
@@ -5246,7 +5244,7 @@ func transformLinearRegression(ec *evalConfig, fe *graphiteql.FuncExpr) (nextSer
 
 func linearRegressionForSeries(ec *evalConfig, fe *graphiteql.FuncExpr, ss, sourceSeries []*series) (nextSeriesFunc, error) {
 	var resp []*series
-	for i := 0; i < len(ss); i++ {
+	for i := range ss {
 		source := sourceSeries[i]
 		s := ss[i]
 		s.Tags["linearRegressions"] = fmt.Sprintf("%d, %d", ec.startTime/1e3, ec.endTime/1e3)
@@ -5260,7 +5258,7 @@ func linearRegressionForSeries(ec *evalConfig, fe *graphiteql.FuncExpr, ss, sour
 			continue
 		}
 		values := s.Values
-		for j := 0; j < len(values); j++ {
+		for j := range values {
 			values[j] = offset + (float64(int(s.Timestamps[0])+j*int(s.step)))*factor
 		}
 		resp = append(resp, s)
@@ -5372,7 +5370,7 @@ func holtWinterConfidenceBands(ec *evalConfig, fe *graphiteql.FuncExpr, args []*
 		valuesLen := len(forecastValues)
 		upperBand := make([]float64, 0, valuesLen)
 		lowerBand := make([]float64, 0, valuesLen)
-		for i := 0; i < valuesLen; i++ {
+		for i := range valuesLen {
 			forecastItem := forecastValues[i]
 			deviationItem := deviationValues[i]
 			if math.IsNaN(forecastItem) || math.IsNaN(deviationItem) {
@@ -5466,7 +5464,7 @@ func transformHoltWintersAberration(ec *evalConfig, fe *graphiteql.FuncExpr) (ne
 			return nil, fmt.Errorf("bug, len mismatch for series: %d and upperBand values: %d or lowerBand values: %d", len(values), len(upperBand), len(lowerBand))
 		}
 		aberration := make([]float64, 0, len(values))
-		for i := 0; i < len(values); i++ {
+		for i := range values {
 			v := values[i]
 			upperValue := upperBand[i]
 			lowerValue := lowerBand[i]

app/vmselect/graphiteql/lexer.go

@@ -280,7 +280,7 @@ func isMetricExprChar(ch byte) bool {
 }
 
 func appendEscapedIdent(dst []byte, s string) []byte {
-	for i := 0; i < len(s); i++ {
+	for i := range len(s) {
 		ch := s[i]
 		if isIdentChar(ch) || isMetricExprChar(ch) {
 			if i == 0 && !isFirstIdentChar(ch) {

app/vmselect/main.go

@@ -9,6 +9,7 @@ import (
 	nethttputil "net/http/httputil"
 	"net/url"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -138,7 +139,9 @@ func main() {
 	if len(listenAddrs) == 0 {
 		listenAddrs = []string{":8481"}
 	}
-	go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{UseProxyProtocol: useProxyProtocol})
+	go httpserver.Serve(listenAddrs, requestHandler, httpserver.ServeOptions{
+		UseProxyProtocol: useProxyProtocol,
+	})
 
 	pushmetrics.Init()
 	sig := procutil.WaitForSigterm()
@@ -471,19 +474,23 @@ func selectHandler(qt *querytracer.Tracer, startTime time.Time, w http.ResponseW
 		return true
 	case "graphite/tags/tagSeries":
 		graphiteTagsTagSeriesRequests.Inc()
-		if err := graphite.TagsTagSeriesHandler(startTime, at, w, r); err != nil {
-			graphiteTagsTagSeriesErrors.Inc()
-			httpserver.Errorf(w, r, "%s", err)
-			return true
+		err := &httpserver.ErrorWithStatusCode{
+			Err: fmt.Errorf("graphite tag registration has been disabled and is planned to be removed in future. " +
+				"See: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10544"),
+			StatusCode: http.StatusNotImplemented,
 		}
+		graphiteTagsTagSeriesErrors.Inc()
+		httpserver.Errorf(w, r, "%s", err)
 		return true
 	case "graphite/tags/tagMultiSeries":
 		graphiteTagsTagMultiSeriesRequests.Inc()
-		if err := graphite.TagsTagMultiSeriesHandler(startTime, at, w, r); err != nil {
-			graphiteTagsTagMultiSeriesErrors.Inc()
-			httpserver.Errorf(w, r, "%s", err)
-			return true
+		err := &httpserver.ErrorWithStatusCode{
+			Err: fmt.Errorf("graphite tag registration has been disabled and is planned to be removed in future. " +
+				"See: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10544"),
+			StatusCode: http.StatusNotImplemented,
 		}
+		graphiteTagsTagMultiSeriesErrors.Inc()
+		httpserver.Errorf(w, r, "%s", err)
 		return true
 	case "graphite/tags":
 		graphiteTagsRequests.Inc()
@@ -597,6 +604,7 @@ func handleStaticAndSimpleRequests(w http.ResponseWriter, r *http.Request, path
 		}
 		w.Header().Add("Content-Type", "text/html; charset=utf-8")
 		fmt.Fprintf(w, "<h2>VictoriaMetrics cluster - vmselect</h2></br>")
+		fmt.Fprintf(w, "Version %s<br>", buildinfo.Version)
 		fmt.Fprintf(w, "See <a href='https://docs.victoriametrics.com/victoriametrics/cluster-victoriametrics/#url-format'>docs</a></br>")
 		fmt.Fprintf(w, "Useful endpoints:</br>")
 		fmt.Fprintf(w, `<a href="vmui">Web UI</a><br>`)
@@ -1039,10 +1047,5 @@ func checkDuplicates(arr []string) string {
 }
 
 func hasEmptyValues(arr []string) bool {
-	for _, s := range arr {
-		if s == "" {
-			return true
-		}
-	}
-	return false
+	return slices.Contains(arr, "")
 }

app/vmselect/netstorage/netstorage.go

@@ -6,14 +6,17 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"net/http"
 	"os"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 	"unsafe"
 
@@ -335,14 +338,12 @@ func (rss *Results) runParallel(qt *querytracer.Tracer, f func(rs *Result, worke
 
 	// Start workers and wait until they finish the work.
 	var wg sync.WaitGroup
-	for i := range workChs {
-		wg.Add(1)
-		qtChild := qt.NewChild("worker #%d", i)
-		go func(workerID uint) {
-			timeseriesWorker(qtChild, workChs, workerID)
+	for workerID := range workChs {
+		qtChild := qt.NewChild("worker #%d", workerID)
+		wg.Go(func() {
+			timeseriesWorker(qtChild, workChs, uint(workerID))
 			qtChild.Done()
-			wg.Done()
-		}(uint(i))
+		})
 	}
 	wg.Wait()
 
@@ -533,10 +534,7 @@ func (pts *packedTimeseries) unpackTo(dst []*sortBlock, tbfs []*tmpBlocksFile, t
 	}
 
 	// Prepare worker channels.
-	workers := min(len(upws), gomaxprocs)
-	if workers < 1 {
-		workers = 1
-	}
+	workers := max(min(len(upws), gomaxprocs), 1)
 	itemsPerWorker := (len(upws) + workers - 1) / workers
 	workChs := make([]chan *unpackWork, workers)
 	for i := range workChs {
@@ -555,12 +553,10 @@ func (pts *packedTimeseries) unpackTo(dst []*sortBlock, tbfs []*tmpBlocksFile, t
 
 	// Start workers and wait until they finish the work.
 	var wg sync.WaitGroup
-	for i := 0; i < workers; i++ {
-		wg.Add(1)
-		go func(workerID uint) {
-			unpackWorker(workChs, workerID)
-			wg.Done()
-		}(uint(i))
+	for workerID := range workers {
+		wg.Go(func() {
+			unpackWorker(workChs, uint(workerID))
+		})
 	}
 	wg.Wait()
 
@@ -623,6 +619,7 @@ func mergeSortBlocks(dst *Result, sbh *sortBlocksHeap, dedupInterval int64) {
 		return
 	}
 	heap.Init(sbh)
+	var dedupSamples int
 	for {
 		sbs := sbh.sbs
 		top := sbs[0]
@@ -638,6 +635,7 @@ func mergeSortBlocks(dst *Result, sbh *sortBlocksHeap, dedupInterval int64) {
 		if n := equalSamplesPrefix(top, sbNext); n > 0 && dedupInterval > 0 {
 			// Skip n replicated samples at top if deduplication is enabled.
 			top.NextIdx = topNextIdx + n
+			dedupSamples += n
 		} else {
 			// Copy samples from top to dst with timestamps not exceeding tsNext.
 			top.NextIdx = topNextIdx + binarySearchTimestamps(top.Timestamps[topNextIdx:], tsNext)
@@ -652,8 +650,8 @@ func mergeSortBlocks(dst *Result, sbh *sortBlocksHeap, dedupInterval int64) {
 		}
 	}
 	timestamps, values := storage.DeduplicateSamples(dst.Timestamps, dst.Values, dedupInterval)
-	dedups := len(dst.Timestamps) - len(timestamps)
-	dedupsDuringSelect.Add(dedups)
+	dedupSamples += len(dst.Timestamps) - len(timestamps)
+	dedupsDuringSelect.Add(dedupSamples)
 	dst.Timestamps = timestamps
 	dst.Values = values
 }
@@ -679,7 +677,7 @@ func equalTimestampsPrefix(a, b []int64) int {
 
 func equalValuesPrefix(a, b []float64) int {
 	for i, v := range a {
-		if i >= len(b) || v != b[i] {
+		if i >= len(b) || math.Float64bits(v) != math.Float64bits(b[i]) {
 			return i
 		}
 	}
@@ -987,12 +985,7 @@ func GraphiteTags(qt *querytracer.Tracer, accountID, projectID uint32, denyParti
 }
 
 func hasString(a []string, s string) bool {
-	for _, x := range a {
-		if x == s {
-			return true
-		}
-	}
-	return false
+	return slices.Contains(a, s)
 }
 
 // LabelValues returns label values matching the given labelName and sq until the given deadline.
@@ -1003,30 +996,19 @@ func LabelValues(qt *querytracer.Tracer, denyPartialResponse bool, labelName str
 		return nil, false, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
 
+	err := populateSqTenantTokensIfNeeded(sq)
+	if err != nil {
+		return nil, false, err
+	}
+
 	if sq.IsMultiTenant && isTenancyLabel(labelName) {
-		tenants, err := Tenants(qt, sq.GetTimeRange(), deadline)
-		if err != nil {
-			return nil, false, err
-		}
-
-		var idx int
-		switch labelName {
-		case "vm_account_id":
-			idx = 0
-		case "vm_project_id":
-			idx = 1
-		default:
-			logger.Panicf("BUG: unexpected labeName=%q", labelName)
-		}
-
-		labelValues := make([]string, 0, len(tenants))
-		for _, t := range tenants {
-			s := strings.Split(t, ":")
-			if len(s) != 2 {
-				logger.Panicf("BUG: unexpected tenant received from storage: %q", t)
+		labelValues := make([]string, 0, len(sq.TenantTokens))
+		for _, t := range sq.TenantTokens {
+			v := t.AccountID
+			if labelName == "vm_project_id" {
+				v = t.ProjectID
 			}
-
-			labelValues = append(labelValues, s[idx])
+			labelValues = append(labelValues, fmt.Sprintf("%d", v))
 		}
 
 		labelValues = prepareLabelValues(qt, labelValues, maxLabelValues)
@@ -1038,10 +1020,6 @@ func LabelValues(qt *querytracer.Tracer, denyPartialResponse bool, labelName str
 		labelValues []string
 		err         error
 	}
-	err := populateSqTenantTokensIfNeeded(sq)
-	if err != nil {
-		return nil, false, err
-	}
 	sns := getStorageNodes()
 	snr := startStorageNodesRequest(qt, sns, denyPartialResponse, func(qt *querytracer.Tracer, _ uint, sn *storageNode) any {
 		return execSearchQuery(qt, sq, func(qt *querytracer.Tracer, requestData []byte, _ storage.TenantToken) any {
@@ -2089,8 +2067,7 @@ func (snr *storageNodesRequest) finishQueryTracer(qt *querytracer.Tracer, msg st
 }
 
 func (snr *storageNodesRequest) collectAllResults(f func(result any) error) error {
-	sns := snr.sns
-	for i := 0; i < len(sns); i++ {
+	for range snr.sns {
 		result := <-snr.resultsCh
 		if err := f(result.data); err != nil {
 			snr.finishQueryTracer(result.qt, fmt.Sprintf("error: %s", err))
@@ -2494,7 +2471,7 @@ func (sn *storageNode) processSearchQuery(qt *querytracer.Tracer, requestData []
 
 func (sn *storageNode) execOnConnWithPossibleRetry(qt *querytracer.Tracer, funcName string, f func(bc *handshake.BufferedConn) error, deadline searchutil.Deadline) error {
 	qtChild := qt.NewChild("rpc call %s()", funcName)
-	err := sn.execOnConn(qtChild, funcName, f, deadline)
+	err := sn.execOnConn(qtChild, funcName, f, deadline, false)
 	defer qtChild.Done()
 	if err == nil {
 		return nil
@@ -2514,14 +2491,22 @@ func (sn *storageNode) execOnConnWithPossibleRetry(qt *querytracer.Tracer, funcN
 	}
 	// Repeat the query in the hope the error was temporary.
 	qtRetry := qtChild.NewChild("retry rpc call %s() after error", funcName)
-	err = sn.execOnConn(qtRetry, funcName, f, deadline)
+	// Retry with a new connection if the error is io.EOF, "broken pipe", or "reset by peer".
+	// These errors usually indicate that the connection was closed by vmstorage
+	// during a rolling restart but is still present in the
+	// connection pool. Reusing such stale connections may cause query failures
+	// or partial responses. Dialing a new connection allows the request to
+	// proceed without waiting for the broken connection to be evicted from the pool.
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10314
+	dialConn := errors.Is(err, io.EOF) || errors.Is(err, syscall.EPIPE) || errors.Is(err, syscall.ECONNRESET)
+	err = sn.execOnConn(qtRetry, funcName, f, deadline, dialConn)
 	qtRetry.Done()
 	return err
 }
 
 var errCannotObtainConn = fmt.Errorf("cannot obtain connection from a pool")
 
-func (sn *storageNode) execOnConn(qt *querytracer.Tracer, funcName string, f func(bc *handshake.BufferedConn) error, deadline searchutil.Deadline) error {
+func (sn *storageNode) execOnConn(qt *querytracer.Tracer, funcName string, f func(bc *handshake.BufferedConn) error, deadline searchutil.Deadline, forceNew bool) error {
 	sn.concurrentQueries.Inc()
 	defer sn.concurrentQueries.Dec()
 
@@ -2532,7 +2517,13 @@ func (sn *storageNode) execOnConn(qt *querytracer.Tracer, funcName string, f fun
 	if timeout <= 0 {
 		return fmt.Errorf("request timeout reached: %s", deadline.String())
 	}
-	bc, err := sn.connPool.Get()
+	var bc *handshake.BufferedConn
+	var err error
+	if forceNew {
+		bc, err = sn.connPool.Dial()
+	} else {
+		bc, err = sn.connPool.Get()
+	}
 	if err != nil {
 		return fmt.Errorf("%w: %w", errCannotObtainConn, err)
 	}
@@ -2857,7 +2848,7 @@ func (sn *storageNode) getTagValueSuffixesOnConn(bc *handshake.BufferedConn, acc
 		return nil, fmt.Errorf("cannot read the number of tag value suffixes: %w", err)
 	}
 	suffixes := make([]string, 0, suffixesCount)
-	for i := 0; i < int(suffixesCount); i++ {
+	for i := range int(suffixesCount) {
 		buf, err = readBytes(buf[:0], bc, maxLabelValueSize)
 		if err != nil {
 			return nil, fmt.Errorf("cannot read tag value suffix #%d: %w", i+1, err)
@@ -2949,7 +2940,7 @@ func readTopHeapEntries(bc *handshake.BufferedConn) ([]storage.TopHeapEntry, err
 	}
 	var a []storage.TopHeapEntry
 	var buf []byte
-	for i := uint64(0); i < n; i++ {
+	for range n {
 		buf, err = readBytes(buf[:0], bc, maxLabelNameSize)
 		if err != nil {
 			return nil, fmt.Errorf("cannot read label name: %w", err)
@@ -3023,7 +3014,7 @@ func (sn *storageNode) processSearchMetricNamesOnConn(bc *handshake.BufferedConn
 		return nil, fmt.Errorf("cannot read metricNamesCount: %w", err)
 	}
 	metricNames := make([]string, metricNamesCount)
-	for i := int64(0); i < int64(metricNamesCount); i++ {
+	for i := range int64(metricNamesCount) {
 		buf, err = readBytes(buf[:0], bc, maxMetricNameSize)
 		if err != nil {
 			return nil, fmt.Errorf("cannot read metricName #%d: %w", i+1, err)
@@ -3232,14 +3223,12 @@ func initStorageNodes(addrs []string) *storageNodesBucket {
 		groupName, addr = netutil.ParseGroupAddr(addr)
 		group := groupsMap[groupName]
 
-		wg.Add(1)
-		go func(addr string) {
-			defer wg.Done()
+		wg.Go(func() {
 			sn := newStorageNode(ms, group, addr)
 			snsLock.Lock()
 			sns = append(sns, sn)
 			snsLock.Unlock()
-		}(addr)
+		})
 	}
 	wg.Wait()
 	metrics.RegisterSet(ms)

app/vmselect/netstorage/netstorage_test.go

@@ -2,30 +2,33 @@ package netstorage
 
 import (
 	"flag"
+	"math"
 	"reflect"
 	"runtime"
 	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
 )
 
 func TestInitStopNodes(t *testing.T) {
 	if err := flag.Set("vmstorageDialTimeout", "1ms"); err != nil {
 		t.Fatalf("cannot set vmstorageDialTimeout flag: %s", err)
 	}
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		Init([]string{"host1", "host2"})
 		runtime.Gosched()
 		MustStop()
 	}
 
 	// Try initializing the netstorage with bigger number of nodes
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		Init([]string{"host1", "host2", "host3"})
 		runtime.Gosched()
 		MustStop()
 	}
 
 	// Try initializing the netstorage with smaller number of nodes
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		Init([]string{"host1"})
 		runtime.Gosched()
 		MustStop()
@@ -221,3 +224,111 @@ func TestMergeSortBlocks(t *testing.T) {
 		Values:     []float64{7, 24, 26},
 	})
 }
+
+func TestEqualSamplesPrefix(t *testing.T) {
+	f := func(a, b *sortBlock, expected int) {
+		t.Helper()
+
+		actual := equalSamplesPrefix(a, b)
+		if actual != expected {
+			t.Fatalf("unexpected result: got %d, want %d", actual, expected)
+		}
+	}
+
+	// Empty blocks
+	f(&sortBlock{}, &sortBlock{}, 0)
+
+	// Identical blocks
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, 4)
+
+	// Non-zero NextIdx
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+		NextIdx:    2,
+	}, &sortBlock{
+		Timestamps: []int64{10, 20, 3, 4},
+		Values:     []float64{50, 60, 7, 8},
+		NextIdx:    2,
+	}, 2)
+
+	// Non-zero NextIdx with mismatch
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+		NextIdx:    1,
+	}, &sortBlock{
+		Timestamps: []int64{10, 2, 3, 4},
+		Values:     []float64{50, 6, 7, 80},
+		NextIdx:    1,
+	}, 2)
+
+	// Different lengths
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2, 3},
+		Values:     []float64{5, 6, 7},
+	}, 3)
+
+	// Timestamps diverge
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2, 30, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, 2)
+
+	// Values diverge
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 60, 7, 8},
+	}, 1)
+
+	// Zero matches
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, 6, 7, 8},
+	}, &sortBlock{
+		Timestamps: []int64{5, 6, 7, 8},
+		Values:     []float64{1, 2, 3, 4},
+	}, 0)
+
+	// Compare staleness markers, matching
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, decimal.StaleNaN, 7, 8},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{5, decimal.StaleNaN, 7, 8},
+	}, 4)
+
+	// Special float values: +Inf, -Inf, 0, -0
+	f(&sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{math.Inf(1), math.Inf(-1), math.Copysign(0, +1), math.Copysign(0, -1)},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2, 3, 4},
+		Values:     []float64{math.Inf(1), math.Inf(-1), math.Copysign(0, +1), math.Copysign(0, -1)},
+	}, 4)
+
+	// Positive zero vs negative zero (bitwise different)
+	f(&sortBlock{
+		Timestamps: []int64{1, 2},
+		Values:     []float64{5, math.Copysign(0, +1)},
+	}, &sortBlock{
+		Timestamps: []int64{1, 2},
+		Values:     []float64{5, math.Copysign(0, -1)},
+	}, 1)
+}

app/vmselect/netstorage/netstorage_timing_test.go

@@ -10,14 +10,14 @@ func BenchmarkMergeSortBlocks(b *testing.B) {
 		b.Run(fmt.Sprintf("replicationFactor-%d", replicationFactor), func(b *testing.B) {
 			const samplesPerBlock = 8192
 			var blocks []*sortBlock
-			for j := 0; j < 10; j++ {
+			for j := range 10 {
 				timestamps := make([]int64, samplesPerBlock)
 				values := make([]float64, samplesPerBlock)
 				for i := range timestamps {
 					timestamps[i] = int64(j*samplesPerBlock + i)
 					values[i] = float64(j*samplesPerBlock + i)
 				}
-				for i := 0; i < replicationFactor; i++ {
+				for range replicationFactor {
 					blocks = append(blocks, &sortBlock{
 						Timestamps: timestamps,
 						Values:     values,
@@ -30,7 +30,7 @@ func BenchmarkMergeSortBlocks(b *testing.B) {
 	b.Run("overlapped-blocks-bestcase", func(b *testing.B) {
 		const samplesPerBlock = 8192
 		var blocks []*sortBlock
-		for j := 0; j < 10; j++ {
+		for j := range 10 {
 			timestamps := make([]int64, samplesPerBlock)
 			values := make([]float64, samplesPerBlock)
 			for i := range timestamps {
@@ -45,7 +45,7 @@ func BenchmarkMergeSortBlocks(b *testing.B) {
 		for j := 1; j < len(blocks); j++ {
 			prev := blocks[j-1].Timestamps
 			curr := blocks[j].Timestamps
-			for i := 0; i < samplesPerBlock/2; i++ {
+			for i := range samplesPerBlock / 2 {
 				prev[i+samplesPerBlock/2], curr[i] = curr[i], prev[i+samplesPerBlock/2]
 			}
 		}
@@ -54,7 +54,7 @@ func BenchmarkMergeSortBlocks(b *testing.B) {
 	b.Run("overlapped-blocks-worstcase", func(b *testing.B) {
 		const samplesPerBlock = 8192
 		var blocks []*sortBlock
-		for j := 0; j < 5; j++ {
+		for j := range 5 {
 			timestamps := make([]int64, samplesPerBlock)
 			values := make([]float64, samplesPerBlock)
 			for i := range timestamps {

app/vmselect/netstorage/tmp_blocks_file_test.go

@@ -32,14 +32,14 @@ func TestTmpBlocksFileSerial(t *testing.T) {
 func TestTmpBlocksFileConcurrent(t *testing.T) {
 	concurrency := 3
 	ch := make(chan error, concurrency)
-	for i := 0; i < concurrency; i++ {
+	for range concurrency {
 		go func() {
 			ch <- testTmpBlocksFile()
 		}()
 	}
 	timer := time.NewTimer(30 * time.Second)
 	defer timer.Stop()
-	for i := 0; i < concurrency; i++ {
+	for range concurrency {
 		select {
 		case err := <-ch:
 			if err != nil {
@@ -56,7 +56,7 @@ func testTmpBlocksFile() error {
 		rowsCount := rand.Intn(8000) + 1
 		var timestamps, values []int64
 		ts := int64(rand.Intn(1023434))
-		for i := 0; i < rowsCount; i++ {
+		for i := range rowsCount {
 			ts += int64(rand.Intn(1000) + 1)
 			timestamps = append(timestamps, ts)
 			values = append(values, int64(i*i+rand.Intn(20)))
@@ -109,7 +109,7 @@ func testTmpBlocksFile() error {
 			concurrency := 2
 			workCh := make(chan int)
 			doneCh := make(chan error)
-			for i := 0; i < concurrency; i++ {
+			for range concurrency {
 				go func() {
 					doneCh <- func() error {
 						var b1 storage.Block
@@ -144,7 +144,7 @@ func testTmpBlocksFile() error {
 				workCh <- i
 			}
 			close(workCh)
-			for i := 0; i < concurrency; i++ {
+			for range concurrency {
 				select {
 				case err := <-doneCh:
 					if err != nil {

app/vmselect/prometheus/prometheus.go

@@ -6,11 +6,13 @@ import (
 	"math"
 	"net/http"
 	"runtime"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
+	"unicode/utf8"
 
 	"github.com/VictoriaMetrics/metrics"
 	"github.com/VictoriaMetrics/metricsql"
@@ -638,6 +640,12 @@ func LabelValuesHandler(qt *querytracer.Tracer, startTime time.Time, at *auth.To
 	if err != nil {
 		return err
 	}
+	if strings.HasPrefix(labelName, "U__") {
+		// This label seems to be Unicode-encoded according to the Prometheus spec.
+		// See https://prometheus.io/docs/prometheus/latest/querying/api/#querying-label-values
+		// Spec: https://github.com/prometheus/proposals/blob/main/proposals/0028-utf8.md
+		labelName = unescapePrometheusLabelName(labelName)
+	}
 	labelValues, isPartial, err := netstorage.LabelValues(qt, denyPartialResponse, labelName, sq, limit, cp.deadline)
 	if err != nil {
 		return fmt.Errorf("cannot obtain values for label %q: %w", labelName, err)
@@ -1201,14 +1209,7 @@ func removeEmptyValuesAndTimeseries(tss []netstorage.Result) []netstorage.Result
 	dst := tss[:0]
 	for i := range tss {
 		ts := &tss[i]
-		hasNaNs := false
-		for _, v := range ts.Values {
-			if math.IsNaN(v) {
-				hasNaNs = true
-				break
-			}
-		}
-		if !hasNaNs {
+		if !slices.ContainsFunc(ts.Values, math.IsNaN) {
 			// Fast path: nothing to remove.
 			if len(ts.Values) > 0 {
 				dst = append(dst, *ts)
@@ -1500,3 +1501,70 @@ func (sw *scalableWriter) flush() error {
 	})
 	return sw.bw.Flush()
 }
+
+// copied from https://github.com/prometheus/common/blob/adea6285c1c7447fcb7bfdeb6abfc6eff893e0a7/model/metric.go#L483
+// it's not possible to use direct import due to increased binary size
+func unescapePrometheusLabelName(name string) string {
+	// lower function taken from strconv.atoi.
+	lower := func(c byte) byte {
+		return c | ('x' - 'X')
+	}
+	if len(name) == 0 {
+		return name
+	}
+	escapedName, found := strings.CutPrefix(name, "U__")
+	if !found {
+		return name
+	}
+
+	var unescaped strings.Builder
+TOP:
+	for i := 0; i < len(escapedName); i++ {
+		// All non-underscores are treated normally.
+		if escapedName[i] != '_' {
+			unescaped.WriteByte(escapedName[i])
+			continue
+		}
+		i++
+		if i >= len(escapedName) {
+			return name
+		}
+		// A double underscore is a single underscore.
+		if escapedName[i] == '_' {
+			unescaped.WriteByte('_')
+			continue
+		}
+		// We think we are in a UTF-8 code, process it.
+		var utf8Val uint
+		for j := 0; i < len(escapedName); j++ {
+			// This is too many characters for a utf8 value based on the MaxRune
+			// value of '\U0010FFFF'.
+			if j >= 6 {
+				return name
+			}
+			// Found a closing underscore, convert to a rune, check validity, and append.
+			if escapedName[i] == '_' {
+				utf8Rune := rune(utf8Val)
+				if !utf8.ValidRune(utf8Rune) {
+					return name
+				}
+				unescaped.WriteRune(utf8Rune)
+				continue TOP
+			}
+			r := lower(escapedName[i])
+			utf8Val *= 16
+			switch {
+			case r >= '0' && r <= '9':
+				utf8Val += uint(r) - '0'
+			case r >= 'a' && r <= 'f':
+				utf8Val += uint(r) - 'a' + 10
+			default:
+				return name
+			}
+			i++
+		}
+		// Didn't find closing underscore, invalid.
+		return name
+	}
+	return unescaped.String()
+}

app/vmselect/promql/aggr.go

@@ -742,7 +742,7 @@ func getRangeTopKTimeseries(tss []*timeseries, modifier *metricsql.ModifierExpr,
 
 func reverseSeries(tss []*timeseries) {
 	j := len(tss)
-	for i := 0; i < len(tss)/2; i++ {
+	for i := range len(tss) / 2 {
 		j--
 		tss[i], tss[j] = tss[j], tss[i]
 	}
@@ -983,7 +983,7 @@ func getPerPointIQRBounds(tss []*timeseries) ([]float64, []float64) {
 	var qs []float64
 	lower := make([]float64, pointsLen)
 	upper := make([]float64, pointsLen)
-	for i := 0; i < pointsLen; i++ {
+	for i := range pointsLen {
 		values = values[:0]
 		for _, ts := range tss {
 			v := ts.Values[i]

app/vmselect/promql/aggr_incremental_test.go

@@ -53,7 +53,7 @@ func TestIncrementalAggr(t *testing.T) {
 			Values:     valuesExpected,
 		}}
 		// run the test multiple times to make sure there are no side effects on concurrency
-		for i := 0; i < 10; i++ {
+		for i := range 10 {
 			iafc := newIncrementalAggrFuncContext(ae, callbacks)
 			tssSrcCopy := copyTimeseries(tssSrc)
 			if err := testIncrementalParallelAggr(iafc, tssSrcCopy, tssExpected); err != nil {
@@ -103,15 +103,13 @@ func testIncrementalParallelAggr(iafc *incrementalAggrFuncContext, tssSrc, tssEx
 	workersCount := netstorage.MaxWorkers()
 	tsCh := make(chan *timeseries)
 	var wg sync.WaitGroup
-	wg.Add(workersCount)
-	for i := 0; i < workersCount; i++ {
-		go func(workerID uint) {
-			defer wg.Done()
+	for workerID := range workersCount {
+		wg.Go(func() {
 			for ts := range tsCh {
 				runtime.Gosched() // allow other goroutines performing the work
-				iafc.updateTimeseries(ts, workerID)
+				iafc.updateTimeseries(ts, uint(workerID))
 			}
-		}(uint(i))
+		})
 	}
 	for _, ts := range tssSrc {
 		tsCh <- ts

app/vmselect/promql/eval.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"math"
 	"regexp"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -496,22 +497,18 @@ func execBinaryOpArgs(qt *querytracer.Tracer, ec *EvalConfig, exprFirst, exprSec
 		var tssFirst []*timeseries
 		var errFirst error
 		qtFirst := qt.NewChild("expr1")
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			tssFirst, errFirst = evalExpr(qtFirst, ec, exprFirst)
 			qtFirst.Done()
-		}()
+		})
 
 		var tssSecond []*timeseries
 		var errSecond error
 		qtSecond := qt.NewChild("expr2")
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			tssSecond, errSecond = evalExpr(qtSecond, ec, exprSecond)
 			qtSecond.Done()
-		}()
+		})
 
 		wg.Wait()
 		if errFirst != nil {
@@ -729,17 +726,13 @@ func evalExprsInParallel(qt *querytracer.Tracer, ec *EvalConfig, es []metricsql.
 	qt.Printf("eval function args in parallel")
 	var wg sync.WaitGroup
 	for i, e := range es {
-		wg.Add(1)
 		qtChild := qt.NewChild("eval arg %d", i)
-		go func(e metricsql.Expr, i int) {
-			defer func() {
-				qtChild.Done()
-				wg.Done()
-			}()
+		wg.Go(func() {
+			defer qtChild.Done()
 			rv, err := evalExpr(qtChild, ec, e)
 			rvs[i] = rv
 			errs[i] = err
-		}(e, i)
+		})
 	}
 	wg.Wait()
 	for _, err := range errs {
@@ -1040,16 +1033,14 @@ func doParallel(tss []*timeseries, f func(ts *timeseries, values []float64, time
 	}
 
 	var wg sync.WaitGroup
-	wg.Add(workers)
-	for i := 0; i < workers; i++ {
-		go func(workerID uint) {
-			defer wg.Done()
+	for workerID := range workers {
+		wg.Go(func() {
 			var tmpValues []float64
 			var tmpTimestamps []int64
 			for ts := range workChs[workerID] {
-				tmpValues, tmpTimestamps = f(ts, tmpValues, tmpTimestamps, workerID)
+				tmpValues, tmpTimestamps = f(ts, tmpValues, tmpTimestamps, uint(workerID))
 			}
-		}(uint(i))
+		})
 	}
 	wg.Wait()
 }
@@ -1206,6 +1197,61 @@ func evalInstantRollup(qt *querytracer.Tracer, ec *EvalConfig, funcName string,
 			},
 		}
 		return evalExpr(qt, ec, be)
+	// the cached rate result could be inaccurate in edge cases, see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10098
+	case "rate":
+		if iafc != nil {
+			if !strings.EqualFold(iafc.ae.Name, "sum") {
+				qt.Printf("do not apply instant rollup optimization for incremental aggregate %s()", iafc.ae.Name)
+				return evalAt(qt, timestamp, window)
+			}
+			qt.Printf("optimized calculation for sum(rate(m[d])) as (sum(increase(m[d])) / d)")
+			afe := expr.(*metricsql.AggrFuncExpr)
+			fe := afe.Args[0].(*metricsql.FuncExpr)
+			feIncrease := *fe
+			feIncrease.Name = "increase"
+			// copy RollupExpr to drop possible offset,
+			// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9762
+			newArg := copyRollupExpr(fe.Args[0].(*metricsql.RollupExpr))
+			newArg.Offset = nil
+			feIncrease.Args = []metricsql.Expr{newArg}
+			d := newArg.Window.Duration(ec.Step)
+			if d == 0 {
+				d = ec.Step
+			}
+			afeIncrease := *afe
+			afeIncrease.Args = []metricsql.Expr{&feIncrease}
+			be := &metricsql.BinaryOpExpr{
+				Op:              "/",
+				KeepMetricNames: true,
+				Left:            &afeIncrease,
+				Right: &metricsql.NumberExpr{
+					N: float64(d) / 1000,
+				},
+			}
+			return evalExpr(qt, ec, be)
+		}
+		qt.Printf("optimized calculation for instant rollup rate(m[d]) as (increase(m[d]) / d)")
+		fe := expr.(*metricsql.FuncExpr)
+		feIncrease := *fe
+		feIncrease.Name = "increase"
+		// copy RollupExpr to drop possible offset,
+		// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9762
+		newArg := copyRollupExpr(fe.Args[0].(*metricsql.RollupExpr))
+		newArg.Offset = nil
+		feIncrease.Args = []metricsql.Expr{newArg}
+		d := newArg.Window.Duration(ec.Step)
+		if d == 0 {
+			d = ec.Step
+		}
+		be := &metricsql.BinaryOpExpr{
+			Op:              "/",
+			KeepMetricNames: fe.KeepMetricNames,
+			Left:            &feIncrease,
+			Right: &metricsql.NumberExpr{
+				N: float64(d) / 1000,
+			},
+		}
+		return evalExpr(qt, ec, be)
 	case "max_over_time":
 		if iafc != nil {
 			if !strings.EqualFold(iafc.ae.Name, "max") {
@@ -1771,6 +1817,7 @@ func evalRollupFuncNoCache(qt *querytracer.Tracer, ec *EvalConfig, funcName stri
 		return nil, err
 	}
 	defer rml.Put(uint64(rollupMemorySize))
+	qs.addMemoryUsage(rollupMemorySize)
 	qt.Printf("the rollup evaluation needs an estimated %d bytes of RAM for %d series and %d points per series (summary %d points)",
 		rollupMemorySize, timeseriesLen, pointsPerSeries, rollupPoints)
 
@@ -1996,14 +2043,7 @@ func dropStaleNaNs(funcName string, values []float64, timestamps []int64) ([]flo
 		return values, timestamps
 	}
 	// Remove Prometheus staleness marks, so non-default rollup functions don't hit NaN values.
-	hasStaleSamples := false
-	for _, v := range values {
-		if decimal.IsStaleNaN(v) {
-			hasStaleSamples = true
-			break
-		}
-	}
-	if !hasStaleSamples {
+	if !slices.ContainsFunc(values, decimal.IsStaleNaN) {
 		// Fast path: values have no Prometheus staleness marks.
 		return values, timestamps
 	}

app/vmselect/promql/exec.go

@@ -44,7 +44,7 @@ func Exec(qt *querytracer.Tracer, ec *EvalConfig, q string, isFirstPointOnly boo
 				return
 			}
 			at := ec.AuthTokens[0]
-			querystats.RegisterQuery(at.AccountID, at.ProjectID, q, ec.End-ec.Start, startTime)
+			querystats.RegisterQuery(at.AccountID, at.ProjectID, q, ec.End-ec.Start, startTime, ec.QueryStats.memoryUsage())
 		}()
 	}
 
@@ -319,7 +319,7 @@ func escapeDots(s string) string {
 		return s
 	}
 	result := make([]byte, 0, len(s)+2*dotsCount)
-	for i := 0; i < len(s); i++ {
+	for i := range len(s) {
 		if s[i] == '.' && (i == 0 || s[i-1] != '\\') && (i+1 == len(s) || i+1 < len(s) && s[i+1] != '*' && s[i+1] != '+' && s[i+1] != '{') {
 			// Escape a dot if the following conditions are met:
 			// - if it isn't escaped already, i.e. if there is no `\` char before the dot.

app/vmselect/promql/exec_test.go

@@ -78,7 +78,7 @@ func TestExecSuccess(t *testing.T) {
 			Deadline:           searchutil.NewDeadline(time.Now(), time.Minute, ""),
 			RoundDigits:        100,
 		}
-		for i := 0; i < 5; i++ {
+		for range 5 {
 			result, err := Exec(nil, ec, q, false)
 			if err != nil {
 				t.Fatalf(`unexpected error when executing %q: %s`, q, err)
@@ -4029,6 +4029,12 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(scalar)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(123, 456, time())`
+		resultExpected := []netstorage.Result{}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(single-value-no-le)`, func(t *testing.T) {
 		t.Parallel()
 		q := `histogram_quantile(0.6, label_set(100, "foo", "bar"))`
@@ -4041,6 +4047,12 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(single-value-no-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(123,456, label_set(100, "foo", "bar"))`
+		resultExpected := []netstorage.Result{}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(single-value-invalid-le)`, func(t *testing.T) {
 		t.Parallel()
 		q := `histogram_quantile(0.6, label_set(100, "le", "foobar"))`
@@ -4053,6 +4065,12 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(single-value-invalid-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(50, 60, label_set(100, "le", "foobar"))`
+		resultExpected := []netstorage.Result{}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(single-value-inf-le)`, func(t *testing.T) {
 		t.Parallel()
 		q := `histogram_quantile(0.6, label_set(100, "le", "+Inf"))`
@@ -4194,6 +4212,28 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(single-value-valid-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(0, 100, label_set(100, "le", "200"))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0.5, 0.5, 0.5, 0.5, 0.5, 0.5},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
+	t.Run(`histogram_fraction(single-value-valid-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(200, 300, label_set(100, "le", "200"))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0, 0, 0, 0, 0, 0},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(single-value-valid-le, boundsLabel)`, func(t *testing.T) {
 		t.Parallel()
 		q := `sort(histogram_quantile(0.6, label_set(100, "le", "200"), "foobar"))`
@@ -4223,7 +4263,7 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1, r2, r3}
 		f(q, resultExpected)
 	})
-	t.Run(`histogram_quantile(single-value-valid-le, boundsLabel)`, func(t *testing.T) {
+	t.Run(`histogram_share(single-value-valid-le, boundsLabel)`, func(t *testing.T) {
 		t.Parallel()
 		q := `sort(histogram_share(120, label_set(100, "le", "200"), "foobar"))`
 		r1 := netstorage.Result{
@@ -4322,7 +4362,37 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
-	t.Run(`histogram_share(single-value-valid-le-mid-le)`, func(t *testing.T) {
+	t.Run(`histogram_fraction(single-value-valid-le-max-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(0,100, (
+			label_set(100, "le", "100"),
+			label_set(40, "le", "50"),
+			label_set(0, "le", "10"),
+		))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{1, 1, 1, 1, 1, 1},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
+	t.Run(`histogram_fraction(single-value-valid-le-min-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(0,10, (
+			label_set(100, "le", "100"),
+			label_set(40, "le", "50"),
+			label_set(0, "le", "10"),
+		))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0, 0, 0, 0, 0, 0},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
+	t.Run(`histogram_share(single-value-valid-le-mid-le-1)`, func(t *testing.T) {
 		t.Parallel()
 		q := `histogram_share(105, (
 			label_set(100, "le", "200"),
@@ -4336,6 +4406,34 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_share(single-value-valid-le-mid-le-2)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_share(55, (
+			label_set(100, "le", "200"),
+			label_set(0, "le", "55"),
+		))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0, 0, 0, 0, 0, 0},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
+	t.Run(`histogram_fraction(single-value-valid-le-mid-le)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(55,105, (
+			label_set(100, "le", "200"),
+			label_set(0, "le", "55"),
+		))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0.3448275862068966, 0.3448275862068966, 0.3448275862068966, 0.3448275862068966, 0.3448275862068966, 0.3448275862068966},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(single-value-valid-le-min-phi-no-zero-bucket)`, func(t *testing.T) {
 		t.Parallel()
 		q := `histogram_quantile(0, label_set(100, "le", "200"))`
@@ -4369,6 +4467,17 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(scalar-phi)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(25, time() / 8, label_set(100, "le", "200"))`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0.5, 0.625, 0.75, 0.875, 0.875, 0.875},
+			Timestamps: timestampsExpected,
+		}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(duplicate-le)`, func(t *testing.T) {
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/pull/3225
 		t.Parallel()
@@ -4450,6 +4559,36 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r1, r2}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(valid)`, func(t *testing.T) {
+		t.Parallel()
+		q := `sort(histogram_fraction(0, 25,
+			label_set(90, "foo", "bar", "le", "10")
+			or label_set(100, "foo", "bar", "le", "30")
+			or label_set(300, "foo", "bar", "le", "+Inf")
+			or label_set(200, "tag", "xx", "le", "10")
+			or label_set(300, "tag", "xx", "le", "30")
+		))`
+		r1 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0.325, 0.325, 0.325, 0.325, 0.325, 0.325},
+			Timestamps: timestampsExpected,
+		}
+		r1.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("foo"),
+			Value: []byte("bar"),
+		}}
+		r2 := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0.9166666666666666, 0.9166666666666666, 0.9166666666666666, 0.9166666666666666, 0.9166666666666666, 0.9166666666666666},
+			Timestamps: timestampsExpected,
+		}
+		r2.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("tag"),
+			Value: []byte("xx"),
+		}}
+		resultExpected := []netstorage.Result{r1, r2}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(negative-bucket-count)`, func(t *testing.T) {
 		t.Parallel()
 		q := `histogram_quantile(0.6,
@@ -4566,6 +4705,25 @@ func TestExecSuccess(t *testing.T) {
 		resultExpected := []netstorage.Result{r}
 		f(q, resultExpected)
 	})
+	t.Run(`histogram_fraction(normal-bucket-count)`, func(t *testing.T) {
+		t.Parallel()
+		q := `histogram_fraction(22,35,
+			label_set(0, "foo", "bar", "le", "10")
+			or label_set(100, "foo", "bar", "le", "30")
+			or label_set(300, "foo", "bar", "le", "+Inf")
+		)`
+		r := netstorage.Result{
+			MetricName: metricNameExpected,
+			Values:     []float64{0.1333333333333333, 0.1333333333333333, 0.1333333333333333, 0.1333333333333333, 0.1333333333333333, 0.1333333333333333},
+			Timestamps: timestampsExpected,
+		}
+		r.MetricName.Tags = []storage.Tag{{
+			Key:   []byte("foo"),
+			Value: []byte("bar"),
+		}}
+		resultExpected := []netstorage.Result{r}
+		f(q, resultExpected)
+	})
 	t.Run(`histogram_quantile(normal-bucket-count, boundsLabel)`, func(t *testing.T) {
 		t.Parallel()
 		q := `sort(histogram_quantile(0.2,
@@ -9842,7 +10000,7 @@ func TestExecError(t *testing.T) {
 			Deadline:           searchutil.NewDeadline(time.Now(), time.Minute, ""),
 			RoundDigits:        100,
 		}
-		for i := 0; i < 4; i++ {
+		for range 4 {
 			rv, err := Exec(nil, ec, q, false)
 			if err == nil {
 				t.Fatalf(`expecting non-nil error on %q`, q)

app/vmselect/promql/parse_cache.go

@@ -55,7 +55,7 @@ type parseCache struct {
 
 func newParseCache() *parseCache {
 	pc := new(parseCache)
-	for i := 0; i < parseBucketCount; i++ {
+	for i := range parseBucketCount {
 		pc.buckets[i] = newParseBucket()
 	}
 	return pc
@@ -75,7 +75,7 @@ func (pc *parseCache) get(q string) *parseCacheValue {
 
 func (pc *parseCache) requests() uint64 {
 	var n uint64
-	for i := 0; i < parseBucketCount; i++ {
+	for i := range parseBucketCount {
 		n += pc.buckets[i].requests.Load()
 	}
 	return n
@@ -83,7 +83,7 @@ func (pc *parseCache) requests() uint64 {
 
 func (pc *parseCache) misses() uint64 {
 	var n uint64
-	for i := 0; i < parseBucketCount; i++ {
+	for i := range parseBucketCount {
 		n += pc.buckets[i].misses.Load()
 	}
 	return n
@@ -91,7 +91,7 @@ func (pc *parseCache) misses() uint64 {
 
 func (pc *parseCache) len() uint64 {
 	var n uint64
-	for i := 0; i < parseBucketCount; i++ {
+	for i := range parseBucketCount {
 		n += pc.buckets[i].len()
 	}
 	return n

app/vmselect/promql/parse_cache_test.go

@@ -17,7 +17,7 @@ func testGetParseCacheValue(q string) *parseCacheValue {
 
 func testGenerateQueries(items int) []string {
 	queries := make([]string, items)
-	for i := 0; i < items; i++ {
+	for i := range items {
 		queries[i] = fmt.Sprintf(`node_time_seconds{instance="node%d", job="job%d"}`, i, i)
 	}
 	return queries
@@ -102,7 +102,7 @@ func TestParseCacheBucketOverflow(t *testing.T) {
 	v := testGetParseCacheValue(queries[0])
 
 	// Fill bucket
-	for i := 0; i < parseBucketMaxLen; i++ {
+	for i := range parseBucketMaxLen {
 		b.put(queries[i], v)
 	}
 	expectedLen = uint64(parseBucketMaxLen)

app/vmselect/promql/parse_cache_timing_test.go

@@ -15,7 +15,7 @@ func BenchmarkCachePutNoOverFlow(b *testing.B) {
 	b.ReportAllocs()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			for i := 0; i < items; i++ {
+			for i := range items {
 				pc.put(queries[i], v)
 			}
 		}
@@ -32,14 +32,14 @@ func BenchmarkCacheGetNoOverflow(b *testing.B) {
 	queries := testGenerateQueries(items)
 	v := testGetParseCacheValue(queries[0])
 
-	for i := 0; i < len(queries); i++ {
+	for i := range queries {
 		pc.put(queries[i], v)
 	}
 	b.ResetTimer()
 	b.ReportAllocs()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			for i := 0; i < items; i++ {
+			for i := range items {
 				if v := pc.get(queries[i]); v == nil {
 					b.Errorf("unexpected nil value obtained from cache for query: %s ", queries[i])
 				}
@@ -59,7 +59,7 @@ func BenchmarkCachePutGetNoOverflow(b *testing.B) {
 	b.ReportAllocs()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			for i := 0; i < items; i++ {
+			for i := range items {
 				pc.put(queries[i], v)
 				if res := pc.get(queries[i]); res == nil {
 					b.Errorf("unexpected nil value obtained from cache for query: %s ", queries[i])
@@ -79,7 +79,7 @@ func BenchmarkCachePutOverflow(b *testing.B) {
 	queries := testGenerateQueries(items)
 	v := testGetParseCacheValue(queries[0])
 
-	for i := 0; i < parseCacheMaxLen; i++ {
+	for i := range parseCacheMaxLen {
 		c.put(queries[i], v)
 	}
 
@@ -105,7 +105,7 @@ func BenchmarkCachePutGetOverflow(b *testing.B) {
 	queries := testGenerateQueries(items)
 	v := testGetParseCacheValue(queries[0])
 
-	for i := 0; i < parseCacheMaxLen; i++ {
+	for i := range parseCacheMaxLen {
 		c.put(queries[i], v)
 	}
 
@@ -141,8 +141,8 @@ var testSimpleQueries = []string{
 
 func BenchmarkParsePromQLWithCacheSimple(b *testing.B) {
 	b.ReportAllocs()
-	for i := 0; i < b.N; i++ {
-		for j := 0; j < len(testSimpleQueries); j++ {
+	for range b.N {
+		for j := range testSimpleQueries {
 			_, err := parsePromQLWithCache(testSimpleQueries[j])
 			if err != nil {
 				b.Errorf("unexpected error: %s", err)
@@ -155,7 +155,7 @@ func BenchmarkParsePromQLWithCacheSimpleParallel(b *testing.B) {
 	b.ReportAllocs()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			for i := 0; i < len(testSimpleQueries); i++ {
+			for i := range testSimpleQueries {
 				_, err := parsePromQLWithCache(testSimpleQueries[i])
 				if err != nil {
 					b.Errorf("unexpected error: %s", err)
@@ -210,8 +210,8 @@ var testComplexQueries = []string{
 
 func BenchmarkParsePromQLWithCacheComplex(b *testing.B) {
 	b.ReportAllocs()
-	for i := 0; i < b.N; i++ {
-		for j := 0; j < len(testComplexQueries); j++ {
+	for range b.N {
+		for j := range testComplexQueries {
 			_, err := parsePromQLWithCache(testComplexQueries[j])
 			if err != nil {
 				b.Errorf("unexpected error: %s", err)
@@ -224,7 +224,7 @@ func BenchmarkParsePromQLWithCacheComplexParallel(b *testing.B) {
 	b.ReportAllocs()
 	b.RunParallel(func(pb *testing.PB) {
 		for pb.Next() {
-			for i := 0; i < len(testComplexQueries); i++ {
+			for i := range testComplexQueries {
 				_, err := parsePromQLWithCache(testComplexQueries[i])
 				if err != nil {
 					b.Errorf("unexpected error: %s", err)

app/vmselect/promql/query_stats.go

@@ -13,6 +13,8 @@ type QueryStats struct {
 	ExecutionDuration atomic.Pointer[time.Duration]
 	// SeriesFetched contains the number of series fetched from storage or cache.
 	SeriesFetched atomic.Int64
+	// MemoryUsage contains the estimated memory consumption of the query
+	MemoryUsage atomic.Int64
 
 	at *auth.Token
 
@@ -53,3 +55,17 @@ func (qs *QueryStats) addExecutionTimeMsec(startTime time.Time) {
 	d := time.Since(startTime)
 	qs.ExecutionDuration.Store(&d)
 }
+
+func (qs *QueryStats) addMemoryUsage(memoryUsage int64) {
+	if qs == nil {
+		return
+	}
+	qs.MemoryUsage.Store(memoryUsage)
+}
+
+func (qs *QueryStats) memoryUsage() int64 {
+	if qs == nil {
+		return 0
+	}
+	return qs.MemoryUsage.Load()
+}

app/vmselect/promql/rollup.go

@@ -535,7 +535,10 @@ type rollupFuncArg struct {
 	timestamps []int64
 
 	// Real value preceding values.
-	// Is populated if preceding value is within the rc.LookbackDelta.
+	// Is populated if the preceding sample falls within the rc.LookbackDelta range, or if rc.LookbackDelta is not set.
+	//
+	// It provides an additional check and value for rollup functions such as increase(), changes(),
+	// when the prevValue is NaN due to a gap or a small lookback window.
 	realPrevValue float64
 
 	// Real value which goes after values.
@@ -718,7 +721,11 @@ func (rc *rollupConfig) doInternal(dstValues []float64, tsm *timeseriesMap, valu
 	// Extend dstValues in order to remove mallocs below.
 	dstValues = decimal.ExtendFloat64sCapacity(dstValues, len(rc.Timestamps))
 
-	// Use step as the scrape interval for instant queries (when start == end).
+	// Set maxPrevInterval for subsequent rfa.prevValue calculations in rollupFunc:
+	// For instant queries, use rc.Step directly as maxPrevInterval.
+	// For range queries, rc.Step is typically too small to serve as the lookback window between two rollup points.
+	// Instead, estimate the scrape interval from raw sample timestamps (using the 0.6 quantile of the last 20 intervals)
+	// and slightly inflate the scrape interval to set maxPrevInterval, allowing for some tolerance to jitter.
 	maxPrevInterval := rc.Step
 	if rc.Start < rc.End {
 		scrapeInterval := getScrapeInterval(timestamps, rc.Step)
@@ -734,22 +741,21 @@ func (rc *rollupConfig) doInternal(dstValues []float64, tsm *timeseriesMap, valu
 		}
 	}
 	window := rc.Window
+	// Adjust lookbehind window only if it isn't set explicitly, e.g. rate(foo).
+	// In the case of missing lookbehind window it should be adjusted in order to return non-empty graph
+	// when the window doesn't cover at least two raw samples (this is what most users expect).
+	//
+	// If the user explicitly sets the lookbehind window to some fixed value, e.g. rate(foo[1s]),
+	// then it is expected he knows what he is doing. Do not adjust the lookbehind window then.
+	//
+	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3483
 	if window <= 0 {
 		window = rc.Step
 		if rc.MayAdjustWindow && window < maxPrevInterval {
-			// Adjust lookbehind window only if it isn't set explicitly, e.g. rate(foo).
-			// In the case of missing lookbehind window it should be adjusted in order to return non-empty graph
-			// when the window doesn't cover at least two raw samples (this is what most users expect).
-			//
-			// If the user explicitly sets the lookbehind window to some fixed value, e.g. rate(foo[1s]),
-			// then it is expected he knows what he is doing. Do not adjust the lookbehind window then.
-			//
-			// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3483
 			window = maxPrevInterval
 		}
+		// Artificial window cannot exceed explicit rc.LookbackDelta, see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/784
 		if rc.isDefaultRollup && rc.LookbackDelta > 0 && window > rc.LookbackDelta {
-			// Implicit window exceeds -search.maxStalenessInterval, so limit it to -search.maxStalenessInterval
-			// according to https://github.com/VictoriaMetrics/VictoriaMetrics/issues/784
 			window = rc.LookbackDelta
 		}
 	}
@@ -2112,9 +2118,15 @@ func rollupChanges(rfa *rollupFuncArg) float64 {
 		if len(values) == 0 {
 			return nan
 		}
-		prevValue = values[0]
-		values = values[1:]
-		n++
+		// Assume that the value didn't change during the current gap
+		// if realPrevValue exists.
+		if !math.IsNaN(rfa.realPrevValue) {
+			prevValue = rfa.realPrevValue
+		} else {
+			n++
+			prevValue = values[0]
+			values = values[1:]
+		}
 	}
 	for _, v := range values {
 		if v != prevValue {

app/vmselect/promql/rollup_result_cache.go

@@ -699,7 +699,7 @@ func (mi *rollupResultCacheMetainfo) Unmarshal(src []byte) error {
 	entriesLen := int(encoding.UnmarshalUint32(src))
 	src = src[4:]
 	mi.entries = slicesutil.SetLength(mi.entries, entriesLen)
-	for i := 0; i < entriesLen; i++ {
+	for i := range entriesLen {
 		tail, err := mi.entries[i].Unmarshal(src)
 		if err != nil {
 			return fmt.Errorf("cannot unmarshal entry #%d: %w", i, err)

app/vmselect/promql/rollup_result_cache_test.go

@@ -13,14 +13,14 @@ import (
 
 func TestRollupResultCacheInitStop(t *testing.T) {
 	t.Run("inmemory", func(_ *testing.T) {
-		for i := 0; i < 5; i++ {
+		for range 5 {
 			InitRollupResultCache("")
 			StopRollupResultCache()
 		}
 	})
 	t.Run("file-based", func(_ *testing.T) {
 		cacheFilePath := "test-rollup-result-cache"
-		for i := 0; i < 3; i++ {
+		for range 3 {
 			InitRollupResultCache(cacheFilePath)
 			StopRollupResultCache()
 		}
@@ -248,12 +248,12 @@ func TestRollupResultCache(t *testing.T) {
 	t.Run("big-timeseries", func(t *testing.T) {
 		ResetRollupResultCache()
 		var tss []*timeseries
-		for i := 0; i < 1000; i++ {
+		for i := range 1000 {
 			ts := &timeseries{
 				Timestamps: []int64{1000, 1200, 1400, 1600, 1800, 2000},
 				Values:     []float64{1, 2, 3, 4, 5, 6},
 			}
-			ts.MetricName.MetricGroup = []byte(fmt.Sprintf("metric %d", i))
+			ts.MetricName.MetricGroup = fmt.Appendf(nil, "metric %d", i)
 			tss = append(tss, ts)
 		}
 		rollupResultCacheV.PutSeries(nil, ec, fe, window, tss)