lib/storage: replace tffs loops cache impl with lrucache

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

.github/workflows/build.yml

@@ -22,7 +22,8 @@ on:
       - '!app/vmui/**'
       - '.github/workflows/build.yml'
 
-permissions: {}
+permissions:
+  contents: read
 
 concurrency:
   cancel-in-progress: true
@@ -31,10 +32,7 @@ concurrency:
 jobs:
   build:
     name: ${{ matrix.os }}-${{ matrix.arch }}
-    permissions:
-      contents: read
-    # Runs on dedicated runner with extra resources to increase build speed.
-    runs-on: 'vm-runner'
+    runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
@@ -59,17 +57,15 @@ jobs:
             arch: amd64
           - os: openbsd
             arch: amd64
-          - os: netbsd
-            arch: amd64
           - os: windows
             arch: amd64
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
 
       - name: Setup Go
         id: go
-        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c  # v6.4.0
+        uses: actions/setup-go@v6
         with:
           cache-dependency-path: |
             go.sum

.github/workflows/changelog-linter.yml

@@ -5,15 +5,11 @@ on:
     paths:
       - "docs/victoriametrics/changelog/CHANGELOG.md"
 
-permissions: {}
-
 jobs:
   tip-lint:
-    permissions:
-      contents: read
     runs-on: 'ubuntu-latest'
     steps:
-      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+      - uses: 'actions/checkout@v6'
         with:
           # needed for proper diff
           fetch-depth: 0

.github/workflows/check-commit-signed.yml

@@ -3,16 +3,12 @@ name: check-commit-signed
 on:
   pull_request:
 
-permissions: {}
-
 jobs:
   check-commit-signed:
-    permissions:
-      contents: read
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
         with:
           fetch-depth: 0  # we need full history for commit verification
 

.github/workflows/check-licenses.yml

@@ -6,22 +6,20 @@ on:
   pull_request:
     paths:
       - 'vendor'
-
-permissions: {}
+permissions:
+  contents: read
 
 jobs:
   build:
     name: Build
-    permissions:
-      contents: read
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@master
 
       - name: Setup Go
         id: go
-        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c  # v6.4.0
+        uses: actions/setup-go@v6
         with:
           go-version-file: 'go.mod'
           cache: false
@@ -29,7 +27,7 @@ jobs:
       - run: go version
 
       - name: Cache Go artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae  # v5.0.5
+        uses: actions/cache@v5
         with:
           path: |
             ~/.cache/go-build

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

@@ -18,8 +18,6 @@ concurrency:
   cancel-in-progress: true
   group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
 
-permissions: {}
-
 jobs:
   analyze:
     name: Analyze
@@ -31,18 +29,18 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
 
       - name: Set up Go
         id: go
-        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c  # v6.4.0
+        uses: actions/setup-go@v6
         with:
           cache: false
           go-version-file: 'go.mod'
       - run: go version
 
       - name: Cache Go artifacts
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae  # v5.0.5
+        uses: actions/cache@v5
         with:
           path: |
             ~/.cache/go-build
@@ -52,14 +50,14 @@ jobs:
           restore-keys: go-artifacts-${{ runner.os }}-codeql-analyze-${{ steps.go.outputs.go-version }}-
 
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@7211b7c8077ea37d8641b6271f6a365a22a5fbfa  # v4.36.0
+        uses: github/codeql-action/init@v4.35.1
         with:
           languages: go
 
       - name: Autobuild
-        uses: github/codeql-action/autobuild@7211b7c8077ea37d8641b6271f6a365a22a5fbfa  # v4.36.0
+        uses: github/codeql-action/autobuild@v4.35.1
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@7211b7c8077ea37d8641b6271f6a365a22a5fbfa  # v4.36.0
+        uses: github/codeql-action/analyze@v4.35.1
         with:
           category: 'language:go'

.github/workflows/docs.yaml

@@ -7,30 +7,28 @@ on:
       - 'docs/**'
       - '.github/workflows/docs.yaml'
   workflow_dispatch: {}
-
-permissions: {}
-
+permissions:
+  contents: read  # This is required for actions/checkout and to commit back image update
+  deployments: write
 jobs:
   build:
     name: Build
-    permissions:
-      contents: read
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
         with:
           path: __vm
 
       - name: Checkout private code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
         with:
           repository: VictoriaMetrics/vmdocs
           token: ${{ secrets.VM_BOT_GH_TOKEN }}
           path: __vm-docs
 
       - name: Import GPG key
-        uses: crazy-max/ghaction-import-gpg@2dc316deee8e90f13e1a351ab510b4d5bc0c82cd  # v7.0.0
+        uses: crazy-max/ghaction-import-gpg@v7
         id: import-gpg
         with:
           gpg_private_key: ${{ secrets.VM_BOT_GPG_PRIVATE_KEY }}

.github/workflows/test.yml

@@ -18,7 +18,8 @@ on:
       - 'go.*'
       - '.github/workflows/main.yml'
 
-permissions: {}
+permissions:
+  contents: read
 
 concurrency:
   cancel-in-progress: true
@@ -28,17 +29,14 @@ concurrency:
 jobs:
   lint:
     name: lint
-    permissions:
-      contents: read
-    # Runs on dedicated runner with extra resources since golangci-lint requires extra memory
-    runs-on: 'vm-runner'
+    runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
 
       - name: Setup Go
         id: go
-        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c  # v6.4.0
+        uses: actions/setup-go@v6
         with:
           cache-dependency-path: |
             go.sum
@@ -49,7 +47,7 @@ jobs:
       - run: go version
 
       - name: Cache golangci-lint
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae  # v5.0.5
+        uses: actions/cache@v5
         with:
           path: |
             ~/.cache/golangci-lint
@@ -63,10 +61,7 @@ jobs:
 
   unit:
     name: unit
-    permissions:
-      contents: read
-    # Runs on dedicated runner with extra resources to increase tests speed.
-    runs-on: 'vm-runner'
+    runs-on: ubuntu-latest
 
     strategy:
       matrix:
@@ -77,11 +72,11 @@ jobs:
 
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
 
       - name: Setup Go
         id: go
-        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c  # v6.4.0
+        uses: actions/setup-go@v6
         with:
           cache-dependency-path: |
             go.sum
@@ -95,18 +90,15 @@ jobs:
 
   apptest:
     name: apptest
-    permissions:
-      contents: read
-    # Runs on dedicated runner to isolate app tests from other tests.
     runs-on: apptest
 
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
 
       - name: Setup Go
         id: go
-        uses: actions/setup-go@4a3601121dd01d1626a1e23e37211e3254c1c06c  # v6.4.0
+        uses: actions/setup-go@v6
         with:
           cache-dependency-path: |
             go.sum

.github/workflows/vmui.yml

@@ -16,7 +16,11 @@ on:
       - 'app/vmui/packages/vmui/**'
       - '.github/workflows/vmui.yml'
 
-permissions: {}
+permissions:
+  contents: read
+  packages: read
+  pull-requests: read
+  checks: write
 
 concurrency:
   cancel-in-progress: true
@@ -25,18 +29,14 @@ concurrency:
 jobs:
   vmui-checks:
     name: VMUI Checks (lint, test, typecheck)
-    permissions:
-      checks: write
-      contents: read
-      pull-requests: read
     runs-on: ubuntu-latest
     steps:
       - name: Code checkout
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
+        uses: actions/checkout@v6
 
       - name: Cache node_modules
         id: cache
-        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae  # v5.0.5
+        uses: actions/cache@v5
         with:
           path: app/vmui/packages/vmui/node_modules
           key: vmui-deps-${{ runner.os }}-${{ hashFiles('app/vmui/packages/vmui/package-lock.json', 'app/vmui/Dockerfile-build') }}
@@ -69,7 +69,7 @@ jobs:
           VMUI_SKIP_INSTALL: true
 
       - name: Annotate Code Linting Results
-        uses: ataylorme/eslint-annotate-action@d57a1193d4c59cbfbf3f86c271f42612f9dbd9e9  # 3.0.0
+        uses: ataylorme/eslint-annotate-action@v3
         with:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           report-json: app/vmui/packages/vmui/vmui-lint-report.json

.golangci.yml

@@ -3,14 +3,27 @@ linters:
   settings:
     errcheck:
       exclude-functions:
+        - fmt.Fprintf
+        - fmt.Fprint
         - (net/http.ResponseWriter).Write
   exclusions:
     generated: lax
     presets:
+      - common-false-positives
+      - legacy
       - std-error-handling
     rules:
       - linters:
           - staticcheck
         text: 'SA(4003|1019|5011):'
     paths:
-      - ^app/vmui/
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

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.12.2
+GOLANGCI_LINT_VERSION := 2.9.0
 
 .PHONY: $(MAKECMDGOALS)
 
@@ -485,8 +485,8 @@ apptest-legacy: victoria-metrics-race vmbackup-race vmrestore-race
 		curl --output-dir /tmp -LO $${URL}/$${VMSINGLE} && tar xzf /tmp/$${VMSINGLE} -C $${DIR} && \
 		curl --output-dir /tmp -LO $${URL}/$${VMCLUSTER} && tar xzf /tmp/$${VMCLUSTER} -C $${DIR} \
 	); \
-	VMSINGLE_V1_132_0_PATH=$${DIR}/victoria-metrics-prod \
-	VMSTORAGE_V1_132_0_PATH=$${DIR}/vmstorage-prod \
+	VM_LEGACY_VMSINGLE_PATH=$${DIR}/victoria-metrics-prod \
+	VM_LEGACY_VMSTORAGE_PATH=$${DIR}/vmstorage-prod \
 	go test ./apptest/tests -run="^TestLegacySingle.*"
 
 benchmark:
@@ -527,7 +527,7 @@ golangci-lint: install-golangci-lint
 	golangci-lint run --build-tags 'synctest'
 
 install-golangci-lint:
-	which golangci-lint && (golangci-lint --version | grep -q $(GOLANGCI_LINT_VERSION)) || curl -sSfL https://golangci-lint.run/install.sh | sh -s -- -b $(shell go env GOPATH)/bin v$(GOLANGCI_LINT_VERSION)
+	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)
 
 remove-golangci-lint:
 	rm -rf `which golangci-lint`
@@ -535,15 +535,6 @@ remove-golangci-lint:
 govulncheck: install-govulncheck
 	govulncheck ./...
 
-govulncheck-docker:
-	docker run -w $(PWD) -v $(PWD):$(PWD) \
-		-v govulncheck-gomod-cache:/root/go/pkg/mod \
-		-v govulncheck-gobuild-cache:/root/.cache/go-build \
-		-v govulncheck-go-bin:/root/go/bin \
-		--env="GOCACHE=/root/.cache/go-build" \
-		--env="GOMODCACHE=/root/go/pkg/mod" \
-		"$(GO_BUILDER_IMAGE)" /bin/sh -c "which govulncheck || go install golang.org/x/vuln/cmd/govulncheck@latest && govulncheck ./..."
-
 install-govulncheck:
 	which govulncheck || go install golang.org/x/vuln/cmd/govulncheck@latest
 

SECURITY.md

@@ -1,4 +1,42 @@
 # Security Policy
 
-You can find out about our security policy and VictoriaMetrics version support on the [security page](https://docs.victoriametrics.com/victoriametrics/#security) in the documentation.
+## Supported Versions
 
+The following versions of VictoriaMetrics receive regular security fixes:
+
+| Version                                                                        | Supported          |
+|--------------------------------------------------------------------------------|--------------------|
+| [Latest release](https://docs.victoriametrics.com/victoriametrics/changelog/)  | :white_check_mark: |
+| [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-releases/) | :white_check_mark: |
+| other releases                                                                 | :x:                |
+
+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/victoria-metrics/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/procutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/pushmetrics"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 )
 
 var (
@@ -29,26 +30,23 @@ 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")
+	minScrapeInterval = flag.Duration("dedup.minScrapeInterval", 0, "Leave only the last sample in every time series per each discrete interval "+
+		"equal to -dedup.minScrapeInterval > 0. See also -streamAggr.dedupInterval and https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
 	dryRun = flag.Bool("dryRun", false, "Whether to check config files without running VictoriaMetrics. The following config files are checked: "+
 		"-promscrape.config, -relabelConfig and -streamAggr.config. Unknown config entries aren't allowed in -promscrape.config by default. "+
 		"This can be changed with -promscrape.config.strictParse=false command-line flag")
+	inmemoryDataFlushInterval = flag.Duration("inmemoryDataFlushInterval", 5*time.Second, "The interval for guaranteed saving of in-memory data to disk. "+
+		"The saved data survives unclean shutdowns such as OOM crash, hardware reset, SIGKILL, etc. "+
+		"Bigger intervals may help increase the lifetime of flash storage with limited write cycles (e.g. Raspberry PI). "+
+		"Smaller intervals increase disk IO load. Minimum supported value is 1s")
 	maxIngestionRate = flag.Int("maxIngestionRate", 0, "The maximum number of samples vmsingle can receive per second. Data ingestion is paused when the limit is exceeded. "+
 		"By default there are no limits on samples ingestion rate.")
-	vmselectMaxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
-		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
-		"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
-	vmselectMaxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests "+
-		"limit is reached; see also -search.maxQueryDuration")
+	finalDedupScheduleInterval = flag.Duration("storage.finalDedupScheduleCheckInterval", time.Hour, "The interval for checking when final deduplication process should be started."+
+		"Storage unconditionally adds 25% jitter to the interval value on each check evaluation."+
+		" Changing the interval to the bigger values may delay downsampling, deduplication for historical data."+
+		" See also https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#deduplication")
 )
 
-func getDefaultMaxConcurrentRequests() int {
-	// A single request can saturate all the CPU cores, so there is no sense
-	// in allowing higher number of concurrent requests - they will just contend
-	// for unavailable CPU time.
-	n := min(cgroup.AvailableCPUs()*2, 16)
-	return n
-}
-
 func main() {
 	// VictoriaMetrics is optimized for reduced memory allocations,
 	// so it can run with the reduced GOGC in order to reduce the used memory,
@@ -89,8 +87,14 @@ func main() {
 	}
 	logger.Infof("starting VictoriaMetrics at %q...", listenAddrs)
 	startTime := time.Now()
-	vmstorage.Init(*vmselectMaxConcurrentRequests, promql.ResetRollupResultCacheIfNeeded)
-	vmselect.Init(*vmselectMaxConcurrentRequests, *vmselectMaxQueueDuration)
+	storage.SetDedupInterval(*minScrapeInterval)
+	storage.SetDataFlushInterval(*inmemoryDataFlushInterval)
+	if *finalDedupScheduleInterval < time.Hour {
+		logger.Fatalf("-dedup.finalDedupScheduleCheckInterval cannot be smaller than 1 hour; got %s", *finalDedupScheduleInterval)
+	}
+	storage.SetFinalDedupScheduleInterval(*finalDedupScheduleInterval)
+	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
+	vmselect.Init()
 	vminsertcommon.StartIngestionRateLimiter(*maxIngestionRate)
 	vminsert.Init()
 

app/victoria-metrics/self_scraper.go

@@ -93,7 +93,7 @@ func selfScraper(scrapeInterval time.Duration) {
 				mr.Value = r.Value
 			}
 		}
-		if err := vmstorage.VMInsertAPI.WriteRows(mrs); err != nil {
+		if err := vmstorage.AddRows(mrs); err != nil {
 			logger.Errorf("cannot store self-scraped metrics: %s", err)
 		}
 		if len(metadataRows.Rows) > 0 {
@@ -105,7 +105,7 @@ func selfScraper(scrapeInterval time.Duration) {
 					Type:             mm.Type,
 				})
 			}
-			if err := vmstorage.VMInsertAPI.WriteMetadata(mms); err != nil {
+			if err := vmstorage.AddMetadataRows(mms); err != nil {
 				logger.Errorf("cannot store self-scraped metrics metadata: %s", err)
 			}
 		}

app/victoria-metrics/test/prom_writer.go

@@ -10,7 +10,7 @@ import (
 func Compress(wr WriteRequest) []byte {
 	data, err := wr.Marshal()
 	if err != nil {
-		panic(fmt.Errorf("BUG: cannot compress WriteRequest: %w", err))
+		panic(fmt.Errorf("BUG: cannot compress WriteRequest: %s", err))
 	}
 	return snappy.Encode(nil, data)
 }

app/vmagent/main.go

@@ -83,9 +83,6 @@ var (
 	maxLabelsPerTimeseries = flag.Int("maxLabelsPerTimeseries", 0, "The maximum number of labels per time series to be accepted. Series with superfluous labels are ignored. In this case the vm_rows_ignored_total{reason=\"too_many_labels\"} metric at /metrics page is incremented")
 	maxLabelNameLen        = flag.Int("maxLabelNameLen", 0, "The maximum length of label names in the accepted time series. Series with longer label name are ignored. In this case the vm_rows_ignored_total{reason=\"too_long_label_name\"} metric at /metrics page is incremented")
 	maxLabelValueLen       = flag.Int("maxLabelValueLen", 0, "The maximum length of label values in the accepted time series. Series with longer label value are ignored. In this case the vm_rows_ignored_total{reason=\"too_long_label_value\"} metric at /metrics page is incremented")
-
-	enableMultitenancyViaHeaders = flag.Bool("enableMultitenancyViaHeaders", false, "Enables multitenancy via HTTP headers. "+
-		"See https://docs.victoriametrics.com/victoriametrics/vmagent/#multitenancy")
 )
 
 var (
@@ -118,7 +115,6 @@ func main() {
 	remotewrite.InitSecretFlags()
 	buildinfo.Init()
 	logger.Init()
-	opentelemetry.Init()
 	timeserieslimits.Init(*maxLabelsPerTimeseries, *maxLabelNameLen, *maxLabelValueLen)
 
 	if promscrape.IsDryRun() {
@@ -220,7 +216,7 @@ func getOpenTSDBHTTPInsertHandler() func(req *http.Request) error {
 	}
 	return func(req *http.Request) error {
 		path := strings.ReplaceAll(req.URL.Path, "//", "/")
-		at, err := getAuthTokenFromPath(path, req.Header)
+		at, err := getAuthTokenFromPath(path)
 		if err != nil {
 			return fmt.Errorf("cannot obtain auth token from path %q: %w", path, err)
 		}
@@ -228,15 +224,8 @@ func getOpenTSDBHTTPInsertHandler() func(req *http.Request) error {
 	}
 }
 
-func parsePath(path string, header http.Header) (*httpserver.Path, error) {
-	if *enableMultitenancyViaHeaders {
-		return httpserver.ParsePathAndHeaders(path, header)
-	}
-	return httpserver.ParsePath(path)
-}
-
-func getAuthTokenFromPath(path string, header http.Header) (*auth.Token, error) {
-	p, err := parsePath(path, header)
+func getAuthTokenFromPath(path string) (*auth.Token, error) {
+	p, err := httpserver.ParsePath(path)
 	if err != nil {
 		return nil, fmt.Errorf("cannot parse multitenant path: %w", err)
 	}
@@ -570,15 +559,14 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 }
 
 func processMultitenantRequest(w http.ResponseWriter, r *http.Request, path string) bool {
-	p, err := parsePath(path, r.Header)
+	p, err := httpserver.ParsePath(path)
 	if err != nil {
 		// Cannot parse multitenant path. Skip it - probably it will be parsed later.
 		return false
 	}
 	if p.Prefix != "insert" {
-		// processMultitenantRequest is called for all unmatched path variants,
-		// but we should try parsing only /insert prefixed to avoid catching all possible paths.
-		return false
+		httpserver.Errorf(w, r, `unsupported multitenant prefix: %q; expected "insert"`, p.Prefix)
+		return true
 	}
 	at, err := auth.NewTokenPossibleMultitenant(p.AuthToken)
 	if err != nil {

app/vmagent/opentelemetry/request_handler.go

@@ -25,11 +25,6 @@ var (
 	rowsPerInsert          = metrics.NewHistogram(`vmagent_rows_per_insert{type="opentelemetry"}`)
 )
 
-// Init must be called after flag.Parse and before using the opentelemetry package.
-func Init() {
-	stream.InitDecodeOptions()
-}
-
 // InsertHandlerForReader processes metrics from given reader.
 func InsertHandlerForReader(at *auth.Token, r io.Reader, encoding string) error {
 	return stream.ParseStream(r, encoding, nil, func(tss []prompb.TimeSeries, mms []prompb.MetricMetadata) error {
@@ -82,6 +77,16 @@ func insertRows(at *auth.Token, tss []prompb.TimeSeries, mms []prompb.MetricMeta
 
 	var metadataTotal int
 	if prommetadata.IsEnabled() {
+		var accountID, projectID uint32
+		if at != nil {
+			accountID = at.AccountID
+			projectID = at.ProjectID
+			for i := range mms {
+				mm := &mms[i]
+				mm.AccountID = accountID
+				mm.ProjectID = projectID
+			}
+		}
 		ctx.WriteRequest.Metadata = mms
 		metadataTotal = len(mms)
 	}

app/vmagent/prometheusimport/request_handler.go

@@ -75,6 +75,11 @@ func insertRows(at *auth.Token, rows []prometheus.Row, mms []prometheus.Metadata
 			Samples: samples[len(samples)-1:],
 		})
 	}
+	var accountID, projectID uint32
+	if at != nil {
+		accountID = at.AccountID
+		projectID = at.ProjectID
+	}
 	for i := range mms {
 		mm := &mms[i]
 		mmsDst = append(mmsDst, prompb.MetricMetadata{
@@ -83,6 +88,8 @@ func insertRows(at *auth.Token, rows []prometheus.Row, mms []prometheus.Metadata
 			Type:             mm.Type,
 			// there is no unit in Prometheus exposition formats
 
+			AccountID: accountID,
+			ProjectID: projectID,
 		})
 	}
 	ctx.WriteRequest.Timeseries = tssDst

app/vmagent/promremotewrite/request_handler.go

@@ -72,6 +72,11 @@ func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, mms []prompb.Met
 
 	var metadataTotal int
 	if prommetadata.IsEnabled() {
+		var accountID, projectID uint32
+		if at != nil {
+			accountID = at.AccountID
+			projectID = at.ProjectID
+		}
 		for i := range mms {
 			mm := &mms[i]
 			mmsDst = append(mmsDst, prompb.MetricMetadata{
@@ -80,8 +85,8 @@ func insertRows(at *auth.Token, timeseries []prompb.TimeSeries, mms []prompb.Met
 				Type:             mm.Type,
 				Unit:             mm.Unit,
 
-				AccountID: mm.AccountID,
-				ProjectID: mm.ProjectID,
+				AccountID: accountID,
+				ProjectID: projectID,
 			})
 		}
 		ctx.WriteRequest.Metadata = mmsDst

app/vmagent/remotewrite/client.go

@@ -2,7 +2,6 @@ package remotewrite
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -60,8 +59,6 @@ var (
 		"Multiple headers must be delimited by '^^': -remoteWrite.headers='header1:value1^^header2:value2'")
 
 	basicAuthUsername     = flagutil.NewArrayString("remoteWrite.basicAuth.username", "Optional basic auth username to use for the corresponding -remoteWrite.url")
-	basicAuthUsernameFile = flagutil.NewArrayString("remoteWrite.basicAuth.usernameFile", "Optional path to basic auth username to use for the corresponding -remoteWrite.url. "+
-		"The file is re-read every second")
 	basicAuthPassword     = flagutil.NewArrayString("remoteWrite.basicAuth.password", "Optional basic auth password to use for the corresponding -remoteWrite.url")
 	basicAuthPasswordFile = flagutil.NewArrayString("remoteWrite.basicAuth.passwordFile", "Optional path to basic auth password to use for the corresponding -remoteWrite.url. "+
 		"The file is re-read every second")
@@ -226,14 +223,12 @@ func getAuthConfig(argIdx int) (*promauth.Config, error) {
 		hdrs = strings.Split(headersValue, "^^")
 	}
 	username := basicAuthUsername.GetOptionalArg(argIdx)
-	usernameFile := basicAuthUsernameFile.GetOptionalArg(argIdx)
 	password := basicAuthPassword.GetOptionalArg(argIdx)
 	passwordFile := basicAuthPasswordFile.GetOptionalArg(argIdx)
 	var basicAuthCfg *promauth.BasicAuthConfig
-	if username != "" || usernameFile != "" || password != "" || passwordFile != "" {
+	if username != "" || password != "" || passwordFile != "" {
 		basicAuthCfg = &promauth.BasicAuthConfig{
 			Username:     username,
-			UsernameFile: usernameFile,
 			Password:     promauth.NewSecret(password),
 			PasswordFile: passwordFile,
 		}
@@ -311,6 +306,11 @@ func (c *client) runWorker() {
 		if !ok {
 			return
 		}
+		if len(block) == 0 {
+			// skip empty data blocks from sending
+			// see https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6241
+			continue
+		}
 		go func() {
 			startTime := time.Now()
 			ch <- c.sendBlock(block)
@@ -326,20 +326,15 @@ func (c *client) runWorker() {
 			c.fq.MustWriteBlockIgnoreDisabledPQ(block)
 			return
 		case <-c.stopCh:
-			// c must be stopped. Wait up to 5 seconds for the in-flight request to complete.
-			// If it succeeds, drain the remaining in-memory queue before returning.
-			stopCtx, cancel := context.WithTimeout(context.Background(), time.Second*5)
-			defer cancel()
-
+			// c must be stopped. Wait for a while in the hope the block will be sent.
+			graceDuration := 5 * time.Second
 			select {
 			case ok := <-ch:
 				if !ok {
 					// Return unsent block to the queue.
 					c.fq.MustWriteBlockIgnoreDisabledPQ(block)
-				} else {
-					c.drainInMemoryQueue(stopCtx, block[:0])
 				}
-			case <-stopCtx.Done():
+			case <-time.After(graceDuration):
 				// Return unsent block to the queue.
 				c.fq.MustWriteBlockIgnoreDisabledPQ(block)
 			}
@@ -471,7 +466,7 @@ again:
 				goto again
 			}
 
-			logger.Warnf("failed to repack zstd block (%d bytes) to snappy: %s; The block will be rejected. "+
+			logger.Warnf("failed to repack zstd block (%s bytes) to snappy: %s; The block will be rejected. "+
 				"Possible cause: ungraceful shutdown leading to persisted queue corruption.",
 				zstdBlockLen, err)
 		}
@@ -509,32 +504,6 @@ again:
 	goto again
 }
 
-func (c *client) drainInMemoryQueue(stopCtx context.Context, block []byte) {
-	var ok bool
-	for {
-		select {
-		case <-stopCtx.Done():
-			return
-		default:
-		}
-
-		block, ok = c.fq.MustReadInMemoryBlock(block[:0])
-		if !ok {
-			// The in memory queue has already been drained,
-			// or persisted queue is being used.
-			// In this case it is guaranteed that fq will be empty
-			return
-		}
-
-		// at this stage c.stopCh should be closed
-		// so sendBlock function should not perform retries
-		if ok := c.sendBlock(block); !ok {
-			c.fq.MustWriteBlockIgnoreDisabledPQ(block)
-			return
-		}
-	}
-}
-
 var remoteWriteRejectedLogger = logger.WithThrottler("remoteWriteRejected", 5*time.Second)
 var remoteWriteRetryLogger = logger.WithThrottler("remoteWriteRetry", 5*time.Second)
 

app/vmagent/remotewrite/client_test.go

@@ -9,7 +9,6 @@ import (
 	"github.com/golang/snappy"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 )
 
 func TestParseRetryAfterHeader(t *testing.T) {
@@ -37,40 +36,6 @@ func TestParseRetryAfterHeader(t *testing.T) {
 	f(time.Now().Add(10*time.Second).Format("Mon, 02 Jan 2006 15:04:05 FAKETZ"), 0)
 }
 
-func TestInitSecretFlags(t *testing.T) {
-	showRemoteWriteURLOrig := *showRemoteWriteURL
-	defer func() {
-		*showRemoteWriteURL = showRemoteWriteURLOrig
-		flagutil.UnregisterAllSecretFlags()
-	}()
-
-	flagutil.UnregisterAllSecretFlags()
-	*showRemoteWriteURL = false
-	InitSecretFlags()
-	if !flagutil.IsSecretFlag("remotewrite.url") {
-		t.Fatalf("expecting remoteWrite.url to be secret")
-	}
-	if !flagutil.IsSecretFlag("remotewrite.headers") {
-		t.Fatalf("expecting remoteWrite.headers to be secret")
-	}
-	if !flagutil.IsSecretFlag("remotewrite.proxyurl") {
-		t.Fatalf("expecting remoteWrite.proxyURL to be secret")
-	}
-
-	flagutil.UnregisterAllSecretFlags()
-	*showRemoteWriteURL = true
-	InitSecretFlags()
-	if flagutil.IsSecretFlag("remotewrite.url") {
-		t.Fatalf("remoteWrite.url must remain visible when -remoteWrite.showURL is set")
-	}
-	if !flagutil.IsSecretFlag("remotewrite.headers") {
-		t.Fatalf("expecting remoteWrite.headers to remain secret")
-	}
-	if !flagutil.IsSecretFlag("remotewrite.proxyurl") {
-		t.Fatalf("expecting remoteWrite.proxyURL to remain secret")
-	}
-}
-
 func TestRepackBlockFromZstdToSnappy(t *testing.T) {
 	expectedPlainBlock := []byte(`foobar`)
 

app/vmagent/remotewrite/pendingseries.go

@@ -211,9 +211,6 @@ func (wr *writeRequest) copyMetadata(dst, src *prompb.MetricMetadata) {
 	dst.Type = src.Type
 	dst.Unit = src.Unit
 
-	dst.AccountID = src.AccountID
-	dst.ProjectID = src.ProjectID
-
 	// Pre-allocate memory for all string fields.
 	neededBufLen := len(src.MetricFamilyName) + len(src.Help)
 	bufLen := len(wr.metadatabuf)

app/vmagent/remotewrite/remotewrite.go

@@ -79,8 +79,7 @@ var (
 		"writing them to remote storage. "+
 		"Examples: -remoteWrite.roundDigits=2 would round 1.236 to 1.24, while -remoteWrite.roundDigits=-1 would round 126.78 to 130. "+
 		"By default, digits rounding is disabled. Set it to 100 for disabling it for a particular remote storage. "+
-		"This option may be used for improving data compression for the stored metrics. "+
-		"See also -remoteWrite.significantFigures")
+		"This option may be used for improving data compression for the stored metrics")
 	sortLabels = flag.Bool("sortLabels", false, `Whether to sort labels for incoming samples before writing them to all the configured remote storage systems. `+
 		`This may be needed for reducing memory usage at remote storage when the order of labels in incoming samples is random. `+
 		`For example, if m{k1="v1",k2="v2"} may be sent as m{k2="v2",k1="v1"}`+
@@ -152,10 +151,6 @@ func InitSecretFlags() {
 		// remoteWrite.url can contain authentication codes, so hide it at `/metrics` output.
 		flagutil.RegisterSecretFlag("remoteWrite.url")
 	}
-	// remoteWrite.proxyURL can contain authentication codes.
-	flagutil.RegisterSecretFlag("remoteWrite.proxyURL")
-	// remoteWrite.headers can contain auth headers such as Authorization and API keys.
-	flagutil.RegisterSecretFlag("remoteWrite.headers")
 }
 
 var (
@@ -172,18 +167,6 @@ func Init() {
 	if len(*remoteWriteURLs) == 0 {
 		logger.Fatalf("at least one `-remoteWrite.url` command-line flag must be set")
 	}
-	if *shardByURL && len(*disableOnDiskQueue) > 1 {
-		disableOnDiskQueues := *disableOnDiskQueue
-
-		firstValue := disableOnDiskQueues[0]
-		for _, v := range disableOnDiskQueues[1:] {
-			if firstValue != v {
-				logger.Fatalf("all -remoteWrite.url targets must have the same -remoteWrite.disableOnDiskQueue setting when -remoteWrite.shardByURL is enabled; " +
-					"either enable or disable -remoteWrite.disableOnDiskQueue for all targets")
-			}
-		}
-	}
-
 	if limit := getMaxHourlySeries(); limit > 0 {
 		hourlySeriesLimiter = bloomfilter.NewLimiter(limit, time.Hour)
 		_ = metrics.NewGauge(`vmagent_hourly_series_limit_max_series`, func() float64 {
@@ -302,7 +285,6 @@ func initRemoteWriteCtxs(urls []string) {
 		rwctxs[i] = newRemoteWriteCtx(i, remoteWriteURL, sanitizedURL)
 		rwctxIdx[i] = i
 	}
-	fs.RegisterPathFsMetrics(*tmpDataPath)
 
 	if *shardByURL {
 		consistentHashNodes := make([]string, 0, len(urls))
@@ -416,7 +398,7 @@ func tryPush(at *auth.Token, wr *prompb.WriteRequest, forceDropSamplesOnFailure
 
 	// Push metadata separately from time series, since it doesn't need sharding,
 	// relabeling, stream aggregation, deduplication, etc.
-	if !tryPushMetadataToRemoteStorages(at, rwctxs, mms, forceDropSamplesOnFailure) {
+	if !tryPushMetadataToRemoteStorages(rwctxs, mms, forceDropSamplesOnFailure) {
 		return false
 	}
 
@@ -516,9 +498,7 @@ func tryPush(at *auth.Token, wr *prompb.WriteRequest, forceDropSamplesOnFailure
 //
 // calculateHealthyRwctxIdx will rely on the order of rwctx to be in ascending order.
 func getEligibleRemoteWriteCtxs(tss []prompb.TimeSeries, forceDropSamplesOnFailure bool) ([]*remoteWriteCtx, bool) {
-	// When -remoteWrite.shardByURL=true always use all configured remote writes to preserve stable metrics distribution across shards.
-	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10507
-	if !disableOnDiskQueueAny || *shardByURL {
+	if !disableOnDiskQueueAny {
 		return rwctxsGlobal, true
 	}
 
@@ -533,6 +513,12 @@ func getEligibleRemoteWriteCtxs(tss []prompb.TimeSeries, forceDropSamplesOnFailu
 				return nil, false
 			}
 			rowsCount := getRowsCount(tss)
+			if *shardByURL {
+				// Todo: When shardByURL is enabled, the following metrics won't be 100% accurate. Because vmagent don't know
+				// which rwctx should data be pushed to yet. Let's consider the hashing algorithm fair and will distribute
+				// data to all rwctxs evenly.
+				rowsCount = rowsCount / len(rwctxsGlobal)
+			}
 			rwctx.rowsDroppedOnPushFailure.Add(rowsCount)
 		}
 	}
@@ -550,18 +536,11 @@ func pushTimeSeriesToRemoteStoragesTrackDropped(tss []prompb.TimeSeries) {
 	}
 }
 
-func tryPushMetadataToRemoteStorages(at *auth.Token, rwctxs []*remoteWriteCtx, mms []prompb.MetricMetadata, forceDropSamplesOnFailure bool) bool {
+func tryPushMetadataToRemoteStorages(rwctxs []*remoteWriteCtx, mms []prompb.MetricMetadata, forceDropSamplesOnFailure bool) bool {
 	if len(mms) == 0 {
 		// Nothing to push
 		return true
 	}
-	if at != nil {
-		for idx := range mms {
-			mm := &mms[idx]
-			mm.AccountID = at.AccountID
-			mm.ProjectID = at.ProjectID
-		}
-	}
 	// Do not shard metadata even if -remoteWrite.shardByURL is set, just replicate it among rwctxs.
 	// Since metadata is usually small and there is no guarantee that metadata can be sent to
 	// the same remote storage with the corresponding metrics.
@@ -712,7 +691,7 @@ func shardAmountRemoteWriteCtx(tssBlock []prompb.TimeSeries, shards [][]prompb.T
 			}
 			tmpLabels.Labels = hashLabels
 		}
-		h := getLabelsHashForShard(hashLabels)
+		h := getLabelsHash(hashLabels)
 
 		// Get the rwctxIdx through consistent hashing and then map it to the index in shards.
 		// The rwctxIdx is not always equal to the shardIdx, for example, when some rwctx are not available.
@@ -803,28 +782,11 @@ var (
 	dailySeriesLimitRowsDropped  = metrics.NewCounter(`vmagent_daily_series_limit_rows_dropped_total`)
 )
 
-// getLabelsHashForShard is a separate function from getLabelsHash because
-// it omits the '=' separator between label name and value for backward compatibility.
-// Changing it would re-shard all series across remoteWrite targets.
-func getLabelsHashForShard(labels []prompb.Label) uint64 {
-	bb := labelsHashBufPool.Get()
-	b := bb.B[:0]
-	for _, label := range labels {
-		b = append(b, label.Name...)
-		b = append(b, label.Value...)
-	}
-	h := xxhash.Sum64(b)
-	bb.B = b
-	labelsHashBufPool.Put(bb)
-	return h
-}
-
 func getLabelsHash(labels []prompb.Label) uint64 {
 	bb := labelsHashBufPool.Get()
 	b := bb.B[:0]
 	for _, label := range labels {
 		b = append(b, label.Name...)
-		b = append(b, '=')
 		b = append(b, label.Value...)
 	}
 	h := xxhash.Sum64(b)

app/vmagent/remotewrite/remotewrite_test.go

@@ -25,7 +25,7 @@ func TestGetLabelsHash_Distribution(t *testing.T) {
 		t.Helper()
 
 		// Distribute itemsCount hashes returned by getLabelsHash() across bucketsCount buckets.
-		itemsCount := 10_000 * bucketsCount
+		itemsCount := 1_000 * bucketsCount
 		m := make([]int, bucketsCount)
 		var labels []prompb.Label
 		for i := range itemsCount {
@@ -44,12 +44,10 @@ func TestGetLabelsHash_Distribution(t *testing.T) {
 		}
 
 		// Verify that the distribution is even
-		expectedItemsPerBucket := float64(itemsCount / bucketsCount)
-		allowedDeviation := math.Round(float64(expectedItemsPerBucket) * 0.04)
+		expectedItemsPerBucket := itemsCount / bucketsCount
 		for _, n := range m {
-			if math.Abs(expectedItemsPerBucket-float64(n)) > allowedDeviation {
-				t.Fatalf("unexpected items in the bucket for %d buckets; got %d; want in range [%.0f, %.0f]",
-					bucketsCount, n, expectedItemsPerBucket-allowedDeviation, expectedItemsPerBucket+allowedDeviation)
+			if math.Abs(1-float64(n)/float64(expectedItemsPerBucket)) > 0.04 {
+				t.Fatalf("unexpected items in the bucket for %d buckets; got %d; want around %d", bucketsCount, n, expectedItemsPerBucket)
 			}
 		}
 	}

app/vmalert-tool/unittest/input.go

@@ -52,7 +52,7 @@ func writeInputSeries(input []series, interval *promutil.Duration, startStamp ti
 	data := testutil.Compress(r)
 	// write input series to vm
 	httpWrite(dst, bytes.NewBuffer(data))
-	vmstorage.DebugFlush()
+	vmstorage.Storage.DebugFlush()
 	return nil
 }
 
@@ -61,15 +61,15 @@ func parseInputSeries(input []series, interval *promutil.Duration, startStamp ti
 	for _, data := range input {
 		expr, err := metricsql.Parse(data.Series)
 		if err != nil {
-			return res, fmt.Errorf("failed to parse series %s: %w", data.Series, err)
+			return res, fmt.Errorf("failed to parse series %s: %v", data.Series, err)
 		}
 		promvals, err := parseInputValue(data.Values, true)
 		if err != nil {
-			return res, fmt.Errorf("failed to parse input series value %s: %w", data.Values, err)
+			return res, fmt.Errorf("failed to parse input series value %s: %v", data.Values, err)
 		}
 		metricExpr, ok := expr.(*metricsql.MetricExpr)
 		if !ok || len(metricExpr.LabelFilterss) != 1 {
-			return res, fmt.Errorf("got invalid input series %s: %w", data.Series, err)
+			return res, fmt.Errorf("got invalid input series %s: %v", data.Series, err)
 		}
 		samples := make([]testutil.Sample, 0, len(promvals))
 		ts := startStamp

app/vmalert-tool/unittest/recording.go

@@ -53,13 +53,13 @@ Outer:
 			if s.Labels != "" {
 				metricsqlExpr, err := metricsql.Parse(s.Labels)
 				if err != nil {
-					checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s, err: %w", mt.Expr,
+					checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s, err: %v", mt.Expr,
 						mt.EvalTime.Duration().String(), fmt.Errorf("failed to parse labels %q: %w", s.Labels, err)))
 					continue Outer
 				}
 				metricsqlMetricExpr, ok := metricsqlExpr.(*metricsql.MetricExpr)
 				if !ok || len(metricsqlMetricExpr.LabelFilterss) > 1 {
-					checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s, err: %w", mt.Expr,
+					checkErrs = append(checkErrs, fmt.Errorf("\n    expr: %q, time: %s, err: %v", mt.Expr,
 						mt.EvalTime.Duration().String(), fmt.Errorf("got invalid exp_samples: %q", s.Labels)))
 					continue Outer
 				}

app/vmalert-tool/unittest/unittest.go

@@ -61,7 +61,7 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	}
 	eu, err := url.Parse(externalURL)
 	if err != nil {
-		logger.Fatalf("failed to parse external URL: %s", err)
+		logger.Fatalf("failed to parse external URL: %w", err)
 	}
 	if err := templates.Load([]string{}, *eu); err != nil {
 		logger.Fatalf("failed to load template: %v", err)
@@ -108,9 +108,7 @@ func UnitTest(files []string, disableGroupLabel bool, externalLabels []string, e
 	storagePath = tmpFolder
 	processFlags()
 	vminsert.Init()
-	const maxConcurrentRequests = 4
-	maxQueueDuration := 5 * time.Second
-	vmselect.Init(maxConcurrentRequests, maxQueueDuration)
+	vmselect.Init()
 	// storagePath will be created again when closing vmselect, so remove it again.
 	defer fs.MustRemoveDir(storagePath)
 	defer vminsert.Stop()
@@ -281,8 +279,7 @@ func processFlags() {
 }
 
 func setUp() {
-	const maxConcurrentRequests = 4
-	vmstorage.Init(maxConcurrentRequests, promql.ResetRollupResultCacheIfNeeded)
+	vmstorage.Init(promql.ResetRollupResultCacheIfNeeded)
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 	readyCheckFunc := func() bool {
@@ -329,11 +326,11 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 
 	q, err := datasource.Init(nil)
 	if err != nil {
-		return []error{fmt.Errorf("failed to init datasource: %w", err)}
+		return []error{fmt.Errorf("failed to init datasource: %v", err)}
 	}
 	rw, err := remotewrite.NewDebugClient()
 	if err != nil {
-		return []error{fmt.Errorf("failed to init wr: %w", err)}
+		return []error{fmt.Errorf("failed to init wr: %v", err)}
 	}
 
 	alertEvalTimesMap := map[time.Duration]struct{}{}
@@ -387,7 +384,7 @@ func (tg *testGroup) test(evalInterval time.Duration, groupOrderMap map[string]i
 				}
 			}
 			// flush series after each group evaluation
-			vmstorage.DebugFlush()
+			vmstorage.Storage.DebugFlush()
 		}
 
 		// check alert_rule_test case at every eval time

app/vmalert/config/config.go

@@ -113,15 +113,15 @@ func (g *Group) Validate(validateTplFn ValidateTplFn, validateExpressions bool)
 			// because correct types must be inherited after unmarshalling.
 			exprValidator := g.Type.ValidateExpr
 			if err := exprValidator(r.Expr); err != nil {
-				return fmt.Errorf("invalid expression for rule %q: %w", ruleName, err)
+				return fmt.Errorf("invalid expression for rule  %q: %w", ruleName, err)
 			}
 		}
 		if validateTplFn != nil {
 			if err := validateTplFn(r.Annotations); err != nil {
-				return fmt.Errorf("invalid annotations for rule %q: %w", ruleName, err)
+				return fmt.Errorf("invalid annotations for rule  %q: %w", ruleName, err)
 			}
 			if err := validateTplFn(r.Labels); err != nil {
-				return fmt.Errorf("invalid labels for rule %q: %w", ruleName, err)
+				return fmt.Errorf("invalid labels for rule  %q: %w", ruleName, err)
 			}
 		}
 	}
@@ -173,9 +173,9 @@ func (r *Rule) String() string {
 	if r.Alert != "" {
 		ruleType = "alerting"
 	}
-	var b strings.Builder
-	fmt.Fprintf(&b, "%s rule %q", ruleType, r.Name())
-	fmt.Fprintf(&b, "; expr: %q", r.Expr)
+	b := strings.Builder{}
+	b.WriteString(fmt.Sprintf("%s rule %q", ruleType, r.Name()))
+	b.WriteString(fmt.Sprintf("; expr: %q", r.Expr))
 
 	kv := sortMap(r.Labels)
 	for i := range kv {

app/vmalert/config/config_test.go

@@ -121,7 +121,7 @@ func TestParse_Failure(t *testing.T) {
 	f([]string{"testdata/dir/rules2-bad.rules"}, "function \"unknown\" not defined")
 	f([]string{"testdata/dir/rules3-bad.rules"}, "either `record` or `alert` must be set")
 	f([]string{"testdata/dir/rules4-bad.rules"}, "either `record` or `alert` must be set")
-	f([]string{"testdata/rules/rules1-bad.rules"}, "bad GraphiteQL expr")
+	f([]string{"testdata/rules/rules1-bad.rules"}, "bad graphite expr")
 	f([]string{"testdata/rules/vlog-rules0-bad.rules"}, "bad LogsQL expr")
 	f([]string{"testdata/dir/rules6-bad.rules"}, "missing ':' in header")
 	f([]string{"testdata/rules/rules-multi-doc-bad.rules"}, "unknown fields")
@@ -283,7 +283,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 				Expr:   "up | 0",
 			},
 		},
-	}, true, "bad MetricsQL expr")
+	}, true, "bad prometheus expr")
 
 	f(&Group{
 		Name: "test graphite expr",
@@ -293,7 +293,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 				"description": "some-description",
 			}},
 		},
-	}, true, "bad GraphiteQL expr")
+	}, true, "bad graphite expr")
 
 	f(&Group{
 		Name: "test vlogs expr",
@@ -327,7 +327,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 				Expr:   "sum(up == 0 ) by (host)",
 			},
 		},
-	}, true, "bad GraphiteQL expr")
+	}, true, "bad graphite expr")
 
 	f(&Group{
 		Name: "test vlogs with prometheus exp",
@@ -351,7 +351,7 @@ func TestGroupValidate_Failure(t *testing.T) {
 				For:    promutil.NewDuration(10 * time.Millisecond),
 			},
 		},
-	}, true, "bad MetricsQL expr")
+	}, true, "bad prometheus expr")
 }
 
 func TestGroupValidate_Success(t *testing.T) {

app/vmalert/config/types.go

@@ -66,11 +66,11 @@ func (t *Type) ValidateExpr(expr string) error {
 	switch t.String() {
 	case "graphite":
 		if _, err := graphiteql.Parse(expr); err != nil {
-			return fmt.Errorf("bad GraphiteQL expr: %q, err: %w", expr, err)
+			return fmt.Errorf("bad graphite expr: %q, err: %w", expr, err)
 		}
 	case "prometheus":
 		if _, err := metricsql.Parse(expr); err != nil {
-			return fmt.Errorf("bad MetricsQL expr: %q, err: %w", expr, err)
+			return fmt.Errorf("bad prometheus expr: %q, err: %w", expr, err)
 		}
 	case "vlogs":
 		q, err := logstorage.ParseStatsQuery(expr, 0)

app/vmalert/datasource/client_prom.go

@@ -89,7 +89,7 @@ func (pi *promInstant) Unmarshal(b []byte) error {
 		labels.Visit(func(key []byte, v *fastjson.Value) {
 			lv, errLocal := v.StringBytes()
 			if errLocal != nil {
-				err = fmt.Errorf("error when parsing label value %q: %w", v, errLocal)
+				err = fmt.Errorf("error when parsing label value %q: %s", v, errLocal)
 				return
 			}
 			r.Labels = append(r.Labels, prompb.Label{
@@ -112,7 +112,7 @@ func (pi *promInstant) Unmarshal(b []byte) error {
 		r.Timestamps = []int64{sample[0].GetInt64()}
 		val, err := sample[1].StringBytes()
 		if err != nil {
-			return fmt.Errorf("error when parsing `value` object %q: %w", sample[1], err)
+			return fmt.Errorf("error when parsing `value` object %q: %s", sample[1], err)
 		}
 		f, err := strconv.ParseFloat(bytesutil.ToUnsafeString(val), 64)
 		if err != nil {

app/vmalert/datasource/client_test.go

@@ -772,7 +772,7 @@ func TestHeaders(t *testing.T) {
 
 	// basic auth
 	f(func() *Client {
-		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBasicAuth("foo", "", "bar", ""))
+		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBasicAuth("foo", "bar", ""))
 		if err != nil {
 			t.Fatalf("Error get auth config: %s", err)
 		}
@@ -817,7 +817,7 @@ func TestHeaders(t *testing.T) {
 
 	// custom header overrides basic auth
 	f(func() *Client {
-		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBasicAuth("foo", "", "bar", ""))
+		cfg, err := vmalertutil.AuthConfig(vmalertutil.WithBasicAuth("foo", "bar", ""))
 		if err != nil {
 			t.Fatalf("Error get auth config: %s", err)
 		}

app/vmalert/datasource/init.go

@@ -27,7 +27,6 @@ var (
 		"Multiple headers must be delimited by '^^': -datasource.headers='header1:value1^^header2:value2'")
 
 	basicAuthUsername     = flag.String("datasource.basicAuth.username", "", "Optional basic auth username for -datasource.url")
-	basicAuthUsernameFile = flag.String("datasource.basicAuth.usernameFile", "", "Optional path to basic auth username to use for -datasource.url")
 	basicAuthPassword     = flag.String("datasource.basicAuth.password", "", "Optional basic auth password for -datasource.url")
 	basicAuthPasswordFile = flag.String("datasource.basicAuth.passwordFile", "", "Optional path to basic auth password to use for -datasource.url")
 
@@ -64,7 +63,6 @@ func InitSecretFlags() {
 	if !*showDatasourceURL {
 		flagutil.RegisterSecretFlag("datasource.url")
 	}
-	flagutil.RegisterSecretFlag("datasource.headers")
 }
 
 // ShowDatasourceURL whether to show -datasource.url with sensitive information
@@ -107,7 +105,7 @@ func Init(extraParams url.Values) (QuerierBuilder, error) {
 		return nil, fmt.Errorf("cannot parse JSON for -datasource.oauth2.endpointParams=%s: %w", *oauth2EndpointParams, err)
 	}
 	authCfg, err := vmalertutil.AuthConfig(
-		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthUsernameFile, *basicAuthPassword, *basicAuthPasswordFile),
+		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
 		vmalertutil.WithBearer(*bearerToken, *bearerTokenFile),
 		vmalertutil.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
 		vmalertutil.WithHeaders(*headers))

app/vmalert/notifier/alertmanager.go

@@ -191,7 +191,7 @@ func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, authCfg proma
 	}
 
 	aCfg, err := vmalertutil.AuthConfig(
-		vmalertutil.WithBasicAuth(ba.Username, ba.UsernameFile, ba.Password.String(), ba.PasswordFile),
+		vmalertutil.WithBasicAuth(ba.Username, ba.Password.String(), ba.PasswordFile),
 		vmalertutil.WithBearer(authCfg.BearerToken.String(), authCfg.BearerTokenFile),
 		vmalertutil.WithOAuth(oauth.ClientID, oauth.ClientSecret.String(), oauth.ClientSecretFile, oauth.TokenURL, strings.Join(oauth.Scopes, ";"), oauth.EndpointParams),
 		vmalertutil.WithHeaders(strings.Join(authCfg.Headers, "^^")),

app/vmalert/notifier/config_watcher.go

@@ -105,7 +105,7 @@ func (cw *configWatcher) add(typeK TargetType, interval time.Duration, targetsFn
 			}
 			targetMetadata, errors := getTargetMetadata(targetsFn, cw.cfg)
 			for _, err := range errors {
-				logger.Errorf("failed to init notifier for %q: %s", typeK, err)
+				logger.Errorf("failed to init notifier for %q: %w", typeK, err)
 			}
 			cw.updateTargets(typeK, targetMetadata, cw.cfg, cw.genFn)
 		}
@@ -274,7 +274,7 @@ func (cw *configWatcher) updateTargets(key TargetType, targetMts map[string]targ
 	for addr, metadata := range targetMts {
 		am, err := NewAlertManager(addr, genFn, cfg.HTTPClientConfig, metadata.alertRelabelConfigs, cfg.Timeout.Duration())
 		if err != nil {
-			logger.Errorf("failed to init %s notifier with addr %q: %s", key, addr, err)
+			logger.Errorf("failed to init %s notifier with addr %q: %w", key, addr, err)
 			continue
 		}
 		updatedTargets = append(updatedTargets, Target{

app/vmalert/notifier/init.go

@@ -36,7 +36,6 @@ var (
 		"For example, -remoteWrite.headers='My-Auth:foobar' would send 'My-Auth: foobar' HTTP header with every request to the corresponding -notifier.url. "+
 		"Multiple headers must be delimited by '^^': -notifier.headers='header1:value1^^header2:value2,header3:value3'")
 	basicAuthUsername     = flagutil.NewArrayString("notifier.basicAuth.username", "Optional basic auth username for -notifier.url")
-	basicAuthUsernameFile = flagutil.NewArrayString("notifier.basicAuth.usernameFile", "Optional path to basic auth username file for -notifier.url")
 	basicAuthPassword     = flagutil.NewArrayString("notifier.basicAuth.password", "Optional basic auth password for -notifier.url")
 	basicAuthPasswordFile = flagutil.NewArrayString("notifier.basicAuth.passwordFile", "Optional path to basic auth password file for -notifier.url")
 
@@ -194,7 +193,6 @@ func InitSecretFlags() {
 	if !*showNotifierURL {
 		flagutil.RegisterSecretFlag("notifier.url")
 	}
-	flagutil.RegisterSecretFlag("notifier.headers")
 }
 
 func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
@@ -215,7 +213,6 @@ func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
 			},
 			BasicAuth: &promauth.BasicAuthConfig{
 				Username:     basicAuthUsername.GetOptionalArg(i),
-				UsernameFile: basicAuthUsernameFile.GetOptionalArg(i),
 				Password:     promauth.NewSecret(basicAuthPassword.GetOptionalArg(i)),
 				PasswordFile: basicAuthPasswordFile.GetOptionalArg(i),
 			},

app/vmalert/notifier/notifier.go

@@ -14,7 +14,7 @@ type Notifier interface {
 	Send(ctx context.Context, alerts []Alert, alertLabels [][]prompb.Label, notifierHeaders map[string]string) error
 	// Addr returns address where alerts are sent.
 	Addr() string
-	// LastError returns error, that occurred during last attempt to send data
+	// LastError returns error, that occured during last attempt to send data
 	LastError() string
 	// Close is a destructor for the Notifier
 	Close()

app/vmalert/remoteread/init.go

@@ -28,7 +28,6 @@ var (
 		"Multiple headers must be delimited by '^^': -remoteRead.headers='header1:value1^^header2:value2'")
 
 	basicAuthUsername     = flag.String("remoteRead.basicAuth.username", "", "Optional basic auth username for -remoteRead.url")
-	basicAuthUsernameFile = flag.String("remoteRead.basicAuth.usernameFile", "", "Optional path to basic auth username to use for -remoteRead.url")
 	basicAuthPassword     = flag.String("remoteRead.basicAuth.password", "", "Optional basic auth password for -remoteRead.url")
 	basicAuthPasswordFile = flag.String("remoteRead.basicAuth.passwordFile", "", "Optional path to basic auth password to use for -remoteRead.url")
 
@@ -59,7 +58,6 @@ func InitSecretFlags() {
 	if !*showRemoteReadURL {
 		flagutil.RegisterSecretFlag("remoteRead.url")
 	}
-	flagutil.RegisterSecretFlag("remoteRead.headers")
 }
 
 // Init creates a Querier from provided flag values.
@@ -82,7 +80,7 @@ func Init() (datasource.QuerierBuilder, error) {
 		return nil, fmt.Errorf("cannot parse JSON for -remoteRead.oauth2.endpointParams=%s: %w", *oauth2EndpointParams, err)
 	}
 	authCfg, err := vmalertutil.AuthConfig(
-		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthUsernameFile, *basicAuthPassword, *basicAuthPasswordFile),
+		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
 		vmalertutil.WithBearer(*bearerToken, *bearerTokenFile),
 		vmalertutil.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
 		vmalertutil.WithHeaders(*headers))

app/vmalert/remotewrite/client.go

@@ -11,7 +11,6 @@ import (
 	"path"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/cespare/xxhash/v2"
@@ -19,8 +18,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
@@ -60,11 +57,6 @@ type Client struct {
 
 	wg     sync.WaitGroup
 	doneCh chan struct{}
-
-	// Whether to encode the write request with VictoriaMetrics remote write protocol.
-	// It is set to true by default, and will be switched to false if the client
-	// receives specific errors indicating that the remote storage doesn't support VictoriaMetrics remote write protocol.
-	isVMRemoteWrite atomic.Bool
 }
 
 // Config is config for remote write client.
@@ -124,7 +116,6 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 		doneCh:        make(chan struct{}),
 		input:         make(chan prompb.TimeSeries, cfg.MaxQueueSize),
 	}
-	c.isVMRemoteWrite.Store(true)
 
 	for i := 0; i < cc; i++ {
 		c.wg.Go(func() {
@@ -274,16 +265,8 @@ func (c *Client) flush(ctx context.Context, wr *prompb.WriteRequest) {
 	defer wr.Reset()
 	defer bufferFlushDuration.UpdateDuration(time.Now())
 
-	bb := writeRequestBufPool.Get()
-	bb.B = wr.MarshalProtobuf(bb.B[:0])
-	zb := compressBufPool.Get()
-	defer compressBufPool.Put(zb)
-	if c.isVMRemoteWrite.Load() {
-		zb.B = zstd.CompressLevel(zb.B[:0], bb.B, 0)
-	} else {
-		zb.B = snappy.Encode(zb.B[:cap(zb.B)], bb.B)
-	}
-	writeRequestBufPool.Put(bb)
+	data := wr.MarshalProtobuf(nil)
+	b := snappy.Encode(nil, data)
 
 	maxRetryInterval := *retryMaxTime
 	bt := timeutil.NewBackoffTimer(*retryMinInterval, maxRetryInterval)
@@ -295,17 +278,17 @@ func (c *Client) flush(ctx context.Context, wr *prompb.WriteRequest) {
 	attempts := 0
 L:
 	for {
-		err := c.send(ctx, zb.B)
+		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
 			// the connection. So we do a one more attempt in hope request will succeed.
-			err = c.send(ctx, zb.B)
+			err = c.send(ctx, b)
 		}
 		if err == nil {
 			sentRows.Add(len(wr.Timeseries))
-			sentBytes.Add(len(zb.B))
+			sentBytes.Add(len(b))
 			flushedRows.Update(float64(len(wr.Timeseries)))
-			flushedBytes.Update(float64(len(zb.B)))
+			flushedBytes.Update(float64(len(b)))
 			return
 		}
 
@@ -357,16 +340,12 @@ func (c *Client) send(ctx context.Context, data []byte) error {
 		return fmt.Errorf("failed to create new HTTP request: %w", err)
 	}
 
-	req.Header.Set("User-Agent", "vmalert")
+	// RFC standard compliant headers
+	req.Header.Set("Content-Encoding", "snappy")
 	req.Header.Set("Content-Type", "application/x-protobuf")
 
-	if encoding.IsZstd(data) {
-		req.Header.Set("Content-Encoding", "zstd")
-		req.Header.Set("X-VictoriaMetrics-Remote-Write-Version", "1")
-	} else {
-		req.Header.Set("Content-Encoding", "snappy")
-		req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
-	}
+	// Prometheus compliant headers
+	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
 
 	if c.authCfg != nil {
 		err = c.authCfg.SetHeaders(req, true)
@@ -395,29 +374,6 @@ func (c *Client) send(ctx context.Context, data []byte) error {
 		// respond with HTTP 2xx status code when write is successful.
 		return nil
 	case 4:
-		// - Remote Write v1 specification implicitly expects a `400 Bad Request` when the encoding is not supported.
-		// - Remote Write v2 specification explicitly specifies a `415 Unsupported Media Type` for unsupported encodings.
-		// - Real-world implementations of v1 use both 400 and 415 status codes.
-		// See more in research: https://github.com/VictoriaMetrics/VictoriaMetrics/pull/8462#issuecomment-2786918054
-		if resp.StatusCode == http.StatusUnsupportedMediaType || resp.StatusCode == http.StatusBadRequest {
-			if encoding.IsZstd(data) {
-				logger.Infof("received unsupported media type or bad request from remote storage at %q. Re-packing the block to Prometheus remote write and retrying."+
-					"See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol", req.URL.Redacted())
-				zstdBlockLen := len(data)
-				data, err = repackBlockFromZstdToSnappy(data)
-				if err == nil {
-					logger.Infof("received unsupported media type or bad request from remote storage at %q. Downgrading protocol from VictoriaMetrics to Prometheus remote write for all future requests. "+
-						"See https://docs.victoriametrics.com/victoriametrics/vmagent/#victoriametrics-remote-write-protocol", req.URL.Redacted())
-					c.isVMRemoteWrite.Store(false)
-					return c.send(ctx, data)
-				}
-
-				logger.Warnf("failed to repack zstd block (%d bytes) to snappy: %s; The block will be rejected. "+
-					"Possible cause: ungraceful shutdown leading to persisted queue corruption.",
-					zstdBlockLen, err)
-			}
-		}
-
 		if resp.StatusCode != http.StatusTooManyRequests {
 			// MUST NOT retry write requests on HTTP 4xx responses other than 429
 			return &nonRetriableError{
@@ -438,19 +394,3 @@ type nonRetriableError struct {
 func (e *nonRetriableError) Error() string {
 	return e.err.Error()
 }
-
-var (
-	writeRequestBufPool bytesutil.ByteBufferPool
-	compressBufPool     bytesutil.ByteBufferPool
-)
-
-// repackBlockFromZstdToSnappy repacks the given zstd-compressed block to snappy-compressed block.
-func repackBlockFromZstdToSnappy(zstdBlock []byte) ([]byte, error) {
-	plainBlock := make([]byte, 0, len(zstdBlock)*2)
-	plainBlock, err := encoding.DecompressZSTD(plainBlock, zstdBlock)
-	if err != nil {
-		return nil, err
-	}
-
-	return snappy.Encode(nil, plainBlock), nil
-}

app/vmalert/remotewrite/client_test.go

@@ -12,7 +12,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
+	"github.com/golang/snappy"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
 )
 
@@ -102,10 +103,7 @@ func TestClient_run_maxBatchSizeDuringShutdown(t *testing.T) {
 
 		// push time series to the client.
 		for range pushCnt {
-			if err = rwClient.Push(prompb.TimeSeries{
-				Labels:  []prompb.Label{{Name: "__name__", Value: "m"}},
-				Samples: []prompb.Sample{{Value: 1, Timestamp: 1000}},
-			}); err != nil {
+			if err = rwClient.Push(prompb.TimeSeries{}); err != nil {
 				t.Fatalf("cannot time series to the client: %s", err)
 			}
 		}
@@ -158,8 +156,8 @@ func (rw *rwServer) handler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	h := r.Header.Get("Content-Encoding")
-	if h != "zstd" {
-		rw.err(w, fmt.Errorf("header read error: Content-Encoding is not zstd (%q)", h))
+	if h != "snappy" {
+		rw.err(w, fmt.Errorf("header read error: Content-Encoding is not snappy (%q)", h))
 	}
 
 	h = r.Header.Get("Content-Type")
@@ -167,9 +165,9 @@ func (rw *rwServer) handler(w http.ResponseWriter, r *http.Request) {
 		rw.err(w, fmt.Errorf("header read error: Content-Type is not x-protobuf (%q)", h))
 	}
 
-	h = r.Header.Get("X-VictoriaMetrics-Remote-Write-Version")
-	if h != "1" {
-		rw.err(w, fmt.Errorf("header read error: X-VictoriaMetrics-Remote-Write-Version is not 1 (%q)", h))
+	h = r.Header.Get("X-Prometheus-Remote-Write-Version")
+	if h != "0.1.0" {
+		rw.err(w, fmt.Errorf("header read error: X-Prometheus-Remote-Write-Version is not 0.1.0 (%q)", h))
 	}
 
 	data, err := io.ReadAll(r.Body)
@@ -179,7 +177,7 @@ func (rw *rwServer) handler(w http.ResponseWriter, r *http.Request) {
 	}
 	defer func() { _ = r.Body.Close() }()
 
-	b, err := zstd.Decompress(nil, data)
+	b, err := snappy.Decode(nil, data)
 	if err != nil {
 		rw.err(w, fmt.Errorf("decode err: %w", err))
 		return

app/vmalert/remotewrite/debug_client.go

@@ -9,7 +9,8 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding/zstd"
+	"github.com/golang/snappy"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
@@ -63,17 +64,19 @@ func (c *DebugClient) Close() error {
 }
 
 func (c *DebugClient) send(data []byte) error {
-	b := zstd.CompressLevel(nil, data, 0)
+	b := snappy.Encode(nil, data)
 	r := bytes.NewReader(b)
 	req, err := http.NewRequest(http.MethodPost, c.addr, r)
 	if err != nil {
 		return fmt.Errorf("failed to create new HTTP request: %w", err)
 	}
 
-	req.Header.Set("Content-Encoding", "zstd")
+	// RFC standard compliant headers
+	req.Header.Set("Content-Encoding", "snappy")
 	req.Header.Set("Content-Type", "application/x-protobuf")
 
-	req.Header.Set("X-VictoriaMetrics-Remote-Write-Version", "1")
+	// Prometheus compliant headers
+	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
 
 	if !*disablePathAppend {
 		req.URL.Path = path.Join(req.URL.Path, "/api/v1/write")

app/vmalert/remotewrite/init.go

@@ -13,8 +13,8 @@ import (
 )
 
 var (
-	addr = flag.String("remoteWrite.url", "", "Optional URL to persist alerts state and recording rules results in form of timeseries. "+
-		"It must support either VictoriaMetrics remote write protocol or Prometheus remote_write protocol. "+
+	addr = flag.String("remoteWrite.url", "", "Optional URL to VictoriaMetrics or vminsert where to persist alerts state "+
+		"and recording rules results in form of timeseries. "+
 		"Supports address in the form of IP address with a port (e.g., http://127.0.0.1:8428) or DNS SRV record. "+
 		"For example, if -remoteWrite.url=http://127.0.0.1:8428 is specified, "+
 		"then the alerts state will be written to http://127.0.0.1:8428/api/v1/write . See also -remoteWrite.disablePathAppend, '-remoteWrite.showURL'.")
@@ -26,7 +26,6 @@ var (
 		"Multiple headers must be delimited by '^^': -remoteWrite.headers='header1:value1^^header2:value2'")
 
 	basicAuthUsername     = flag.String("remoteWrite.basicAuth.username", "", "Optional basic auth username for -remoteWrite.url")
-	basicAuthUsernameFile = flag.String("remoteWrite.basicAuth.usernameFile", "", "Optional path to basic auth username to use for -remoteWrite.url")
 	basicAuthPassword     = flag.String("remoteWrite.basicAuth.password", "", "Optional basic auth password for -remoteWrite.url")
 	basicAuthPasswordFile = flag.String("remoteWrite.basicAuth.passwordFile", "", "Optional path to basic auth password to use for -remoteWrite.url")
 
@@ -62,7 +61,6 @@ func InitSecretFlags() {
 	if !*showRemoteWriteURL {
 		flagutil.RegisterSecretFlag("remoteWrite.url")
 	}
-	flagutil.RegisterSecretFlag("remoteWrite.headers")
 }
 
 // Init creates Client object from given flags.
@@ -85,7 +83,7 @@ func Init(ctx context.Context) (*Client, error) {
 		return nil, fmt.Errorf("cannot parse JSON for -remoteWrite.oauth2.endpointParams=%s: %w", *oauth2EndpointParams, err)
 	}
 	authCfg, err := vmalertutil.AuthConfig(
-		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthUsernameFile, *basicAuthPassword, *basicAuthPasswordFile),
+		vmalertutil.WithBasicAuth(*basicAuthUsername, *basicAuthPassword, *basicAuthPasswordFile),
 		vmalertutil.WithBearer(*bearerToken, *bearerTokenFile),
 		vmalertutil.WithOAuth(*oauth2ClientID, *oauth2ClientSecret, *oauth2ClientSecretFile, *oauth2TokenURL, *oauth2Scopes, endpointParams),
 		vmalertutil.WithHeaders(*headers))

app/vmalert/rule/alerting.go

@@ -601,7 +601,7 @@ func (ar *AlertingRule) exec(ctx context.Context, ts time.Time, limit int) ([]pr
 func (ar *AlertingRule) expandLabelTemplates(m datasource.Metric, qFn templates.QueryFn) (*labelSet, error) {
 	ls, err := ar.toLabels(m, qFn)
 	if err != nil {
-		return ls, fmt.Errorf("failed to expand label templates: %w", err)
+		return ls, fmt.Errorf("failed to expand label templates: %s", err)
 	}
 	return ls, nil
 }
@@ -620,7 +620,7 @@ func (ar *AlertingRule) expandAnnotationTemplates(m datasource.Metric, qFn templ
 	}
 	as, err := notifier.ExecTemplate(qFn, ar.Annotations, tplData)
 	if err != nil {
-		return as, fmt.Errorf("failed to expand annotation templates: %w", err)
+		return as, fmt.Errorf("failed to expand annotation templates: %s", err)
 	}
 	return as, nil
 }

app/vmalert/rule/group.go

@@ -8,7 +8,6 @@ import (
 	"hash/fnv"
 	"maps"
 	"net/url"
-	"path"
 	"sync"
 	"time"
 
@@ -43,9 +42,6 @@ var (
 		"For example, if lookback=1h then range from now() to now()-1h will be scanned.")
 	maxStartDelay = flag.Duration("group.maxStartDelay", 5*time.Minute, "Defines the max delay before starting the group evaluation. Group's start is artificially delayed for random duration on interval"+
 		" [0..min(--group.maxStartDelay, group.interval)]. This helps smoothing out the load on the configured datasource, so evaluations aren't executed too close to each other.")
-	ruleStripFilePath = flag.Bool("rule.stripFilePath", false, "Whether to strip rule file paths in logs and all API responses, including /metrics. "+
-		"For example, file path '/path/to/tenant_id/rules.yml' will be stripped to 'groupHashID/rules.yml'. "+
-		"This flag may be useful for hiding sensitive information in file paths, such as S3 bucket details.")
 )
 
 // Group is an entity for grouping rules
@@ -95,7 +91,6 @@ type groupMetrics struct {
 	iterationTotal    *metrics.Counter
 	iterationDuration *metrics.Summary
 	iterationMissed   *metrics.Counter
-	iterationReset    *metrics.Counter
 	iterationInterval *metrics.Gauge
 }
 
@@ -152,12 +147,6 @@ func NewGroup(cfg config.Group, qb datasource.QuerierBuilder, defaultInterval ti
 		g.EvalDelay = &cfg.EvalDelay.D
 	}
 	g.id = g.CreateID()
-	// strip file path from group.File after generated group ID when ruleStripFilePath is set,
-	// so it won't be exposed in logs and api responses
-	if *ruleStripFilePath {
-		_, filename := path.Split(g.File)
-		g.File = fmt.Sprintf("%d/%s", g.id, filename)
-	}
 	for _, h := range cfg.Headers {
 		g.Headers[h.Key] = h.Value
 	}
@@ -331,7 +320,6 @@ func (g *Group) Init() {
 	g.metrics.iterationTotal = g.metrics.set.NewCounter(fmt.Sprintf(`vmalert_iteration_total{%s}`, labels))
 	g.metrics.iterationDuration = g.metrics.set.NewSummary(fmt.Sprintf(`vmalert_iteration_duration_seconds{%s}`, labels))
 	g.metrics.iterationMissed = g.metrics.set.NewCounter(fmt.Sprintf(`vmalert_iteration_missed_total{%s}`, labels))
-	g.metrics.iterationReset = g.metrics.set.NewCounter(fmt.Sprintf(`vmalert_iteration_reset_total{%s}`, labels))
 	g.metrics.iterationInterval = g.metrics.set.NewGauge(fmt.Sprintf(`vmalert_iteration_interval_seconds{%s}`, labels), func() float64 {
 		i := g.Interval.Seconds()
 		return i
@@ -476,16 +464,14 @@ func (g *Group) Start(ctx context.Context, rw remotewrite.RWClient, rr datasourc
 			if missed < 0 {
 				// missed can become < 0 due to irregular delays during evaluation
 				// which can result in time.Since(evalTS) < g.Interval;
-				// or the system wall clock was changed backward,
-				// Reset the evalTS to the current time.
+				// or the system wall clock was changed backward
+				missed = 0
 				evalTS = time.Now()
-				g.metrics.iterationReset.Inc()
-			} else {
-				evalTS = evalTS.Add((missed + 1) * g.Interval)
 			}
 			if missed > 0 {
 				g.metrics.iterationMissed.Inc()
 			}
+			evalTS = evalTS.Add((missed + 1) * g.Interval)
 
 			eval(evalCtx, evalTS)
 		}

app/vmalert/rule/group_test.go

@@ -742,64 +742,3 @@ func parseTime(t *testing.T, s string) time.Time {
 	}
 	return tt
 }
-
-func TestRuleStripFilePath(t *testing.T) {
-	configG := config.Group{
-		Name:        "group",
-		File:        "/var/local/test/rules.yaml",
-		Type:        config.NewRawType("prometheus"),
-		Concurrency: 1,
-		Rules: []config.Rule{
-			{
-				ID:    0,
-				Alert: "alert",
-			},
-			{
-				ID:     1,
-				Record: "record",
-			},
-		}}
-	qb := &datasource.FakeQuerier{}
-	g := NewGroup(configG, qb, 1*time.Minute, nil)
-
-	gID := g.id
-	if g.File != "/var/local/test/rules.yaml" {
-		t.Fatalf("expected file path to be unchanged; got %q instead", g.File)
-	}
-
-	for _, r := range g.Rules {
-		if ar, ok := r.(*AlertingRule); ok {
-			if ar.File != "/var/local/test/rules.yaml" {
-				t.Fatalf("expected rule file path to be unchanged; got %q instead", ar.File)
-			}
-		}
-		if rr, ok := r.(*RecordingRule); ok {
-			if rr.File != "/var/local/test/rules.yaml" {
-				t.Fatalf("expected rule file path to be unchanged; got %q instead", rr.File)
-			}
-		}
-	}
-
-	oldRuleStripFilePath := *ruleStripFilePath
-	*ruleStripFilePath = true
-	defer func() {
-		*ruleStripFilePath = oldRuleStripFilePath
-	}()
-	g = NewGroup(configG, qb, 1*time.Minute, nil)
-
-	if g.File != fmt.Sprintf("%d/rules.yaml", gID) {
-		t.Fatalf("expected file path to be stripped to %q; got %q instead", fmt.Sprintf("%d/rules.yaml", gID), g.File)
-	}
-	for _, r := range g.Rules {
-		if ar, ok := r.(*AlertingRule); ok {
-			if ar.File != fmt.Sprintf("%d/rules.yaml", gID) {
-				t.Fatalf("expected rule file path to be unchanged; got %q instead", ar.File)
-			}
-		}
-		if rr, ok := r.(*RecordingRule); ok {
-			if rr.File != fmt.Sprintf("%d/rules.yaml", gID) {
-				t.Fatalf("expected rule file path to be unchanged; got %q instead", rr.File)
-			}
-		}
-	}
-}

app/vmalert/rule/web.go

@@ -252,9 +252,6 @@ func (r *ApiRule) ExtendState() {
 
 // ToAPI returns ApiGroup representation of g
 func (g *Group) ToAPI() *ApiGroup {
-	if g == nil {
-		return &ApiGroup{}
-	}
 	g.mu.RLock()
 	defer g.mu.RUnlock()
 	ag := ApiGroup{

app/vmalert/templates/template.go

@@ -402,20 +402,6 @@ func templateFuncs() textTpl.FuncMap {
 			return t, nil
 		},
 
-		// formatTime formats the given Unix timestamp with the provided layout.
-		// For example: {{ now | formatTime "2006-01-02T15:04:05Z07:00" }}
-		"formatTime": func(layout string, i any) (string, error) {
-			v, err := toFloat64(i)
-			if err != nil {
-				return "", fmt.Errorf("formatTime: %w", err)
-			}
-			if math.IsNaN(v) || math.IsInf(v, 0) {
-				return "", fmt.Errorf("formatTime: cannot convert %v to time", v)
-			}
-			t := timeFromUnixTimestamp(v).Time().UTC()
-			return t.Format(layout), nil
-		},
-
 		/* URLs */
 
 		// externalURL returns value of `external.url` flag

app/vmalert/templates/template_test.go

@@ -6,7 +6,6 @@ import (
 	"strings"
 	"testing"
 	textTpl "text/template"
-	"time"
 )
 
 func TestTemplateFuncs_StringConversion(t *testing.T) {
@@ -104,26 +103,6 @@ func TestTemplateFuncs_Formatting(t *testing.T) {
 	f("humanizeTimestamp", 1679055557, "2023-03-17 12:19:17 +0000 UTC")
 }
 
-func TestTemplateFuncs_FormatTime(t *testing.T) {
-	funcs := templateFuncs()
-	formatTime := funcs["formatTime"].(func(layout string, i any) (string, error))
-
-	f := func(layout string, input any, expected string) {
-		t.Helper()
-		result, err := formatTime(layout, input)
-		if err != nil {
-			t.Fatalf("unexpected error for formatTime(%q, %v): %s", layout, input, err)
-		}
-		if result != expected {
-			t.Fatalf("unexpected result for formatTime(%q, %v); got\n%s\nwant\n%s", layout, input, result, expected)
-		}
-	}
-
-	f(time.RFC3339, float64(1679055557), "2023-03-17T12:19:17Z")
-	f("2006-01-02T15:04:05", int64(1679055557), "2023-03-17T12:19:17")
-	f(time.RFC822, int(1679055557), "17 Mar 23 12:19 UTC")
-}
-
 func mkTemplate(current, replacement any) textTemplate {
 	tmpl := textTemplate{}
 	if current != nil {

app/vmalert/vmalertutil/auth.go

@@ -20,12 +20,11 @@ func AuthConfig(filterOptions ...AuthConfigOptions) (*promauth.Config, error) {
 }
 
 // WithBasicAuth returns AuthConfigOptions and initialized promauth.BasicAuthConfig based on given params
-func WithBasicAuth(username, usernameFile, password, passwordFile string) AuthConfigOptions {
+func WithBasicAuth(username, password, passwordFile string) AuthConfigOptions {
 	return func(config *promauth.HTTPClientConfig) {
-		if username != "" || usernameFile != "" || password != "" || passwordFile != "" {
+		if username != "" || password != "" || passwordFile != "" {
 			config.BasicAuth = &promauth.BasicAuthConfig{
 				Username:     username,
-				UsernameFile: usernameFile,
 				Password:     promauth.NewSecret(password),
 				PasswordFile: passwordFile,
 			}

app/vmalert/web.go

@@ -11,8 +11,6 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/VictoriaMetrics/metricsql"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/rule"
@@ -77,7 +75,7 @@ var (
 func marshalJson(v any, kind string) ([]byte, *httpserver.ErrorWithStatusCode) {
 	data, err := json.Marshal(v)
 	if err != nil {
-		return nil, errResponse(fmt.Errorf("failed to marshal %s: %w", kind, err), http.StatusInternalServerError)
+		return nil, errResponse(fmt.Errorf("failed to marshal %s: %s", kind, err), http.StatusInternalServerError)
 	}
 	return data, nil
 }
@@ -162,12 +160,12 @@ func (rh *requestHandler) handler(w http.ResponseWriter, r *http.Request) bool {
 
 	case "/vmalert/api/v1/alerts", "/api/v1/alerts":
 		// path used by Grafana for ng alerting
-		af, err := newAlertsFilter(r)
+		gf, err := newGroupsFilter(r)
 		if err != nil {
 			errJson(w, r, err)
 			return true
 		}
-		data, err := rh.listAlerts(af)
+		data, err := rh.listAlerts(gf)
 		if err != nil {
 			errJson(w, r, err)
 			return true
@@ -327,48 +325,6 @@ func (gf *groupsFilter) matches(group *rule.Group) bool {
 	return true
 }
 
-type alertsFilter struct {
-	gf    *groupsFilter
-	match [][]metricsql.LabelFilter
-}
-
-func getMatchFilters(matches []string) ([][]metricsql.LabelFilter, *httpserver.ErrorWithStatusCode) {
-	if len(matches) == 0 {
-		return nil, nil
-	}
-	tfss := make([][]metricsql.LabelFilter, 0, len(matches))
-	for _, s := range matches {
-		expr, err := metricsql.Parse(s)
-		if err != nil {
-			return nil, errResponse(fmt.Errorf(`invalid parameter "match[]": failed to parse %q: %w`, s, err), http.StatusBadRequest)
-		}
-		me, ok := expr.(*metricsql.MetricExpr)
-		if !ok {
-			return nil, errResponse(fmt.Errorf(`invalid parameter "match[]": expecting metricSelector; got %q`, expr.AppendString(nil)), http.StatusBadRequest)
-		}
-		if len(me.LabelFilterss) == 0 {
-			return nil, errResponse(fmt.Errorf(`invalid parameter "match[]": labelFilterss cannot be empty`), http.StatusBadRequest)
-		}
-		tfss = append(tfss, me.LabelFilterss...)
-	}
-	return tfss, nil
-}
-
-func newAlertsFilter(r *http.Request) (*alertsFilter, *httpserver.ErrorWithStatusCode) {
-	gf, err := newGroupsFilter(r)
-	if err != nil {
-		return nil, err
-	}
-
-	var af alertsFilter
-	af.gf = gf
-	af.match, err = getMatchFilters(r.Form["match[]"])
-	if err != nil {
-		return nil, err
-	}
-	return &af, nil
-}
-
 // see https://prometheus.io/docs/prometheus/latest/querying/api/#rules
 type rulesFilter struct {
 	gf             *groupsFilter
@@ -379,7 +335,6 @@ type rulesFilter struct {
 	maxGroups      int
 	pageNum        int
 	search         string
-	match          [][]metricsql.LabelFilter
 	extendedStates bool
 }
 
@@ -400,10 +355,7 @@ func newRulesFilter(r *http.Request) (*rulesFilter, *httpserver.ErrorWithStatusC
 			return nil, errResponse(fmt.Errorf(`invalid parameter "type": not supported value %q`, ruleTypeParam), http.StatusBadRequest)
 		}
 	}
-	rf.match, err = getMatchFilters(r.Form["match[]"])
-	if err != nil {
-		return nil, err
-	}
+
 	states := vs["state"]
 	if len(states) == 0 {
 		states = vs["filter"]
@@ -464,47 +416,12 @@ func (rf *rulesFilter) matchesRule(r *rule.ApiRule) bool {
 	if len(rf.ruleNames) > 0 && !slices.Contains(rf.ruleNames, r.Name) {
 		return false
 	}
-	if !areLabelsMatch(r.Labels, rf.match) {
-		return false
-	}
 	if len(rf.states) == 0 {
 		return true
 	}
 	return slices.Contains(rf.states, r.State)
 }
 
-func areLabelsMatch(labels map[string]string, matches [][]metricsql.LabelFilter) bool {
-	if len(matches) == 0 {
-		return true
-	}
-	// labels need to match at least one of the provided match[] arg
-	return slices.ContainsFunc(matches, func(filters []metricsql.LabelFilter) bool {
-		for _, mf := range filters {
-			if !isLabelFilterMatch(labels[mf.Label], mf) {
-				return false
-			}
-		}
-		return true
-	})
-}
-
-func isLabelFilterMatch(s string, match metricsql.LabelFilter) bool {
-	if !match.IsRegexp {
-		if match.IsNegative {
-			return s != match.Value
-		}
-		return s == match.Value
-	}
-	re, err := metricsql.CompileRegexpAnchored(match.Value)
-	if err != nil {
-		return false
-	}
-	if match.IsNegative {
-		return !re.MatchString(s)
-	}
-	return re.MatchString(s)
-}
-
 func (rh *requestHandler) groups(rf *rulesFilter) *listGroupsResponse {
 	rh.m.groupsMu.RLock()
 	defer rh.m.groupsMu.RUnlock()
@@ -626,14 +543,14 @@ func (rh *requestHandler) groupAlerts() []rule.GroupAlerts {
 	return gAlerts
 }
 
-func (rh *requestHandler) listAlerts(af *alertsFilter) ([]byte, *httpserver.ErrorWithStatusCode) {
+func (rh *requestHandler) listAlerts(gf *groupsFilter) ([]byte, *httpserver.ErrorWithStatusCode) {
 	rh.m.groupsMu.RLock()
 	defer rh.m.groupsMu.RUnlock()
 
 	lr := listAlertsResponse{Status: "success"}
 	lr.Data.Alerts = make([]*rule.ApiAlert, 0)
 	for _, group := range rh.m.groups {
-		if !af.gf.matches(group) {
+		if !gf.matches(group) {
 			continue
 		}
 		g := group.ToAPI()
@@ -641,11 +558,7 @@ func (rh *requestHandler) listAlerts(af *alertsFilter) ([]byte, *httpserver.Erro
 			if r.Type != rule.TypeAlerting {
 				continue
 			}
-			for _, alert := range r.Alerts {
-				if areLabelsMatch(alert.Labels, af.match) {
-					lr.Data.Alerts = append(lr.Data.Alerts, alert)
-				}
-			}
+			lr.Data.Alerts = append(lr.Data.Alerts, r.Alerts...)
 		}
 	}
 

app/vmalert/web.qtpl

@@ -348,7 +348,7 @@
                 typeK, ns := keys[i], targets[notifier.TargetType(keys[i])]
                 count := len(ns)
             %}
-            <div class="w-100 flex-column">
+            <div class="w-100 flex-column vm-group">
                 <span class="d-flex justify-content-between" id="group-{%s typeK %}">
                     <a href="#group-{%s typeK %}">{%s typeK %} ({%d count %})</a>
                     <span
@@ -361,7 +361,7 @@
                 <div id="item-{%s typeK %}" class="collapse show">
                     <table class="table table-striped table-hover table-sm">
                         <thead>
-                            <tr>
+                            <tr class="vm-item">
                                 <th scope="col">Labels</th>
                                 <th scope="col">Address</th>
                             </tr>

app/vmalert/web.qtpl.go

@@ -1115,7 +1115,7 @@ func StreamListTargets(qw422016 *qt422016.Writer, r *http.Request, targets map[n
 
 //line app/vmalert/web.qtpl:350
 			qw422016.N().S(`
-            <div class="w-100 flex-column">
+            <div class="w-100 flex-column vm-group">
                 <span class="d-flex justify-content-between" id="group-`)
 //line app/vmalert/web.qtpl:352
 			qw422016.E().S(typeK)
@@ -1152,7 +1152,7 @@ func StreamListTargets(qw422016 *qt422016.Writer, r *http.Request, targets map[n
 			qw422016.N().S(`" class="collapse show">
                     <table class="table table-striped table-hover table-sm">
                         <thead>
-                            <tr>
+                            <tr class="vm-item">
                                 <th scope="col">Labels</th>
                                 <th scope="col">Address</th>
                             </tr>

app/vmalert/web_test.go

@@ -10,8 +10,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/VictoriaMetrics/metricsql"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
@@ -39,14 +37,12 @@ func TestHandler(t *testing.T) {
 			Concurrency: 1,
 			Rules: []config.Rule{
 				{
-					ID:     0,
-					Alert:  "alert",
-					Labels: map[string]string{"job": "foo"},
+					ID:    0,
+					Alert: "alert",
 				},
 				{
 					ID:     1,
 					Record: "record",
-					Labels: map[string]string{"job": "bar"},
 				},
 			},
 		}, fq, 1*time.Minute, nil)
@@ -132,18 +128,6 @@ func TestHandler(t *testing.T) {
 		if length := len(lr.Data.Alerts); length != 2 {
 			t.Fatalf("expected 2 alert got %d", length)
 		}
-
-		lr = listAlertsResponse{}
-		getResp(t, ts.URL+`/api/v1/alerts?match[]={job="foo"}`, &lr, 200)
-		if length := len(lr.Data.Alerts); length != 3 {
-			t.Fatalf("expected 3 alerts got %d", length)
-		}
-
-		lr = listAlertsResponse{}
-		getResp(t, ts.URL+`/api/v1/alerts?match[]={job="bar"}`, &lr, 200)
-		if length := len(lr.Data.Alerts); length != 0 {
-			t.Fatalf("expected 0 alerts got %d", length)
-		}
 	})
 	t.Run("/api/v1/alert?alertID&groupID", func(t *testing.T) {
 		expAlert := rule.NewAlertAPI(ar, ar.GetAlerts()[0])
@@ -258,13 +242,6 @@ func TestHandler(t *testing.T) {
 		check("/vmalert/api/v1/rules?datasource_type=graphite", 200, 1, 2)
 		check("/vmalert/api/v1/rules?datasource_type=graphiti", 400, 0, 0)
 
-		// invalid match[] params
-		check(`/vmalert/api/v1/rules?match[]={job=!"foo"}`, 400, 0, 0)
-		check(`/vmalert/api/v1/rules?match[]={job="foo"}`, 200, 3, 3)
-		check(`/vmalert/api/v1/rules?match[]={job="bar"}`, 200, 3, 3)
-		check(`/vmalert/api/v1/rules?match[]={job="bar"}&match[]={job="foo"}`, 200, 3, 6)
-		check(`/vmalert/api/v1/rules?match[]={job="barzz"}`, 200, 0, 0)
-
 		// no filtering expected due to bad params
 		check("/api/v1/rules?type=badParam", 400, 0, 0)
 		check("/api/v1/rules?foo=bar", 200, 3, 6)
@@ -390,116 +367,3 @@ func TestEmptyResponse(t *testing.T) {
 		}
 	})
 }
-
-func TestMatchesRule(t *testing.T) {
-	parseMatch := func(t *testing.T, selectors []string) [][]metricsql.LabelFilter {
-		t.Helper()
-		var match [][]metricsql.LabelFilter
-		for _, s := range selectors {
-			expr, err := metricsql.Parse(s)
-			if err != nil {
-				t.Fatalf("failed to parse selector %q: %v", s, err)
-			}
-			me, ok := expr.(*metricsql.MetricExpr)
-			if !ok {
-				t.Fatalf("expected MetricExpr for %q, got %T", s, expr)
-			}
-			match = append(match, me.LabelFilterss...)
-		}
-		return match
-	}
-
-	f := func(t *testing.T, selectors []string, labels map[string]string, wantMatch bool) {
-		t.Helper()
-		rf := &rulesFilter{
-			gf:    &groupsFilter{},
-			match: parseMatch(t, selectors),
-		}
-		r := &rule.ApiRule{Labels: labels}
-		got := rf.matchesRule(r)
-		if got != wantMatch {
-			t.Fatalf("matchesRule(%v) with selectors %v: got %v, want %v",
-				labels, selectors, got, wantMatch)
-		}
-	}
-
-	f(t, nil, map[string]string{"foo": "bar"}, true)
-
-	f(t, []string{`{foo="bar"}`}, map[string]string{"foo": "bar"}, true)
-	f(t, []string{`{foo="bar"}`}, map[string]string{"foo": "baz"}, false)
-
-	f(t, []string{`{foo="bar"}`}, map[string]string{"bar": "baz"}, false)
-	f(t, []string{`{foo=""}`}, map[string]string{"bar": "baz"}, true)
-
-	f(t, []string{`{foo!="bar"}`}, map[string]string{"foo": "baz"}, true)
-	f(t, []string{`{foo!="bar"}`}, map[string]string{"foo": "bar"}, false)
-
-	f(t, []string{`{foo=~"bar.*"}`}, map[string]string{"foo": "bar"}, true)
-	f(t, []string{`{foo=~"bar.*"}`}, map[string]string{"foo": "baz"}, false)
-	f(t, []string{`{bar=~"baz|bar"}`}, map[string]string{"bar": "baz"}, true)
-	f(t, []string{`{bar=~"baz|bar"}`}, map[string]string{"bar": "bar"}, true)
-	f(t, []string{`{bar=~"baz|bar"}`}, map[string]string{"bar": "foo"}, false)
-
-	f(t, []string{`{foo!~"bar.*"}`}, map[string]string{"foo": "baz"}, true)
-	f(t, []string{`{foo!~"bar.*"}`}, map[string]string{"foo": "bar"}, false)
-
-	// single match[] with multiple filters
-	f(t,
-		[]string{`{job="foo",instance="bar"}`},
-		map[string]string{"job": "foo", "instance": "bar"},
-		true,
-	)
-	f(t,
-		[]string{`{job="foo",instance="bar"}`},
-		map[string]string{"job": "other", "instance": "bar"},
-		false,
-	)
-
-	f(t,
-		[]string{`{foo="bar",baz=~"b.*"}`},
-		map[string]string{"foo": "bar", "baz": "bazinga"},
-		true,
-	)
-	f(t,
-		[]string{`{foo="bar",baz=~"b.*"}`},
-		map[string]string{"foo": "other", "baz": "bazinga"},
-		false,
-	)
-
-	// multiple matches[]
-	f(t,
-		[]string{`{foo="bar"}`, `{foo="baz"}`},
-		map[string]string{"foo": "baz"},
-		true,
-	)
-	f(t,
-		[]string{`{foo="bar"}`, `{foo="baz"}`},
-		map[string]string{"foo": "unknown"},
-		false,
-	)
-	f(t,
-		[]string{`{foo=~"bar.*"}`, `{bar=~"baz.*"}`},
-		map[string]string{"bar": "bazinga"},
-		true,
-	)
-	f(t,
-		[]string{`{foo=~"bar.*"}`, `{bar=~"baz.*"}`},
-		map[string]string{"foo": "bartender"},
-		true,
-	)
-	f(t,
-		[]string{`{foo=~"bar.*"}`, `{bar=~"baz.*"}`},
-		map[string]string{"foo": "other", "bar": "other"},
-		false,
-	)
-	f(t,
-		[]string{`{job="foo",instance="bar"}`, `{foo="bar"}`},
-		map[string]string{"foo": "bar"},
-		true,
-	)
-	f(t,
-		[]string{`{job="foo", instance="bar"}`, `{foo="bar"}`},
-		map[string]string{"instance": "barr", "job": "foo"},
-		false,
-	)
-}

app/vmauth/auth_config.go

@@ -610,7 +610,6 @@ func areEqualBackendURLs(a, b []*backendURL) bool {
 }
 
 // getFirstAvailableBackendURL returns the first available backendURL, which isn't broken.
-// If all backendURLs are broken, then returns the first backendURL.
 //
 // backendURL.put() must be called on the returned backendURL after the request is complete.
 func getFirstAvailableBackendURL(bus []*backendURL) *backendURL {
@@ -629,22 +628,21 @@ func getFirstAvailableBackendURL(bus []*backendURL) *backendURL {
 			return bu
 		}
 	}
-
-	// All backend urls are unavailable, then returning a first one, it could help increase the success rate of the requests。
-	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10837#issuecomment-4307050980.
-	bu.get()
-	return bu
+	return nil
 }
 
 // getLeastLoadedBackendURL returns a non-broken backendURL with the lowest number of concurrent requests.
-// If all backendURLs are broken, then returns the first backendURL.
 //
 // backendURL.put() must be called on the returned backendURL after the request is complete.
 func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *backendURL {
-	firstBu := bus[0]
 	if len(bus) == 1 {
-		firstBu.get()
-		return firstBu
+		// Fast path - return the only backend url.
+		bu := bus[0]
+		if bu.isBroken() {
+			return nil
+		}
+		bu.get()
+		return bu
 	}
 
 	// Slow path - select other backend urls.
@@ -682,10 +680,7 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 	}
 	buMin := bus[buMinIdx]
 	if buMin.isBroken() {
-		// If all backendURLs are broken, then returns the first backendURL.
-		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/10837#issuecomment-4307050980.
-		firstBu.get()
-		return firstBu
+		return nil
 	}
 	buMin.get()
 	atomicCounter.CompareAndSwap(n+1, buMinIdx+1)
@@ -889,8 +884,7 @@ func reloadAuthConfig() (bool, error) {
 	}
 
 	mp := authUsers.Load()
-	jwtc := jwtAuthCache.Load()
-	logger.Infof("loaded information about %d users from -auth.config=%q", len(*mp)+len(jwtc.users), *authConfigPath)
+	logger.Infof("loaded information about %d users from -auth.config=%q", len(*mp), *authConfigPath)
 	return true, nil
 }
 

app/vmauth/auth_config_test.go

@@ -1031,33 +1031,6 @@ func TestLogRequest(t *testing.T) {
 	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 TestGetFirstAvailableBackend(t *testing.T) {
-	f := func(broken []bool, expectedIdx int) {
-		t.Helper()
-		bus := make([]*backendURL, len(broken))
-		for i := range broken {
-			bus[i] = &backendURL{
-				url: &url.URL{Host: fmt.Sprintf("server-%d", i)},
-			}
-			bus[i].broken.Store(broken[i])
-		}
-		bu := getFirstAvailableBackendURL(bus)
-		if bu == nil {
-			t.Fatalf("unexpected nil backend")
-		}
-		if bu.url.Host != fmt.Sprintf("server-%d", expectedIdx) {
-			t.Fatalf("unexpected backend, expected server-%d, got %s", expectedIdx, bu.url.Host)
-		}
-	}
-
-	f([]bool{false, false, false}, 0)
-	f([]bool{true, true, false}, 2)
-	// all backend are broken, then return the first one.
-	f([]bool{true, true, true}, 0)
-	f([]bool{true}, 0)
-
-}
-
 func getRegexs(paths []string) []*Regex {
 	var sps []*Regex
 	for _, path := range paths {

app/vmauth/example_config.yml

@@ -130,16 +130,6 @@ users:
   - "http://vmselect1:8481/select/{{.MetricsTenant}}/prometheus"
   - "http://vmselect2:8481/select/{{.MetricsTenant}}/prometheus"
 
-# JWT-based routing using header-based tenant identification (VictoriaMetrics cluster)
-# The AccountID and ProjectID from JWT vm_access claims are injected as HTTP headers.
-- name: jwt-header-tenant
-  jwt:
-    skip_verify: true
-  headers:
-  - "AccountID: {{.MetricsAccountID}}"
-  - "ProjectID: {{.MetricsProjectID}}"
-  url_prefix: "http://vminsert:8480/insert/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.

app/vmauth/jwt.go

@@ -17,8 +17,6 @@ import (
 
 const (
 	metricsTenantPlaceholder       = `{{.MetricsTenant}}`
-	metricsAccountIDPlaceholder    = `{{.MetricsAccountID}}`
-	metricsProjectIDPlaceholder    = `{{.MetricsProjectID}}`
 	metricsExtraLabelsPlaceholder  = `{{.MetricsExtraLabels}}`
 	metricsExtraFiltersPlaceholder = `{{.MetricsExtraFilters}}`
 
@@ -32,8 +30,6 @@ const (
 
 var allPlaceholders = []string{
 	metricsTenantPlaceholder,
-	metricsAccountIDPlaceholder,
-	metricsProjectIDPlaceholder,
 	metricsExtraLabelsPlaceholder,
 	metricsExtraFiltersPlaceholder,
 	logsAccountIDPlaceholder,
@@ -44,8 +40,6 @@ var allPlaceholders = []string{
 
 var urlPathPlaceHolders = []string{
 	metricsTenantPlaceholder,
-	metricsAccountIDPlaceholder,
-	metricsProjectIDPlaceholder,
 	logsAccountIDPlaceholder,
 	logsProjectIDPlaceholder,
 }
@@ -377,8 +371,6 @@ 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)},
-		metricsAccountIDPlaceholder:    {fmt.Sprintf("%d", vma.MetricsAccountID)},
-		metricsProjectIDPlaceholder:    {fmt.Sprintf("%d", vma.MetricsProjectID)},
 		metricsExtraLabelsPlaceholder:  vma.MetricsExtraLabels,
 		metricsExtraFiltersPlaceholder: vma.MetricsExtraFilters,
 

app/vmauth/jwt_test.go

@@ -170,13 +170,13 @@ users:
   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 URL path
+	// 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}}, {{.MetricsAccountID}}, {{.MetricsProjectID}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
+		"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(`
@@ -187,7 +187,7 @@ users:
     - "AccountID: {{.UnsupportedPlaceholder}}"
   url_prefix: http://foo.bar
 `,
-		"request header: \"AccountID\" has unsupported placeholder: \"{{.UnsupportedPlaceholder}}\", supported values are: {{.MetricsTenant}}, {{.MetricsAccountID}}, {{.MetricsProjectID}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
+		"request header: \"AccountID\" has unsupported placeholder: \"{{.UnsupportedPlaceholder}}\", supported values are: {{.MetricsTenant}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
 	)
 
 	// spaces in templating not allowed
@@ -199,19 +199,7 @@ users:
     - "AccountID: {{ .LogsAccountID }}"
   url_prefix: http://foo.bar
 `,
-		"request header: \"AccountID\" has unsupported placeholder: \"{{ .LogsAccountID }}\", supported values are: {{.MetricsTenant}}, {{.MetricsAccountID}}, {{.MetricsProjectID}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
-	)
-
-	// placeholder must match the entire header value
-	f(`
-users:
-- jwt: 
-    skip_verify: true
-  headers:
-    - "AccountID: foo {{.MetricsAccountID}}"
-  url_prefix: http://foo.bar
-`,
-		"request header: \"AccountID\" has unsupported placeholder: \"foo {{.MetricsAccountID}}\", supported values are: {{.MetricsTenant}}, {{.MetricsAccountID}}, {{.MetricsProjectID}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
+		"request header: \"AccountID\" has unsupported placeholder: \"{{ .LogsAccountID }}\", supported values are: {{.MetricsTenant}}, {{.MetricsExtraLabels}}, {{.MetricsExtraFilters}}, {{.LogsAccountID}}, {{.LogsProjectID}}, {{.LogsExtraFilters}}, {{.LogsExtraStreamFilters}}",
 	)
 
 	// oidc is not an object
@@ -376,25 +364,10 @@ users:
   url_prefix: http://foo.bar
 `, validRSAPublicKey, validECDSAPublicKey))
 
-	// metrics header placeholders
 	f(`
 users:
 - jwt:
     skip_verify: true
-  headers:
-  - "MetricsAccountID: {{.MetricsAccountID}}"
-  - "MetricsProjectID: {{.MetricsProjectID}}"
-  url_prefix: http://foo.bar
-`)
-
-	// logs header placeholders
-	f(`
-users:
-- jwt:
-    skip_verify: true
-  headers:
-  - "LogsAccountID: {{.LogsAccountID}}"
-  - "LogsProjectID: {{.LogsProjectID}}"
   url_prefix: http://foo.bar
 `)
 

app/vmauth/main.go

@@ -51,7 +51,7 @@ var (
 		"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 retries. "+
+		"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. "+
@@ -317,7 +317,7 @@ func processUserRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo, tk
 	defer ui.endConcurrencyLimit()
 
 	// Process the request.
-	processRequest(w, r, ui, tkn, userName)
+	processRequest(w, r, ui, tkn)
 }
 
 func beginConcurrencyLimit(ctx context.Context) error {
@@ -391,7 +391,7 @@ func bufferRequestBody(ctx context.Context, r io.ReadCloser, userName string) (i
 	return bb, nil
 }
 
-func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo, tkn *jwt.Token, userName string) {
+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
@@ -409,7 +409,7 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo, tkn *j
 			if ui.DumpRequestOnErrors {
 				di = debugInfo(u, r)
 			}
-			httpserver.Errorf(w, r, "user %s missing route for %q%s", userName, u.String(), di)
+			httpserver.Errorf(w, r, "missing route for %q%s", u.String(), di)
 			return
 		}
 		up, hc = ui.DefaultURL, ui.HeadersConf
@@ -455,7 +455,7 @@ func processRequest(w http.ResponseWriter, r *http.Request, ui *UserInfo, tkn *j
 		ui.backendErrors.Inc()
 	}
 	err := &httpserver.ErrorWithStatusCode{
-		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(), userName),
+		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)
@@ -850,18 +850,14 @@ func (bb *bufferedBody) Read(p []byte) (int, error) {
 }
 
 func (bb *bufferedBody) canRetry() bool {
-	if bb.r != nil {
-		return false
-	}
-	maxRetrySize := maxRequestBodySizeToRetry.IntN()
-	return len(bb.buf) == 0 || (maxRetrySize > 0 && len(bb.buf) <= maxRetrySize)
+	return bb.r == nil
 }
 
 // Close implements io.Closer interface.
 func (bb *bufferedBody) Close() error {
 	bb.resetReader()
-	bb.cannotRetry = !bb.canRetry()
 	if bb.r != nil {
+		bb.cannotRetry = true
 		return bb.r.Close()
 	}
 	return nil

app/vmauth/main_test.go

@@ -19,7 +19,6 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
-	"strconv"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -307,24 +306,6 @@ statusCode=200
 requested_url={BACKEND}/bar/a/b`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 
-	// correct authorization but unexisted path, hence missing route error.
-	cfgStr = `
-users:
-- username: foo
-  password: secret
-  url_map:
-  - src_paths:
-    - "/api/v1/write"
-    url_prefix: "{BACKEND}/bar"`
-	requestURL = "http://foo:secret@some-host.com/a/b"
-	backendHandler = func(w http.ResponseWriter, r *http.Request) {
-		fmt.Fprintf(w, "requested_url=http://%s%s", r.Host, r.URL)
-	}
-	responseExpected = `
-statusCode=400
-user foo missing route for "http://foo:secret@some-host.com/a/b"`
-	f(cfgStr, requestURL, backendHandler, responseExpected)
-
 	// verify how path cleanup works
 	cfgStr = `
 unauthorized_user:
@@ -421,7 +402,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=400
-user unauthorized missing route for "http://some-host.com/abc?de=fg"`
+missing route for "http://some-host.com/abc?de=fg"`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 
 	// missing default_url and default url_prefix for unauthorized user with dump_request_on_errors enabled
@@ -437,7 +418,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=400
-user unauthorized missing route for "http://some-host.com/abc?de=fg" (host: "some-host.com"; path: "/abc"; args: "de=fg"; headers:Connection: Some-Header,Other-Header
+missing route for "http://some-host.com/abc?de=fg" (host: "some-host.com"; path: "/abc"; args: "de=fg"; headers:Connection: Some-Header,Other-Header
 Pass-Header: abc
 Some-Header: foobar
 X-Forwarded-For: 12.34.56.78
@@ -479,7 +460,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=502
-all the 2 backends for the user "unauthorized" are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend`
+all the 2 backends for the user "" are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 
 	// all the backend_urls are unavailable for authorized user
@@ -519,7 +500,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=502
-all the 0 backends for the user "unauthorized" are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend`
+all the 0 backends for the user "" are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 	netutil.Resolver = origResolver
 
@@ -536,7 +517,7 @@ unauthorized_user:
 	}
 	responseExpected = `
 statusCode=502
-all the 2 backends for the user "unauthorized" are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend`
+all the 2 backends for the user "" are unavailable for proxying the request - check previous WARN logs to see the exact error for each failed backend`
 	f(cfgStr, requestURL, backendHandler, responseExpected)
 	if n := retries.Load(); n != 2 {
 		t.Fatalf("unexpected number of retries; got %d; want 2", n)
@@ -563,31 +544,6 @@ requested_url={BACKEND}/path2/foo/?de=fg`
 	if n := retries.Load(); n != 2 {
 		t.Fatalf("unexpected number of retries; got %d; want 2", n)
 	}
-
-	// make sure that empty config value erases client extra filters and extra labels
-	cfgStr = `
-unauthorized_user:
-  url_prefix: {BACKEND}/foo?bar=baz&extra_filters[]=&extra_label=&extra_filters=`
-	requestURL = "http://some-host.com/abc/def?some_arg=some_value&extra_filters[]=baz&extra_label=tenant=admin&extra_filters=bar"
-	backendHandler = func(w http.ResponseWriter, r *http.Request) {
-		h := w.Header()
-		h.Set("Connection", "close")
-		h.Set("Foo", "bar")
-
-		var bb bytes.Buffer
-		if err := r.Header.Write(&bb); err != nil {
-			panic(fmt.Errorf("unexpected error when marshaling headers: %w", err))
-		}
-		fmt.Fprintf(w, "requested_url=http://%s%s\n%s", r.Host, r.URL, bb.String())
-	}
-	responseExpected = `
-statusCode=200
-Foo: bar
-requested_url={BACKEND}/foo/abc/def?bar=baz&extra_filters=&extra_filters%5B%5D=&extra_label=&some_arg=some_value
-Pass-Header: abc
-User-Agent: vmauth
-X-Forwarded-For: 12.34.56.78, 42.2.3.84`
-	f(cfgStr, requestURL, backendHandler, responseExpected)
 }
 
 func TestJWTRequestHandler(t *testing.T) {
@@ -894,30 +850,6 @@ users:
 		responseExpected,
 	)
 
-	// test header injection and URL templating with individual placeholders
-	request = httptest.NewRequest(`GET`, "http://some-host.com/api/v1/query", nil)
-	request.Header.Set(`Authorization`, `Bearer `+fullToken)
-	responseExpected = `
-statusCode=200
-path: /select/123/234/api/v1/query
-query:
-headers:
-    AccountID=123
-    ProjectID=234`
-	f(fmt.Sprintf(
-		`
-users:
-- jwt:
-    public_keys:
-    - %q
-  url_prefix: {BACKEND}/select/{{.MetricsAccountID}}/{{.MetricsProjectID}}
-  headers:
-  - "AccountID: {{.MetricsAccountID}}"
-  - "ProjectID: {{.MetricsProjectID}}"`, string(publicKeyPEM)),
-		request,
-		responseExpected,
-	)
-
 	// extra_label and extra_filters from vm_access claim merged with statically defined
 	request = httptest.NewRequest(`GET`, "http://some-host.com/api/v1/query", nil)
 	request.Header.Set(`Authorization`, `Bearer `+fullToken)
@@ -1639,7 +1571,7 @@ func (w *fakeResponseWriter) WriteHeader(statusCode int) {
 		"X-Content-Type-Options": true,
 	})
 	if err != nil {
-		panic(fmt.Errorf("cannot marshal headers: %w", err))
+		panic(fmt.Errorf("cannot marshal headers: %s", err))
 	}
 }
 
@@ -1899,7 +1831,7 @@ func (r *mockBody) Read(p []byte) (n int, err error) {
 }
 
 func TestBufferedBody_RetrySuccess(t *testing.T) {
-	f := func(s string, maxSizeToRetry, bufferSize int) {
+	f := func(s string, maxBodySize int) {
 		t.Helper()
 
 		defaultRequestBufferSize := requestBufferSize.String()
@@ -1908,7 +1840,7 @@ func TestBufferedBody_RetrySuccess(t *testing.T) {
 				t.Fatalf("cannot reset requestBufferSize: %s", err)
 			}
 		}()
-		if err := requestBufferSize.Set(strconv.Itoa(bufferSize)); err != nil {
+		if err := requestBufferSize.Set(fmt.Sprintf("%d", maxBodySize)); err != nil {
 			t.Fatalf("cannot set requestBufferSize: %s", err)
 		}
 
@@ -1918,7 +1850,7 @@ func TestBufferedBody_RetrySuccess(t *testing.T) {
 				t.Fatalf("cannot reset maxRequestBodySizeToRetry: %s", err)
 			}
 		}()
-		if err := maxRequestBodySizeToRetry.Set(strconv.Itoa(maxSizeToRetry)); err != nil {
+		if err := maxRequestBodySizeToRetry.Set("0"); err != nil {
 			t.Fatalf("cannot set maxRequestBodySizeToRetry: %s", err)
 		}
 
@@ -1947,20 +1879,16 @@ func TestBufferedBody_RetrySuccess(t *testing.T) {
 		}
 	}
 
-	f("", 0, 2000)
-	f("", 0, 0)
-	f("", -1, 2000)
-	f("", 100, 2000)
-	f("foo", 100, 2000)
-	f("foobar", 100, 2000)
-	f("foobar", 100, 0)
-	f("foobar", 100, -1)
-	f(newTestString(1000), 1001, 2000)
-	f(newTestString(1000), 1001, 500)
+	f("", 0)
+	f("", -1)
+	f("", 100)
+	f("foo", 100)
+	f("foobar", 100)
+	f(newTestString(1000), 1001)
 }
 
 func TestBufferedBody_RetrySuccessPartialRead(t *testing.T) {
-	f := func(s string, maxSizeToRetry, bufferSize int) {
+	f := func(s string, maxBodySize int) {
 		t.Helper()
 
 		// Check the case with partial read
@@ -1970,7 +1898,7 @@ func TestBufferedBody_RetrySuccessPartialRead(t *testing.T) {
 				t.Fatalf("cannot reset requestBufferSize: %s", err)
 			}
 		}()
-		if err := requestBufferSize.Set(strconv.Itoa(bufferSize)); err != nil {
+		if err := requestBufferSize.Set(fmt.Sprintf("%d", maxBodySize)); err != nil {
 			t.Fatalf("cannot set requestBufferSize: %s", err)
 		}
 
@@ -1980,7 +1908,7 @@ func TestBufferedBody_RetrySuccessPartialRead(t *testing.T) {
 				t.Fatalf("cannot reset maxRequestBodySizeToRetry: %s", err)
 			}
 		}()
-		if err := maxRequestBodySizeToRetry.Set(strconv.Itoa(maxSizeToRetry)); err != nil {
+		if err := maxRequestBodySizeToRetry.Set("0"); err != nil {
 			t.Fatalf("cannot set maxRequestBodySizeToRetry: %s", err)
 		}
 
@@ -2024,20 +1952,16 @@ func TestBufferedBody_RetrySuccessPartialRead(t *testing.T) {
 		}
 	}
 
-	f("", 0, 2000)
-	f("", 0, 0)
-	f("", -1, 2000)
-	f("", 100, 2000)
-	f("foo", 100, 2000)
-	f("foobar", 100, 2000)
-	f("foobar", 100, 0)
-	f("foobar", 100, -1)
-	f(newTestString(1000), 1001, 2000)
-	f(newTestString(1000), 1001, 500)
+	f("", 0)
+	f("", -1)
+	f("", 100)
+	f("foo", 100)
+	f("foobar", 100)
+	f(newTestString(1000), 1001)
 }
 
 func TestBufferedBody_RetryFailureTooBigBody(t *testing.T) {
-	f := func(s string, maxSizeToRetry, bufferSize int) {
+	f := func(s string, maxBodySize int) {
 		t.Helper()
 
 		defaultRequestBufferSize := requestBufferSize.String()
@@ -2046,7 +1970,7 @@ func TestBufferedBody_RetryFailureTooBigBody(t *testing.T) {
 				t.Fatalf("cannot reset requestBufferSize: %s", err)
 			}
 		}()
-		if err := requestBufferSize.Set(strconv.Itoa(bufferSize)); err != nil {
+		if err := requestBufferSize.Set("0"); err != nil {
 			t.Fatalf("cannot set requestBufferSize: %s", err)
 		}
 
@@ -2056,7 +1980,7 @@ func TestBufferedBody_RetryFailureTooBigBody(t *testing.T) {
 				t.Fatalf("cannot reset maxRequestBodySizeToRetry: %s", err)
 			}
 		}()
-		if err := maxRequestBodySizeToRetry.Set(strconv.Itoa(maxSizeToRetry)); err != nil {
+		if err := maxRequestBodySizeToRetry.Set(fmt.Sprintf("%d", maxBodySize)); err != nil {
 			t.Fatalf("cannot set maxRequestBodySizeToRetry: %s", err)
 		}
 
@@ -2101,17 +2025,12 @@ func TestBufferedBody_RetryFailureTooBigBody(t *testing.T) {
 	}
 
 	const maxBodySize = 1000
-	f(newTestString(maxBodySize+1), 0, 2*maxBodySize)
-	f(newTestString(maxBodySize+1), -1, 2*maxBodySize)
-	f(newTestString(maxBodySize+1), maxBodySize, 0)
-	f(newTestString(maxBodySize+1), maxBodySize, -1)
-	f(newTestString(maxBodySize+1), maxBodySize, maxBodySize)
-	f(newTestString(maxBodySize+1), maxBodySize, 2*maxBodySize)
-	f(newTestString(2*maxBodySize), maxBodySize, 0)
+	f(newTestString(maxBodySize+1), maxBodySize)
+	f(newTestString(2*maxBodySize), maxBodySize)
 }
 
-func TestBufferedBody_RetryDisabledByMaxRequestBodySizeToRetry(t *testing.T) {
-	f := func(s string, maxSizeToRetry, bufferSize int) {
+func TestBufferedBody_RetryFailureZeroOrNegativeMaxBodySize(t *testing.T) {
+	f := func(s string, maxBodySize int) {
 		t.Helper()
 
 		defaultRequestBufferSize := requestBufferSize.String()
@@ -2120,20 +2039,10 @@ func TestBufferedBody_RetryDisabledByMaxRequestBodySizeToRetry(t *testing.T) {
 				t.Fatalf("cannot reset requestBufferSize: %s", err)
 			}
 		}()
-		if err := requestBufferSize.Set(strconv.Itoa(bufferSize)); err != nil {
+		if err := requestBufferSize.Set(fmt.Sprintf("%d", maxBodySize)); err != nil {
 			t.Fatalf("cannot set requestBufferSize: %s", err)
 		}
 
-		defaultMaxRequestBodySizeToRetry := maxRequestBodySizeToRetry.String()
-		defer func() {
-			if err := maxRequestBodySizeToRetry.Set(defaultMaxRequestBodySizeToRetry); err != nil {
-				t.Fatalf("cannot reset maxRequestBodySizeToRetry: %s", err)
-			}
-		}()
-		if err := maxRequestBodySizeToRetry.Set(strconv.Itoa(maxSizeToRetry)); err != nil {
-			t.Fatalf("cannot set maxRequestBodySizeToRetry: %s", err)
-		}
-
 		ctx := context.Background()
 		rb, err := bufferRequestBody(ctx, io.NopCloser(bytes.NewBufferString(s)), "foo")
 		if err != nil {
@@ -2142,8 +2051,8 @@ func TestBufferedBody_RetryDisabledByMaxRequestBodySizeToRetry(t *testing.T) {
 		bb, ok := rb.(*bufferedBody)
 		canRetry := !ok || bb.canRetry()
 
-		if canRetry {
-			t.Fatalf("canRetry() must return false before reading anything")
+		if !canRetry {
+			t.Fatalf("canRetry() must return true before reading anything")
 		}
 		data, err := io.ReadAll(rb)
 		if err != nil {
@@ -2157,19 +2066,19 @@ func TestBufferedBody_RetryDisabledByMaxRequestBodySizeToRetry(t *testing.T) {
 		}
 
 		data, err = io.ReadAll(rb)
-		if err == nil {
-			t.Fatalf("expecting non-nil error")
+		if err != nil {
+			t.Fatalf("unexpected error in io.ReadAll: %s", err)
 		}
-		if len(data) != 0 {
-			t.Fatalf("unexpected non-empty data read: %q", data)
+		if string(data) != s {
+			t.Fatalf("unexpected data read\ngot\n%s\nwant\n%s", data, s)
 		}
 	}
 
-	f("foobar", 0, 2048)
-	f(newTestString(1000), 0, 2048)
+	f("foobar", 0)
+	f(newTestString(1000), 0)
 
-	f("foobar", -1, 2048)
-	f(newTestString(1000), -1, 2048)
+	f("foobar", -1)
+	f(newTestString(1000), -1)
 }
 
 func newTestString(sLen int) string {

app/vmauth/oidc.go

@@ -161,7 +161,7 @@ func fetchAndParseJWKs(ctx context.Context, jwksURI string) (*jwt.VerifierPool,
 
 	vp, err := jwt.ParseJWKs(b)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse jwks keys from %q: %w", jwksURI, err)
+		return nil, fmt.Errorf("failed to parse jwks keys from %q: %v", jwksURI, err)
 	}
 
 	return vp, nil
@@ -188,7 +188,7 @@ func getOpenIDConfiguration(ctx context.Context, issuer string) (openidConfig, e
 
 	var cfg openidConfig
 	if err := json.NewDecoder(resp.Body).Decode(&cfg); err != nil {
-		return openidConfig{}, fmt.Errorf("failed to decode openid config from %q: %w", configURL, err)
+		return openidConfig{}, fmt.Errorf("failed to decode openid config from %q: %s", configURL, err)
 	}
 
 	return cfg, nil

app/vmctl/flags.go

@@ -146,8 +146,7 @@ var (
 			Name:  vmRoundDigits,
 			Value: 100,
 			Usage: "Round metric values to the given number of decimal digits after the point. " +
-				"This option may be used for increasing on-disk compression level for the stored metrics. " +
-				"See also --vm-significant-figures option",
+				"This option may be used for increasing on-disk compression level for the stored metrics",
 		},
 		&cli.StringSliceFlag{
 			Name:  vmExtraLabel,
@@ -501,96 +500,6 @@ var (
 	}
 )
 
-const (
-	mimirPath             = "mimir-path"
-	mimirTenantID         = "mimir-tenant-id"
-	mimirConcurrency      = "mimir-concurrency"
-	mimirFilterTimeStart  = "mimir-filter-time-start"
-	mimirFilterTimeEnd    = "mimir-filter-time-end"
-	mimirFilterLabel      = "mimir-filter-label"
-	mimirFilterLabelValue = "mimir-filter-label-value"
-
-	mimirCredsFilePath           = "mimir-creds-file-path"
-	mimirConfigFilePath          = "mimir-config-file-path"
-	mimirConfigProfile           = "mimir-config-profile"
-	mimirCustomS3Endpoint        = "mimir-custom-s3-endpoint"
-	mimirS3ForcePathStyle        = "mimir-s3-force-path-style"
-	mimirS3TLSInsecureSkipVerify = "mimir-s3-tls-insecure-skip-verify"
-	mimirSSEKMSKeyID             = "mimir-s3-sse-kms-key-id"
-	mimirSSEAlgorithm            = "mimir-s3-sse-algorithm"
-)
-
-var (
-	mimirFlags = []cli.Flag{
-		&cli.StringFlag{
-			Name:     mimirPath,
-			Usage:    "Path to Mimir storage bucket or local folder.",
-			Required: true,
-		},
-		&cli.StringFlag{
-			Name:  mimirTenantID,
-			Usage: "Tenant ID for Mimir storage",
-		},
-		&cli.IntFlag{
-			Name:  mimirConcurrency,
-			Usage: "Number of concurrently running block readers",
-			Value: 1,
-		},
-		&cli.StringFlag{
-			Name:     mimirFilterTimeStart,
-			Usage:    "The time filter in RFC3339 format to select timeseries with timestamp equal or higher than provided value. E.g. '2020-01-01T20:07:00Z'",
-			Required: true,
-		},
-		&cli.StringFlag{
-			Name:     mimirFilterTimeEnd,
-			Usage:    "The time filter in RFC3339 format to select timeseries with timestamp equal or lower than provided value. E.g. '2020-01-01T20:07:00Z'",
-			Required: true,
-		},
-		&cli.StringFlag{
-			Name:  mimirFilterLabel,
-			Usage: "Mimir label name to filter timeseries by. E.g. '__name__' will filter timeseries by name.",
-		},
-		&cli.StringFlag{
-			Name:  mimirFilterLabelValue,
-			Usage: fmt.Sprintf("Regular expression to filter label from %q flag.", mimirFilterLabel),
-			Value: ".*",
-		},
-		&cli.StringFlag{
-			Name:  mimirCredsFilePath,
-			Usage: "Path to file with GCS or S3 credentials. Credentials are loaded from default locations if not set. See https://cloud.google.com/iam/docs/creating-managing-service-account-keys and https://docs.aws.amazon.com/general/latest/gr/aws-security-credentials.html",
-		},
-		&cli.StringFlag{
-			Name:  mimirConfigFilePath,
-			Usage: "Path to file with S3 configs. Configs are loaded from default location if not set. See https://docs.aws.amazon.com/general/latest/gr/aws-security-credentials.html",
-		},
-		&cli.StringFlag{
-			Name:  mimirConfigProfile,
-			Usage: "Profile name for S3 configs. If no set, the value of the environment variable will be loaded (AWS_PROFILE or AWS_DEFAULT_PROFILE), or if both not set, DefaultSharedConfigProfile is used",
-		},
-		&cli.StringFlag{
-			Name:  mimirCustomS3Endpoint,
-			Usage: "Custom S3 endpoint for use with S3-compatible storages (e.g. MinIO). S3 is used if not set",
-		},
-		&cli.BoolFlag{
-			Name:  mimirS3ForcePathStyle,
-			Usage: "Prefixing endpoint with bucket name when set false, true by default.",
-			Value: true,
-		},
-		&cli.BoolFlag{
-			Name:  mimirS3TLSInsecureSkipVerify,
-			Usage: "Whether to skip TLS verification when connecting to the S3 endpoint.",
-		},
-		&cli.StringFlag{
-			Name:  mimirSSEKMSKeyID,
-			Usage: "SSE KMS Key ID for use with S3-compatible storages.",
-		},
-		&cli.StringFlag{
-			Name:  mimirSSEAlgorithm,
-			Usage: "SSE algorithm for use with S3-compatible storages.",
-		},
-	}
-)
-
 const (
 	vmNativeFilterMatch       = "vm-native-filter-match"
 	vmNativeFilterTimeStart   = "vm-native-filter-time-start"

app/vmctl/influx.go

@@ -43,7 +43,7 @@ func newInfluxProcessor(ic *influx.Client, im *vm.Importer, cc int, separator st
 func (ip *influxProcessor) run(ctx context.Context) error {
 	series, err := ip.ic.Explore()
 	if err != nil {
-		return fmt.Errorf("explore query failed: %w", err)
+		return fmt.Errorf("explore query failed: %s", err)
 	}
 	if len(series) < 1 {
 		return fmt.Errorf("found no timeseries to import")
@@ -71,7 +71,7 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 			for s := range seriesCh {
 				if err := ip.do(s); err != nil {
 					influxErrorsTotal.Inc()
-					errCh <- fmt.Errorf("request failed for %q.%q: %w", s.Measurement, s.Field, err)
+					errCh <- fmt.Errorf("request failed for %q.%q: %s", s.Measurement, s.Field, err)
 					return
 				}
 				influxSeriesProcessed.Inc()
@@ -84,10 +84,10 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 	for _, s := range series {
 		select {
 		case infErr := <-errCh:
-			return fmt.Errorf("influx error: %w", infErr)
+			return fmt.Errorf("influx error: %s", infErr)
 		case vmErr := <-ip.im.Errors():
 			influxErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, ip.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, ip.isVerbose))
 		case seriesCh <- s:
 		}
 	}
@@ -100,11 +100,11 @@ func (ip *influxProcessor) run(ctx context.Context) error {
 	for vmErr := range ip.im.Errors() {
 		if vmErr.Err != nil {
 			influxErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, ip.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, ip.isVerbose))
 		}
 	}
 	for err := range errCh {
-		return fmt.Errorf("import process failed: %w", err)
+		return fmt.Errorf("import process failed: %s", err)
 	}
 
 	log.Println("Import finished!")
@@ -119,7 +119,7 @@ const valueField = "value"
 func (ip *influxProcessor) do(s *influx.Series) error {
 	cr, err := ip.ic.FetchDataPoints(s)
 	if err != nil {
-		return fmt.Errorf("failed to fetch datapoints: %w", err)
+		return fmt.Errorf("failed to fetch datapoints: %s", err)
 	}
 	defer func() {
 		_ = cr.Close()

app/vmctl/influx/influx.go

@@ -96,10 +96,10 @@ func NewClient(cfg Config) (*Client, error) {
 	}
 	hc, err := influx.NewHTTPClient(c)
 	if err != nil {
-		return nil, fmt.Errorf("failed to establish conn: %w", err)
+		return nil, fmt.Errorf("failed to establish conn: %s", err)
 	}
 	if _, _, err := hc.Ping(time.Second); err != nil {
-		return nil, fmt.Errorf("ping failed: %w", err)
+		return nil, fmt.Errorf("ping failed: %s", err)
 	}
 
 	chunkSize := cfg.ChunkSize
@@ -155,7 +155,7 @@ func (c *Client) Explore() ([]*Series, error) {
 	// {"measurement1": ["value1", "value2"]}
 	mFields, err := c.fieldsByMeasurement()
 	if err != nil {
-		return nil, fmt.Errorf("failed to get field keys: %w", err)
+		return nil, fmt.Errorf("failed to get field keys: %s", err)
 	}
 
 	if len(mFields) < 1 {
@@ -165,12 +165,12 @@ func (c *Client) Explore() ([]*Series, error) {
 	// {"measurement1": {"tag1", "tag2"}}
 	measurementTags, err := c.getMeasurementTags()
 	if err != nil {
-		return nil, fmt.Errorf("failed to get tags of measurements: %w", err)
+		return nil, fmt.Errorf("failed to get tags of measurements: %s", err)
 	}
 
 	series, err := c.getSeries()
 	if err != nil {
-		return nil, fmt.Errorf("failed to get series: %w", err)
+		return nil, fmt.Errorf("failed to get series: %s", err)
 	}
 
 	var iSeries []*Series
@@ -237,7 +237,7 @@ func (cr *ChunkedResponse) Next() ([]int64, []float64, error) {
 		return nil, nil, err
 	}
 	if resp.Error() != nil {
-		return nil, nil, fmt.Errorf("response error for %s: %w", cr.iq.Command, resp.Error())
+		return nil, nil, fmt.Errorf("response error for %s: %s", cr.iq.Command, resp.Error())
 	}
 	if len(resp.Results) != 1 {
 		return nil, nil, fmt.Errorf("unexpected number of results in response: %d", len(resp.Results))
@@ -265,7 +265,8 @@ func (cr *ChunkedResponse) Next() ([]int64, []float64, error) {
 	for i, fv := range fieldValues {
 		v, err := toFloat64(fv)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to convert value %q.%v to float64: %w", cr.field, v, err)
+			return nil, nil, fmt.Errorf("failed to convert value %q.%v to float64: %s",
+				cr.field, v, err)
 		}
 		values[i] = v
 	}
@@ -293,7 +294,7 @@ func (c *Client) FetchDataPoints(s *Series) (*ChunkedResponse, error) {
 	}
 	cr, err := c.QueryAsChunk(iq)
 	if err != nil {
-		return nil, fmt.Errorf("query %q err: %w", iq.Command, err)
+		return nil, fmt.Errorf("query %q err: %s", iq.Command, err)
 	}
 	return &ChunkedResponse{cr, iq, s.Field}, nil
 }
@@ -307,7 +308,7 @@ func (c *Client) fieldsByMeasurement() (map[string][]string, error) {
 	log.Printf("fetching fields: %s", stringify(q))
 	qValues, err := c.do(q)
 	if err != nil {
-		return nil, fmt.Errorf("error while executing query %q: %w", q.Command, err)
+		return nil, fmt.Errorf("error while executing query %q: %s", q.Command, err)
 	}
 
 	var total int
@@ -351,7 +352,7 @@ func (c *Client) getSeries() ([]*Series, error) {
 	log.Printf("fetching series: %s", stringify(q))
 	cr, err := c.QueryAsChunk(q)
 	if err != nil {
-		return nil, fmt.Errorf("error while executing query %q: %w", q.Command, err)
+		return nil, fmt.Errorf("error while executing query %q: %s", q.Command, err)
 	}
 
 	const key = "key"
@@ -365,7 +366,7 @@ func (c *Client) getSeries() ([]*Series, error) {
 			return nil, err
 		}
 		if resp.Error() != nil {
-			return nil, fmt.Errorf("response error for query %q: %w", q.Command, resp.Error())
+			return nil, fmt.Errorf("response error for query %q: %s", q.Command, resp.Error())
 		}
 		qValues, err := parseResult(resp.Results[0])
 		if err != nil {
@@ -416,7 +417,7 @@ func (c *Client) getMeasurementTags() (map[string]map[string]struct{}, error) {
 	log.Printf("fetching tag keys: %s", stringify(q))
 	cr, err := c.QueryAsChunk(q)
 	if err != nil {
-		return nil, fmt.Errorf("error while executing query %q: %w", q.Command, err)
+		return nil, fmt.Errorf("error while executing query %q: %s", q.Command, err)
 	}
 
 	const tagKey = "tagKey"
@@ -431,7 +432,7 @@ func (c *Client) getMeasurementTags() (map[string]map[string]struct{}, error) {
 			return nil, err
 		}
 		if resp.Error() != nil {
-			return nil, fmt.Errorf("response error for query %q: %w", q.Command, resp.Error())
+			return nil, fmt.Errorf("response error for query %q: %s", q.Command, resp.Error())
 		}
 		qValues, err := parseResult(resp.Results[0])
 		if err != nil {
@@ -454,10 +455,10 @@ func (c *Client) getMeasurementTags() (map[string]map[string]struct{}, error) {
 func (c *Client) do(q influx.Query) ([]queryValues, error) {
 	res, err := c.Query(q)
 	if err != nil {
-		return nil, fmt.Errorf("query error: %w", err)
+		return nil, fmt.Errorf("query error: %s", err)
 	}
 	if res.Error() != nil {
-		return nil, fmt.Errorf("response error: %w", res.Error())
+		return nil, fmt.Errorf("response error: %s", res.Error())
 	}
 	if len(res.Results) < 1 {
 		return nil, fmt.Errorf("query returned 0 results")

app/vmctl/influx/parser.go

@@ -71,7 +71,7 @@ func toFloat64(v any) (float64, error) {
 func parseDate(dateStr string) (int64, error) {
 	startTime, err := time.Parse(time.RFC3339, dateStr)
 	if err != nil {
-		return 0, fmt.Errorf("cannot parse %q: %w", dateStr, err)
+		return 0, fmt.Errorf("cannot parse %q: %s", dateStr, err)
 	}
 	return startTime.UnixNano() / 1e6, nil
 }
@@ -92,7 +92,7 @@ func (s *Series) unmarshal(v string) error {
 	var err error
 	s.LabelPairs, err = unmarshalTags(v[n+1:], noEscapeChars)
 	if err != nil {
-		return fmt.Errorf("failed to unmarhsal tags: %w", err)
+		return fmt.Errorf("failed to unmarhsal tags: %s", err)
 	}
 	return nil
 }

app/vmctl/main.go

@@ -18,7 +18,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/auth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/backoff"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/mimir"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/native"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/remoteread"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -88,7 +87,7 @@ func main() {
 
 					tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_opentsdb")
 					if err != nil {
-						return fmt.Errorf("failed to create transport for -%s=%q: %w", otsdbAddr, addr, err)
+						return fmt.Errorf("failed to create transport for -%s=%q: %s", otsdbAddr, addr, err)
 					}
 					oCfg := opentsdb.Config{
 						Addr:       addr,
@@ -103,17 +102,17 @@ func main() {
 					}
 					otsdbClient, err := opentsdb.NewClient(oCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create opentsdb client: %w", err)
+						return fmt.Errorf("failed to create opentsdb client: %s", err)
 					}
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %w", err)
+						return fmt.Errorf("failed to init VM configuration: %s", err)
 					}
 
 					importer, err := vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %w", err)
+						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
 
 					otsdbProcessor := newOtsdbProcessor(otsdbClient, importer, c.Int(otsdbConcurrency), c.Bool(globalVerbose))
@@ -137,7 +136,7 @@ func main() {
 
 					tc, err := promauth.NewTLSConfig(certFile, keyFile, caFile, serverName, insecureSkipVerify)
 					if err != nil {
-						return fmt.Errorf("failed to create TLS Config: %w", err)
+						return fmt.Errorf("failed to create TLS Config: %s", err)
 					}
 
 					iCfg := influx.Config{
@@ -157,17 +156,17 @@ func main() {
 
 					influxClient, err := influx.NewClient(iCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create influx client: %w", err)
+						return fmt.Errorf("failed to create influx client: %s", err)
 					}
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %w", err)
+						return fmt.Errorf("failed to init VM configuration: %s", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %w", err)
+						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
 
 					processor := newInfluxProcessor(
@@ -203,7 +202,7 @@ func main() {
 
 					tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_remoteread")
 					if err != nil {
-						return fmt.Errorf("failed to create transport for -%s=%q: %w", remoteReadSrcAddr, addr, err)
+						return fmt.Errorf("failed to create transport for -%s=%q: %s", remoteReadSrcAddr, addr, err)
 					}
 
 					// Backwards compatible default values if none provided by user
@@ -227,17 +226,17 @@ func main() {
 						DisablePathAppend: c.Bool(remoteReadDisablePathAppend),
 					})
 					if err != nil {
-						return fmt.Errorf("error create remote read client: %w", err)
+						return fmt.Errorf("error create remote read client: %s", err)
 					}
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %w", err)
+						return fmt.Errorf("failed to init VM configuration: %s", err)
 					}
 
 					importer, err := vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %w", err)
+						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
 
 					rmp := remoteReadProcessor{
@@ -265,12 +264,12 @@ func main() {
 
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %w", err)
+						return fmt.Errorf("failed to init VM configuration: %s", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %w", err)
+						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
 
 					promCfg := prometheus.Config{
@@ -285,7 +284,7 @@ func main() {
 					}
 					cl, err := prometheus.NewClient(promCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create prometheus client: %w", err)
+						return fmt.Errorf("failed to create prometheus client: %s", err)
 					}
 
 					pp := prometheusProcessor{
@@ -297,56 +296,6 @@ func main() {
 					return pp.run(ctx)
 				},
 			},
-			{
-				Name:   "mimir",
-				Usage:  "Migrate time series from Mimir object storage or local filesystem",
-				Flags:  mergeFlags(globalFlags, mimirFlags, vmFlags),
-				Before: beforeFn,
-				Action: func(c *cli.Context) error {
-					fmt.Println("Mimir import mode")
-
-					vmCfg, err := initConfigVM(c)
-					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %w", err)
-					}
-
-					importer, err = vm.NewImporter(ctx, vmCfg)
-					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %w", err)
-					}
-
-					mCfg := mimir.Config{
-						Filter: mimir.Filter{
-							TimeMin:    c.String(mimirFilterTimeStart),
-							TimeMax:    c.String(mimirFilterTimeEnd),
-							Label:      c.String(mimirFilterLabel),
-							LabelValue: c.String(mimirFilterLabelValue),
-						},
-						Path:                    c.String(mimirPath),
-						TenantID:                c.String(mimirTenantID),
-						CredsFilePath:           c.String(mimirCredsFilePath),
-						ConfigFilePath:          c.String(mimirConfigFilePath),
-						ConfigProfile:           c.String(mimirConfigProfile),
-						CustomS3Endpoint:        c.String(mimirCustomS3Endpoint),
-						S3ForcePathStyle:        c.Bool(mimirS3ForcePathStyle),
-						S3TLSInsecureSkipVerify: c.Bool(mimirS3TLSInsecureSkipVerify),
-						SSEKMSKeyID:             c.String(mimirSSEKMSKeyID),
-						SSEAlgorithm:            c.String(mimirSSEAlgorithm),
-					}
-					cl, err := mimir.NewClient(ctx, mCfg)
-					if err != nil {
-						return fmt.Errorf("failed to create mimir client: %w", err)
-					}
-
-					pp := prometheusProcessor{
-						cl:        cl,
-						im:        importer,
-						cc:        c.Int(mimirConcurrency),
-						isVerbose: c.Bool(globalVerbose),
-					}
-					return pp.run(ctx)
-				},
-			},
 			{
 				Name:   "thanos",
 				Usage:  "Migrate time series from Thanos blocks (supports raw and downsampled data)",
@@ -354,15 +303,17 @@ func main() {
 				Before: beforeFn,
 				Action: func(c *cli.Context) error {
 					fmt.Println("Thanos import mode")
+
 					vmCfg, err := initConfigVM(c)
 					if err != nil {
-						return fmt.Errorf("failed to init VM configuration: %w", err)
+						return fmt.Errorf("failed to init VM configuration: %s", err)
 					}
 
 					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create VM importer: %w", err)
+						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
+
 					thanosCfg := thanos.Config{
 						Snapshot: c.String(thanosSnapshot),
 						Filter: thanos.Filter{
@@ -374,7 +325,7 @@ func main() {
 					}
 					cl, err := thanos.NewClient(thanosCfg)
 					if err != nil {
-						return fmt.Errorf("failed to create thanos client: %w", err)
+						return fmt.Errorf("failed to create thanos client: %s", err)
 					}
 
 					var aggrTypes []thanos.AggrType
@@ -382,7 +333,7 @@ func main() {
 						for _, typeStr := range aggrTypesStr {
 							aggrType, err := thanos.ParseAggrType(typeStr)
 							if err != nil {
-								return fmt.Errorf("failed to parse aggregate type %q: %w", typeStr, err)
+								return fmt.Errorf("failed to parse aggregate type %q: %s", typeStr, err)
 							}
 							aggrTypes = append(aggrTypes, aggrType)
 						}
@@ -415,7 +366,7 @@ func main() {
 					bfMinDuration := c.Duration(vmNativeBackoffMinDuration)
 					bf, err := backoff.New(bfRetries, bfFactor, bfMinDuration)
 					if err != nil {
-						return fmt.Errorf("failed to create backoff object: %w", err)
+						return fmt.Errorf("failed to create backoff object: %s", err)
 					}
 
 					disableKeepAlive := c.Bool(vmNativeDisableHTTPKeepAlive)
@@ -439,7 +390,7 @@ func main() {
 
 					srcTC, err := promauth.NewTLSConfig(srcCertFile, srcKeyFile, srcCAFile, srcServerName, srcInsecureSkipVerify)
 					if err != nil {
-						return fmt.Errorf("failed to create TLS Config: %w", err)
+						return fmt.Errorf("failed to create TLS Config: %s", err)
 					}
 
 					trSrc := httputil.NewTransport(false, "vmctl_src")
@@ -469,7 +420,7 @@ func main() {
 
 					dstTC, err := promauth.NewTLSConfig(dstCertFile, dstKeyFile, dstCAFile, dstServerName, dstInsecureSkipVerify)
 					if err != nil {
-						return fmt.Errorf("failed to create TLS Config: %w", err)
+						return fmt.Errorf("failed to create TLS Config: %s", err)
 					}
 
 					trDst := httputil.NewTransport(false, "vmctl_dst")
@@ -534,7 +485,7 @@ func main() {
 					log.Printf("verifying block at path=%q", blockPath)
 					f, err := os.OpenFile(blockPath, os.O_RDONLY, 0600)
 					if err != nil {
-						return cli.Exit(fmt.Errorf("cannot open exported block at path=%q: %w", blockPath, err), 1)
+						return cli.Exit(fmt.Errorf("cannot open exported block at path=%q err=%w", blockPath, err), 1)
 					}
 					defer f.Close()
 					var blocksCount atomic.Uint64
@@ -542,7 +493,7 @@ func main() {
 						blocksCount.Add(1)
 						return nil
 					}); err != nil {
-						return cli.Exit(fmt.Errorf("cannot parse block at path=%q, blocksCount=%d: %w", blockPath, blocksCount.Load(), err), 1)
+						return cli.Exit(fmt.Errorf("cannot parse block at path=%q, blocksCount=%d, err=%w", blockPath, blocksCount.Load(), err), 1)
 					}
 					log.Printf("successfully verified block at path=%q, blockCount=%d", blockPath, blocksCount.Load())
 					return nil
@@ -585,7 +536,7 @@ func initConfigVM(c *cli.Context) (vm.Config, error) {
 
 	tr, err := promauth.NewTLSTransport(certFile, keyFile, caFile, serverName, insecureSkipVerify, "vmctl_client")
 	if err != nil {
-		return vm.Config{}, fmt.Errorf("failed to create transport for -%s=%q: %w", vmAddr, addr, err)
+		return vm.Config{}, fmt.Errorf("failed to create transport for -%s=%q: %s", vmAddr, addr, err)
 	}
 
 	bfRetries := c.Int(vmBackoffRetries)
@@ -593,7 +544,7 @@ func initConfigVM(c *cli.Context) (vm.Config, error) {
 	bfMinDuration := c.Duration(vmBackoffMinDuration)
 	bf, err := backoff.New(bfRetries, bfFactor, bfMinDuration)
 	if err != nil {
-		return vm.Config{}, fmt.Errorf("failed to create backoff object: %w", err)
+		return vm.Config{}, fmt.Errorf("failed to create backoff object: %s", err)
 	}
 
 	return vm.Config{

app/vmctl/mimir/lazyreader.go

@@ -1,195 +0,0 @@
-package mimir
-
-import (
-	"fmt"
-	"log"
-	"os"
-	"path/filepath"
-	"sync"
-
-	"github.com/oklog/ulid/v2"
-	"github.com/prometheus/prometheus/tsdb"
-	"github.com/prometheus/prometheus/tsdb/tombstones"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/common"
-)
-
-var _ tsdb.BlockReader = (*lazyBlockReader)(nil)
-
-// lazyBlockReader is stores block id and segment num information.
-// It is used to lazily fetch and parse block data.
-// It implements tsdb.BlockReader interface.
-type lazyBlockReader struct {
-	// Block ID.
-	ID ulid.ULID
-	// SegmentsNum stores the number of chunks segments in the block.
-	SegmentsNum int
-
-	mu          sync.Mutex
-	reader      *tsdb.Block
-	tempDirPath string
-	fs          common.RemoteFS
-	err         error
-}
-
-// newLazyBlockReader returns a new LazyBlockReader for the given block.
-func newLazyBlockReader(block *Block, fs common.RemoteFS) (*lazyBlockReader, error) {
-	if block.SegmentsFormat != "1b6d" {
-		return nil, fmt.Errorf("unsupported segments format: %s", block.SegmentsFormat)
-	}
-
-	return &lazyBlockReader{
-		ID:          block.ID,
-		SegmentsNum: block.SegmentsNum,
-		fs:          fs,
-	}, nil
-}
-
-func (lbr *lazyBlockReader) initialize() error {
-	lbr.mu.Lock()
-	defer lbr.mu.Unlock()
-	if lbr.reader != nil {
-		return nil
-	}
-	// fetching block and parse it and store it in lbr.reader
-	temp, err := lbr.mkTempDir()
-	if err != nil {
-		return fmt.Errorf("failed to create temp dir: %w", err)
-	}
-
-	lbr.tempDirPath = temp
-
-	// TODO: replace fetchFile and writeFile with buffered IO if needed
-	meta, err := lbr.fetchFile(metaFilename)
-	if err != nil {
-		return err
-	}
-	if err := lbr.writeFile(temp, metaFilename, meta); err != nil {
-		return fmt.Errorf("failed to write meta file: %w", err)
-	}
-	idx, err := lbr.fetchFile(indexFilename)
-	if err != nil {
-		return fmt.Errorf("failed to fetch index file %q: %w", indexFilename, err)
-	}
-	if err := lbr.writeFile(temp, indexFilename, idx); err != nil {
-		return err
-	}
-
-	for i := 1; i <= lbr.SegmentsNum; i++ {
-		// segments formats has format 1b06d
-		// https://github.com/grafana/mimir/blob/main/pkg/storage/tsdb/bucketindex/index.go#L32
-		chunkName := fmt.Sprintf("%06d", i)
-		blockChunkPath := filepath.Join("chunks", chunkName)
-		chunk, err := lbr.fetchFile(blockChunkPath)
-		if err != nil {
-			return fmt.Errorf("failed to fetch chunk file: %q: %w", chunkName, err)
-		}
-		if err := lbr.writeFile(temp, blockChunkPath, chunk); err != nil {
-			return fmt.Errorf("failed to write chunk file: %q: %w", chunkName, err)
-		}
-	}
-
-	// Set postingDecoder to nil because
-	// If it is nil then a default decoder is used, compatible with Prometheus v2.
-	pb, err := tsdb.OpenBlock(nil, temp, nil, nil)
-	if err != nil {
-		return fmt.Errorf("failed to open block %q: %w", lbr.ID, err)
-	}
-	lbr.reader = pb
-	return nil
-}
-
-// Index returns an IndexReader over the block's data.
-func (lbr *lazyBlockReader) Index() (tsdb.IndexReader, error) {
-	if err := lbr.initialize(); err != nil {
-		return nil, err
-	}
-	return lbr.reader.Index()
-}
-
-// Chunks returns a ChunkReader over the block's data.
-func (lbr *lazyBlockReader) Chunks() (tsdb.ChunkReader, error) {
-	if err := lbr.initialize(); err != nil {
-		return nil, err
-	}
-	return lbr.reader.Chunks()
-}
-
-// Tombstones returns a tombstones.Reader over the block's deleted data.
-func (lbr *lazyBlockReader) Tombstones() (tombstones.Reader, error) {
-	if err := lbr.initialize(); err != nil {
-		return nil, err
-	}
-	return lbr.reader.Tombstones()
-}
-
-// Meta provides meta information about the block reader.
-func (lbr *lazyBlockReader) Meta() tsdb.BlockMeta {
-	if err := lbr.initialize(); err != nil {
-		lbr.err = fmt.Errorf("cannot get BlockMeta: %w", err)
-		return tsdb.BlockMeta{}
-	}
-	return lbr.reader.Meta()
-}
-
-// Size returns the number of bytes that the block takes up on disk.
-func (lbr *lazyBlockReader) Size() int64 {
-	if err := lbr.initialize(); err != nil {
-		lbr.err = fmt.Errorf("error get Size of the block: %w, return zero size", err)
-		return 0
-	}
-	return lbr.reader.Size()
-}
-
-// Err returns the last error that occurred on the block reader.
-func (lbr *lazyBlockReader) Err() error {
-	return lbr.err
-}
-
-// Close closes block and releases all resources
-func (lbr *lazyBlockReader) Close() error {
-	lbr.mu.Lock()
-	defer lbr.mu.Unlock()
-	if lbr.reader == nil {
-		return nil
-	}
-
-	err := lbr.reader.Close()
-	if err := os.RemoveAll(lbr.tempDirPath); err != nil {
-		log.Printf("failed to remove temp dir: %s", err)
-	}
-	lbr.reader = nil
-	lbr.tempDirPath = ""
-
-	return err
-}
-
-func (lbr *lazyBlockReader) mkTempDir() (string, error) {
-	temp, err := os.MkdirTemp("", lbr.ID.String())
-	if err != nil {
-		return "", fmt.Errorf("failed to create temp dir: %w", err)
-	}
-	err = os.Mkdir(filepath.Join(temp, "chunks"), os.ModePerm)
-	if err != nil {
-		return "", fmt.Errorf("failed to create temp dir: %w", err)
-	}
-	return temp, nil
-}
-
-func (lbr *lazyBlockReader) fetchFile(filePath string) ([]byte, error) {
-	blockID := lbr.ID.String()
-	blockPath := filepath.Join(blockID, filePath)
-	has, err := lbr.fs.HasFile(blockPath)
-	if err != nil {
-		return nil, err
-	}
-	if !has {
-		return nil, fmt.Errorf("block meta %s not found", blockID)
-	}
-	return lbr.fs.ReadFile(blockPath)
-}
-
-func (lbr *lazyBlockReader) writeFile(folder string, filename string, file []byte) error {
-	fileName := filepath.Join(folder, filename)
-	return os.WriteFile(fileName, file, os.ModePerm)
-}

app/vmctl/mimir/mimir.go

@@ -1,238 +0,0 @@
-package mimir
-
-import (
-	"bytes"
-	"compress/gzip"
-	"context"
-	"encoding/json"
-	"fmt"
-	"log"
-
-	"github.com/oklog/ulid/v2"
-	"github.com/prometheus/prometheus/model/labels"
-	"github.com/prometheus/prometheus/tsdb"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/prometheus"
-	utils "github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vmctlutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/common"
-)
-
-const (
-	bucketIndex                   = "bucket-index.json"
-	bucketIndexCompressedFilename = bucketIndex + ".gz"
-	metaFilename                  = "meta.json"
-	indexFilename                 = "index"
-)
-
-// BlockDeletionMark holds the information about a block's deletion mark in the index.
-// This type was copied from the mimir repository https://github.com/grafana/mimir/blob/main/pkg/storage/tsdb/bucketindex/index.go#L234.
-type BlockDeletionMark struct {
-	// Block ID.
-	ID ulid.ULID `json:"block_id"`
-
-	// DeletionTime is a unix timestamp (seconds precision) of when the block was marked to be deleted.
-	DeletionTime int64 `json:"deletion_time"`
-}
-
-// Block holds the information about a block in the index.
-// This is a partial implementation of the https://github.com/grafana/mimir/blob/main/pkg/storage/tsdb/bucketindex/index.go#L73
-type Block struct {
-	// Block ID.
-	ID ulid.ULID `json:"block_id"`
-
-	// MinTime and MaxTime specify the time range all samples in the block are in (millis precision).
-	MinTime int64 `json:"min_time"`
-	MaxTime int64 `json:"max_time"`
-
-	// SegmentsFormat and SegmentsNum stores the format and number of chunks segments
-	// in the block.
-	SegmentsFormat string `json:"segments_format,omitempty"`
-	SegmentsNum    int    `json:"segments_num,omitempty"`
-}
-
-// Index contains all known blocks and markers of a tenant.
-// This is a partial implementation pof the https://github.com/grafana/mimir/blob/main/pkg/storage/tsdb/bucketindex/index.go#L36
-type Index struct {
-	// Version of the index format.
-	Version int `json:"version"`
-
-	// List of complete blocks (partial blocks are excluded from the index).
-	Blocks []*Block `json:"blocks"`
-}
-
-// Config contains a list of params needed
-// for reading mimir snapshots
-type Config struct {
-	// Path to remote storage bucket
-	Path string
-	// TenantID is the tenant id for the storage
-	TenantID string
-
-	Filter Filter
-
-	CredsFilePath           string
-	ConfigFilePath          string
-	ConfigProfile           string
-	CustomS3Endpoint        string
-	S3ForcePathStyle        bool
-	S3TLSInsecureSkipVerify bool
-
-	SSEKMSKeyID  string
-	SSEAlgorithm string
-}
-
-// Filter contains configuration for filtering
-// the timeseries
-type Filter struct {
-	TimeMin    string
-	TimeMax    string
-	Label      string
-	LabelValue string
-}
-
-// Client is a wrapper over Prometheus tsdb.DBReader
-type Client struct {
-	common.RemoteFS
-	filter filter
-}
-
-type filter struct {
-	min, max   int64
-	label      string
-	labelValue string
-}
-
-func (f filter) inRange(minTime, maxTime int64) bool {
-	fmin, fmax := f.min, f.max
-	if minTime == 0 {
-		fmin = minTime
-	}
-	if fmax == 0 {
-		fmax = maxTime
-	}
-	return minTime <= fmax && fmin <= maxTime
-}
-
-// NewClient creates and validates new Client
-// with given Config
-func NewClient(ctx context.Context, cfg Config) (*Client, error) {
-	if cfg.Path == "" {
-		return nil, fmt.Errorf("path cannot be empty")
-	}
-
-	if cfg.TenantID != "" {
-		cfg.Path = fmt.Sprintf("%s/%s", cfg.Path, cfg.TenantID)
-	}
-
-	var c Client
-	rfs, err := newRemoteFS(ctx, cfg)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse `-src`=%q: %w", cfg.Path, err)
-	}
-
-	c.RemoteFS = rfs
-	timeMin, err := utils.ParseTime(cfg.Filter.TimeMin)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse min time in filter: %w", err)
-	}
-	timeMax, err := utils.ParseTime(cfg.Filter.TimeMax)
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse max time in filter: %w", err)
-	}
-	c.filter = filter{
-		min:        timeMin.UnixMilli(),
-		max:        timeMax.UnixMilli(),
-		label:      cfg.Filter.Label,
-		labelValue: cfg.Filter.LabelValue,
-	}
-	return &c, nil
-}
-
-// Explore a fetches bucket-index.json file from a remote storage or local filesystem
-// and filter blocks via the defined time range, but does not take into account label filters.
-func (c *Client) Explore() ([]tsdb.BlockReader, error) {
-
-	log.Printf("Fetching blocks from remote storage")
-
-	indexFile, err := c.fetchIndexFile()
-	if err != nil {
-		return nil, fmt.Errorf("failed to fetch index file: %w", err)
-	}
-
-	var blocksToImport []tsdb.BlockReader
-	for _, block := range indexFile.Blocks {
-		if !c.filter.inRange(block.MinTime, block.MaxTime) {
-			// Skipping block outside of time range
-			continue
-		}
-
-		if block.ID.String() == "" {
-			continue
-		}
-
-		lazyBlockReader, err := newLazyBlockReader(block, c.RemoteFS)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create lazy block reader: %w", err)
-		}
-		blocksToImport = append(blocksToImport, lazyBlockReader)
-	}
-
-	return blocksToImport, nil
-}
-
-// Read reads the given BlockReader according to configured
-// time and label filters.
-func (c *Client) Read(ctx context.Context, block tsdb.BlockReader) (*prometheus.CloseableSeriesSet, error) {
-	meta := block.Meta()
-	if b, ok := block.(*lazyBlockReader); ok && b.Err() != nil {
-		return nil, fmt.Errorf("failed to read block: %w", b.Err())
-	}
-
-	if meta.ULID.String() == "" {
-		return nil, fmt.Errorf("unexpected block without id")
-	}
-
-	minTime, maxTime := meta.MinTime, meta.MaxTime
-	if c.filter.min != 0 {
-		minTime = c.filter.min
-	}
-	if c.filter.max != 0 {
-		maxTime = c.filter.max
-	}
-	q, err := tsdb.NewBlockQuerier(block, minTime, maxTime)
-	if err != nil {
-		return nil, err
-	}
-	ss := q.Select(ctx, false, nil, labels.MustNewMatcher(labels.MatchRegexp, c.filter.label, c.filter.labelValue))
-	return &prometheus.CloseableSeriesSet{SeriesSet: ss, Close: q.Close}, nil
-}
-
-func (c *Client) fetchIndexFile() (*Index, error) {
-	has, err := c.HasFile(bucketIndexCompressedFilename)
-	if err != nil {
-		return nil, err
-	}
-	if !has {
-		return nil, fmt.Errorf("bucket-index.json.gz not found")
-	}
-
-	file, err := c.ReadFile(bucketIndexCompressedFilename)
-	if err != nil {
-		return nil, fmt.Errorf("failed to read bucket index: %w", err)
-	}
-
-	r := bytes.NewReader(file)
-	// Read all the content.
-	gzipReader, err := gzip.NewReader(r)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create gzip reader: %w", err)
-	}
-
-	var indexFile Index
-	err = json.NewDecoder(gzipReader).Decode(&indexFile)
-	if err != nil {
-		return nil, fmt.Errorf("failed to decode bucket index: %w", err)
-	}
-
-	return &indexFile, nil
-}

app/vmctl/mimir/remotefs.go

@@ -1,93 +0,0 @@
-package mimir
-
-import (
-	"context"
-	"fmt"
-	"path/filepath"
-	"strings"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/azremote"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/common"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/fsremote"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/gcsremote"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/backup/s3remote"
-)
-
-// newRemoteFS returns new remote fs from the given Config.
-func newRemoteFS(ctx context.Context, cfg Config) (common.RemoteFS, error) {
-	if len(cfg.Path) == 0 {
-		return nil, fmt.Errorf("path cannot be empty")
-	}
-	n := strings.Index(cfg.Path, "://")
-	if n < 0 {
-		return nil, fmt.Errorf("missing scheme in path %q. Supported schemes: `gs://`, `s3://`, `azblob://`, `fs://`", cfg.Path)
-	}
-	scheme := cfg.Path[:n]
-	dir := cfg.Path[n+len("://"):]
-	switch scheme {
-	case "fs":
-		if !filepath.IsAbs(dir) {
-			return nil, fmt.Errorf("dir must be absolute; got %q", dir)
-		}
-		fsr := &fsremote.FS{
-			Dir: filepath.Clean(dir),
-		}
-		return fsr, nil
-	case "gcs", "gs":
-		n := strings.Index(dir, "/")
-		if n < 0 {
-			return nil, fmt.Errorf("missing directory on the gcs bucket %q", dir)
-		}
-		bucket := dir[:n]
-		dir = dir[n:]
-		fsr := &gcsremote.FS{
-			CredsFilePath: cfg.CredsFilePath,
-			Bucket:        bucket,
-			Dir:           dir,
-		}
-		if err := fsr.Init(ctx); err != nil {
-			return nil, fmt.Errorf("cannot initialize connection to gcs: %w", err)
-		}
-		return fsr, nil
-	case "azblob":
-		n := strings.Index(dir, "/")
-		if n < 0 {
-			return nil, fmt.Errorf("missing directory on the AZBlob container %q", dir)
-		}
-		bucket := dir[:n]
-		dir = dir[n:]
-		fsr := &azremote.FS{
-			Container: bucket,
-			Dir:       dir,
-		}
-		if err := fsr.Init(ctx); err != nil {
-			return nil, fmt.Errorf("cannot initialize connection to AZBlob: %w", err)
-		}
-		return fsr, nil
-	case "s3":
-		n := strings.Index(dir, "/")
-		if n < 0 {
-			return nil, fmt.Errorf("missing directory on the s3 bucket %q", dir)
-		}
-		bucket := dir[:n]
-		dir = dir[n:]
-		fsr := &s3remote.FS{
-			CredsFilePath:         cfg.CredsFilePath,
-			ConfigFilePath:        cfg.ConfigFilePath,
-			CustomEndpoint:        cfg.CustomS3Endpoint,
-			TLSInsecureSkipVerify: cfg.S3TLSInsecureSkipVerify,
-			S3ForcePathStyle:      cfg.S3ForcePathStyle,
-			ProfileName:           cfg.ConfigProfile,
-			Bucket:                bucket,
-			Dir:                   dir,
-			SSEKMSKeyId:           cfg.SSEKMSKeyID,
-			SSEAlgorithm:          s3remote.StringToEncryptionAlgorithm(cfg.SSEAlgorithm),
-		}
-		if err := fsr.Init(ctx); err != nil {
-			return nil, fmt.Errorf("cannot initialize connection to s3: %w", err)
-		}
-		return fsr, nil
-	default:
-		return nil, fmt.Errorf("unsupported scheme %q", scheme)
-	}
-}

app/vmctl/native/client.go

@@ -47,7 +47,7 @@ func (c *Client) Explore(ctx context.Context, f Filter, tenantID string, start,
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		exploreRequestsErrorsTotal.Inc()
-		return nil, fmt.Errorf("cannot create request to %q: %w", url, err)
+		return nil, fmt.Errorf("cannot create request to %q: %s", url, err)
 	}
 
 	params := req.URL.Query()
@@ -60,14 +60,14 @@ func (c *Client) Explore(ctx context.Context, f Filter, tenantID string, start,
 	if err != nil {
 		exploreRequestsErrorsTotal.Inc()
 		exploreDuration.UpdateDuration(startTime)
-		return nil, fmt.Errorf("series request failed: %w", err)
+		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: %w", err)
+		return nil, fmt.Errorf("cannot decode series response: %s", err)
 	}
 	exploreDuration.UpdateDuration(startTime)
 	return response.MetricNames, resp.Body.Close()
@@ -80,19 +80,19 @@ func (c *Client) ImportPipe(ctx context.Context, dstURL string, pr *io.PipeReade
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, dstURL, pr)
 	if err != nil {
 		importRequestsErrorsTotal.Inc()
-		return fmt.Errorf("cannot create import request to %q: %w", c.Addr, err)
+		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: %w", err)
+		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: %w", err)
+		return fmt.Errorf("cannot close import response body: %s", err)
 	}
 	importDuration.UpdateDuration(startTime)
 	return nil
@@ -105,7 +105,7 @@ func (c *Client) ExportPipe(ctx context.Context, url string, f Filter) (io.ReadC
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		exportRequestsErrorsTotal.Inc()
-		return nil, fmt.Errorf("cannot create request to %q: %w", c.Addr, err)
+		return nil, fmt.Errorf("cannot create request to %q: %s", c.Addr, err)
 	}
 
 	params := req.URL.Query()
@@ -136,7 +136,7 @@ func (c *Client) GetSourceTenants(ctx context.Context, f Filter) ([]string, erro
 	u := fmt.Sprintf("%s/%s", c.Addr, nativeTenantsAddr)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
 	if err != nil {
-		return nil, fmt.Errorf("cannot create request to %q: %w", u, err)
+		return nil, fmt.Errorf("cannot create request to %q: %s", u, err)
 	}
 
 	params := req.URL.Query()
@@ -150,18 +150,18 @@ func (c *Client) GetSourceTenants(ctx context.Context, f Filter) ([]string, erro
 
 	resp, err := c.do(req, http.StatusOK)
 	if err != nil {
-		return nil, fmt.Errorf("tenants request failed: %w", err)
+		return nil, fmt.Errorf("tenants request failed: %s", err)
 	}
 
 	var r struct {
 		Tenants []string `json:"data"`
 	}
 	if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
-		return nil, fmt.Errorf("cannot decode tenants response: %w", err)
+		return nil, fmt.Errorf("cannot decode tenants response: %s", err)
 	}
 
 	if err := resp.Body.Close(); err != nil {
-		return nil, fmt.Errorf("cannot close tenants response body: %w", err)
+		return nil, fmt.Errorf("cannot close tenants response body: %s", err)
 	}
 
 	return r.Tenants, nil
@@ -180,7 +180,7 @@ func (c *Client) do(req *http.Request, expSC int) (*http.Response, error) {
 	if resp.StatusCode != expSC {
 		body, err := io.ReadAll(resp.Body)
 		if err != nil {
-			return nil, fmt.Errorf("failed to read response body for status code %d: %w", resp.StatusCode, err)
+			return nil, fmt.Errorf("failed to read response body for status code %d: %s", resp.StatusCode, err)
 		}
 		return nil, fmt.Errorf("unexpected response code %d: %s", resp.StatusCode, string(body))
 	}

app/vmctl/opentsdb.go

@@ -47,7 +47,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 		q := fmt.Sprintf("%s/api/suggest?type=metrics&q=%s&max=%d", op.oc.Addr, filter, op.oc.Limit)
 		m, err := op.oc.FindMetrics(q)
 		if err != nil {
-			return fmt.Errorf("metric discovery failed for %q: %w", q, err)
+			return fmt.Errorf("metric discovery failed for %q: %s", q, err)
 		}
 		metrics = append(metrics, m...)
 	}
@@ -76,7 +76,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 		log.Printf("Starting work on %s", metric)
 		serieslist, err := op.oc.FindSeries(metric)
 		if err != nil {
-			return fmt.Errorf("couldn't retrieve series list for %s: %w", metric, err)
+			return fmt.Errorf("couldn't retrieve series list for %s : %s", metric, err)
 		}
 		/*
 			Create channels for collecting/processing series and errors
@@ -95,7 +95,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 				for s := range seriesCh {
 					if err := op.do(s); err != nil {
 						otsdbErrorsTotal.Inc()
-						errCh <- fmt.Errorf("couldn't retrieve series for %s: %w", metric, err)
+						errCh <- fmt.Errorf("couldn't retrieve series for %s : %s", metric, err)
 						return
 					}
 					otsdbSeriesProcessed.Inc()
@@ -112,7 +112,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 		// check for any lingering errors on the query side
 		for otsdbErr := range errCh {
 			if runErr == nil {
-				runErr = fmt.Errorf("import process failed:\n%w", otsdbErr)
+				runErr = fmt.Errorf("import process failed: \n%s", otsdbErr)
 			}
 		}
 		bar.Finish()
@@ -125,7 +125,7 @@ func (op *otsdbProcessor) run(ctx context.Context) error {
 	for vmErr := range op.im.Errors() {
 		if vmErr.Err != nil {
 			otsdbErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, op.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, op.isVerbose))
 		}
 	}
 	log.Println("Import finished!")
@@ -141,12 +141,12 @@ func (op *otsdbProcessor) sendQueries(ctx context.Context, serieslist []opentsdb
 			for _, tr := range rt.QueryRanges {
 				select {
 				case <-ctx.Done():
-					return fmt.Errorf("context canceled: %w", ctx.Err())
+					return fmt.Errorf("context canceled: %s", ctx.Err())
 				case otsdbErr := <-errCh:
 					otsdbErrorsTotal.Inc()
-					return fmt.Errorf("opentsdb error: %w", otsdbErr)
+					return fmt.Errorf("opentsdb error: %s", otsdbErr)
 				case vmErr := <-op.im.Errors():
-					return fmt.Errorf("import process failed: %w", wrapErr(vmErr, op.isVerbose))
+					return fmt.Errorf("import process failed: %s", wrapErr(vmErr, op.isVerbose))
 				case seriesCh <- queryObj{
 					Tr: tr, StartTime: startTime,
 					Series: series, Rt: opentsdb.RetentionMeta{
@@ -166,7 +166,7 @@ func (op *otsdbProcessor) do(s queryObj) error {
 	end := s.StartTime - s.Tr.End
 	data, err := op.oc.GetData(s.Series, s.Rt, start, end, op.oc.MsecsTime)
 	if err != nil {
-		return fmt.Errorf("failed to collect data for %v in %v:%v :: %w", s.Series, s.Rt, s.Tr, err)
+		return fmt.Errorf("failed to collect data for %v in %v:%v :: %v", s.Series, s.Rt, s.Tr, err)
 	}
 	if len(data.Timestamps) < 1 || len(data.Values) < 1 {
 		log.Printf("no data found for %v in %v:%v...skipping", s.Series, s.Rt, s.Tr)

app/vmctl/opentsdb/opentsdb.go

@@ -106,7 +106,7 @@ func (c Client) FindMetrics(q string) ([]string, error) {
 
 	resp, err := c.c.Get(q)
 	if err != nil {
-		return nil, fmt.Errorf("failed to send GET request to %q: %w", q, err)
+		return nil, fmt.Errorf("failed to send GET request to %q: %s", q, err)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode != 200 {
@@ -114,12 +114,12 @@ func (c Client) FindMetrics(q string) ([]string, error) {
 	}
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return nil, fmt.Errorf("could not retrieve metric data from %q: %w", q, err)
+		return nil, fmt.Errorf("could not retrieve metric data from %q: %s", q, err)
 	}
 	var metriclist []string
 	err = json.Unmarshal(body, &metriclist)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read response from %q: %w", q, err)
+		return nil, fmt.Errorf("failed to read response from %q: %s", q, err)
 	}
 	return metriclist, nil
 }
@@ -130,7 +130,7 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
 	q := fmt.Sprintf("%s/api/search/lookup?m=%s&limit=%d", c.Addr, metric, c.Limit)
 	resp, err := c.c.Get(q)
 	if err != nil {
-		return nil, fmt.Errorf("failed to send GET request to %q: %w", q, err)
+		return nil, fmt.Errorf("failed to send GET request to %q: %s", q, err)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode != 200 {
@@ -138,12 +138,12 @@ func (c Client) FindSeries(metric string) ([]Meta, error) {
 	}
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return nil, fmt.Errorf("could not retrieve series data from %q: %w", q, err)
+		return nil, fmt.Errorf("could not retrieve series data from %q: %s", q, err)
 	}
 	var results MetaResults
 	err = json.Unmarshal(body, &results)
 	if err != nil {
-		return nil, fmt.Errorf("failed to read response from %q: %w", q, err)
+		return nil, fmt.Errorf("failed to read response from %q: %s", q, err)
 	}
 	return results.Results, nil
 }
@@ -183,7 +183,7 @@ func (c Client) GetData(series Meta, rt RetentionMeta, start int64, end int64, m
 	q := fmt.Sprintf("%s/api/query?%s", c.Addr, queryStr)
 	resp, err := c.c.Get(q)
 	if err != nil {
-		return Metric{}, fmt.Errorf("failed to send GET request to %q: %w", q, err)
+		return Metric{}, fmt.Errorf("failed to send GET request to %q: %s", q, err)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	/*
@@ -303,7 +303,7 @@ func NewClient(cfg Config) (*Client, error) {
 	for _, r := range cfg.Retentions {
 		ret, err := convertRetention(r, offsetSecs, cfg.MsecsTime)
 		if err != nil {
-			return &Client{}, fmt.Errorf("couldn't parse retention %q :: %w", r, err)
+			return &Client{}, fmt.Errorf("couldn't parse retention %q :: %v", r, err)
 		}
 		retentions = append(retentions, ret)
 	}

app/vmctl/opentsdb/parser.go

@@ -88,7 +88,7 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
 	}
 	queryLengthDuration, err := convertDuration(chunks[2])
 	if err != nil {
-		return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %w", chunks[2], err)
+		return Retention{}, fmt.Errorf("invalid ttl (second order) duration string: %q: %s", chunks[2], err)
 	}
 	// set ttl in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
 	queryLength := queryLengthDuration.Milliseconds()
@@ -110,7 +110,7 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
 
 	aggTimeDuration, err := convertDuration(aggregates[1])
 	if err != nil {
-		return Retention{}, fmt.Errorf("invalid aggregation time duration string: %q: %w", aggregates[1], err)
+		return Retention{}, fmt.Errorf("invalid aggregation time duration string: %q: %s", aggregates[1], err)
 	}
 	aggTime := aggTimeDuration.Milliseconds()
 	if !msecTime {
@@ -119,7 +119,7 @@ func convertRetention(retention string, offset int64, msecTime bool) (Retention,
 
 	rowLengthDuration, err := convertDuration(chunks[1])
 	if err != nil {
-		return Retention{}, fmt.Errorf("invalid row length (first order) duration string: %q: %w", chunks[1], err)
+		return Retention{}, fmt.Errorf("invalid row length (first order) duration string: %q: %s", chunks[1], err)
 	}
 	// set length of each row in milliseconds, unless we aren't using millisecond time in OpenTSDB...then use seconds
 	rowLength := rowLengthDuration.Milliseconds()

app/vmctl/prometheus.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"io"
 	"log"
 	"strings"
 	"sync"
@@ -19,17 +18,10 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
 )
 
-// Runner is an interface for fetching and reading
-// snapshot blocks
-type Runner interface {
-	Explore() ([]tsdb.BlockReader, error)
-	Read(context.Context, tsdb.BlockReader) (*prometheus.CloseableSeriesSet, error)
-}
-
 type prometheusProcessor struct {
-	// Runner fetches and reads
+	// prometheus client fetches and reads
 	// snapshot blocks
-	cl Runner
+	cl *prometheus.Client
 	// importer performs import requests
 	// for timeseries data returned from
 	// snapshot blocks
@@ -46,7 +38,7 @@ type prometheusProcessor struct {
 func (pp *prometheusProcessor) run(ctx context.Context) error {
 	blocks, err := pp.cl.Explore()
 	if err != nil {
-		return fmt.Errorf("explore failed: %w", err)
+		return fmt.Errorf("explore failed: %s", err)
 	}
 	if len(blocks) < 1 {
 		return fmt.Errorf("found no blocks to import")
@@ -56,8 +48,8 @@ func (pp *prometheusProcessor) run(ctx context.Context) error {
 		return nil
 	}
 
-	if err := pp.processBlocks(ctx, blocks); err != nil {
-		return fmt.Errorf("migration failed: %w", err)
+	if err := pp.processBlocks(blocks); err != nil {
+		return fmt.Errorf("migration failed: %s", err)
 	}
 
 	log.Println("Import finished!")
@@ -65,17 +57,11 @@ func (pp *prometheusProcessor) run(ctx context.Context) error {
 	return nil
 }
 
-func (pp *prometheusProcessor) do(ctx context.Context, b tsdb.BlockReader) error {
-	css, err := pp.cl.Read(ctx, b)
+func (pp *prometheusProcessor) do(b tsdb.BlockReader) error {
+	ss, err := pp.cl.Read(b)
 	if err != nil {
-		return fmt.Errorf("failed to read block: %w", err)
+		return fmt.Errorf("failed to read block: %s", err)
 	}
-	defer func() {
-		if err := css.Close(); err != nil {
-			log.Printf("cannot close SeriesSet for block: %q : %s\n", b.Meta().ULID, err)
-		}
-	}()
-	ss := css.SeriesSet
 	var it chunkenc.Iterator
 	for ss.Next() {
 		var name string
@@ -128,7 +114,7 @@ func (pp *prometheusProcessor) do(ctx context.Context, b tsdb.BlockReader) error
 	return ss.Err()
 }
 
-func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []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 {
@@ -144,16 +130,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
 	for range pp.cc {
 		wg.Go(func() {
 			for br := range blockReadersCh {
-				if err := pp.do(ctx, br); err != nil {
+				if err := pp.do(br); err != nil {
 					promErrorsTotal.Inc()
-					errCh <- fmt.Errorf("cannot read block %q: %w", br.Meta().ULID, err)
+					errCh <- fmt.Errorf("read failed for block %q: %s", br.Meta().ULID, err)
 					return
 				}
-				if cb, ok := br.(io.Closer); ok {
-					if err := cb.Close(); err != nil {
-						errCh <- fmt.Errorf("cannot close block: %q: %w", br.Meta().ULID, err)
-					}
-				}
 				promBlocksProcessed.Inc()
 				bar.Increment()
 			}
@@ -164,11 +145,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
 		select {
 		case promErr := <-errCh:
 			close(blockReadersCh)
-			return fmt.Errorf("prometheus error: %w", promErr)
+			return fmt.Errorf("prometheus error: %s", promErr)
 		case vmErr := <-pp.im.Errors():
 			close(blockReadersCh)
 			promErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, pp.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
 		case blockReadersCh <- br:
 		}
 	}
@@ -182,11 +163,11 @@ func (pp *prometheusProcessor) processBlocks(ctx context.Context, blocks []tsdb.
 	for vmErr := range pp.im.Errors() {
 		if vmErr.Err != nil {
 			promErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, pp.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, pp.isVerbose))
 		}
 	}
 	for err := range errCh {
-		return fmt.Errorf("import process failed: %w", err)
+		return fmt.Errorf("import process failed: %s", err)
 	}
 
 	return nil

app/vmctl/prometheus/prometheus.go

@@ -8,8 +8,6 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/storage"
 	"github.com/prometheus/prometheus/tsdb"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vmctlutil"
 )
 
 // Config contains a list of params needed
@@ -59,16 +57,16 @@ func (f filter) inRange(minV, maxV int64) bool {
 func NewClient(cfg Config) (*Client, error) {
 	db, err := tsdb.OpenDBReadOnly(cfg.Snapshot, cfg.TemporaryDir, nil)
 	if err != nil {
-		return nil, fmt.Errorf("failed to open snapshot %q: %w", cfg.Snapshot, err)
+		return nil, fmt.Errorf("failed to open snapshot %q: %s", cfg.Snapshot, err)
 	}
 	c := &Client{DBReadOnly: db}
-	timeMin, timeMax, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
+	minTime, maxTime, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse time in filter: %w", err)
+		return nil, fmt.Errorf("failed to parse time in filter: %s", err)
 	}
 	c.filter = filter{
-		min:        timeMin,
-		max:        timeMax,
+		min:        minTime,
+		max:        maxTime,
 		label:      cfg.Filter.Label,
 		labelValue: cfg.Filter.LabelValue,
 	}
@@ -83,9 +81,9 @@ func NewClient(cfg Config) (*Client, error) {
 func (c *Client) Explore() ([]tsdb.BlockReader, error) {
 	blocks, err := c.Blocks()
 	if err != nil {
-		return nil, fmt.Errorf("failed to fetch blocks: %w", err)
+		return nil, fmt.Errorf("failed to fetch blocks: %s", err)
 	}
-	s := &vmctlutil.Stats{
+	s := &Stats{
 		Filtered: c.filter.min != 0 || c.filter.max != 0 || c.filter.label != "",
 		Blocks:   len(blocks),
 	}
@@ -110,15 +108,9 @@ func (c *Client) Explore() ([]tsdb.BlockReader, error) {
 	return blocksToImport, nil
 }
 
-// CloseableSeriesSet defines a SeriesSet with Close method
-type CloseableSeriesSet struct {
-	SeriesSet storage.SeriesSet
-	Close     func() error
-}
-
 // Read reads the given BlockReader according to configured
 // time and label filters.
-func (c *Client) Read(ctx context.Context, block tsdb.BlockReader) (*CloseableSeriesSet, error) {
+func (c *Client) Read(block tsdb.BlockReader) (storage.SeriesSet, error) {
 	minTime, maxTime := block.Meta().MinTime, block.Meta().MaxTime
 	if c.filter.min != 0 {
 		minTime = c.filter.min
@@ -130,8 +122,8 @@ func (c *Client) Read(ctx context.Context, block tsdb.BlockReader) (*CloseableSe
 	if err != nil {
 		return nil, err
 	}
-	ss := q.Select(ctx, false, nil, labels.MustNewMatcher(labels.MatchRegexp, c.filter.label, c.filter.labelValue))
-	return &CloseableSeriesSet{ss, q.Close}, nil
+	ss := q.Select(context.Background(), false, nil, labels.MustNewMatcher(labels.MatchRegexp, c.filter.label, c.filter.labelValue))
+	return ss, nil
 }
 
 func parseTime(start, end string) (int64, int64, error) {
@@ -142,14 +134,14 @@ func parseTime(start, end string) (int64, int64, error) {
 	if start != "" {
 		v, err := time.Parse(time.RFC3339, start)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to parse %q: %w", start, err)
+			return 0, 0, fmt.Errorf("failed to parse %q: %s", start, err)
 		}
 		s = v.UnixNano() / int64(time.Millisecond)
 	}
 	if end != "" {
 		v, err := time.Parse(time.RFC3339, end)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to parse %q: %w", end, err)
+			return 0, 0, fmt.Errorf("failed to parse %q: %s", end, err)
 		}
 		e = v.UnixNano() / int64(time.Millisecond)
 	}

app/vmctl/prometheus/stats.go

@@ -1,4 +1,4 @@
-package vmctlutil
+package prometheus
 
 import (
 	"fmt"
@@ -18,7 +18,7 @@ type Stats struct {
 
 // String returns string representation for s.
 func (s Stats) String() string {
-	str := fmt.Sprintf("Snapshot stats:\n"+
+	str := fmt.Sprintf("Prometheus snapshot stats:\n"+
 		"  blocks found: %d;\n"+
 		"  blocks skipped by time filter: %d;\n"+
 		"  min time: %d (%v);\n"+

app/vmctl/remoteread.go

@@ -44,7 +44,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 
 	ranges, err := stepper.SplitDateRange(*rrp.filter.timeStart, *rrp.filter.timeEnd, rrp.filter.chunk, rrp.filter.timeReverse)
 	if err != nil {
-		return fmt.Errorf("failed to create date ranges for the given time filters: %w", err)
+		return fmt.Errorf("failed to create date ranges for the given time filters: %v", err)
 	}
 
 	question := fmt.Sprintf("Selected time range %q - %q will be split into %d ranges according to %q step. Continue?",
@@ -74,7 +74,7 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 			for r := range rangeC {
 				if err := rrp.do(ctx, r); err != nil {
 					remoteReadErrorsTotal.Inc()
-					errCh <- fmt.Errorf("request failed for: %w", err)
+					errCh <- fmt.Errorf("request failed for: %s", err)
 					return
 				}
 				remoteReadRangesProcessed.Inc()
@@ -86,10 +86,10 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 	for _, r := range ranges {
 		select {
 		case infErr := <-errCh:
-			return fmt.Errorf("remote read error: %w", infErr)
+			return fmt.Errorf("remote read error: %s", infErr)
 		case vmErr := <-rrp.dst.Errors():
 			remoteReadErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, rrp.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
 		case rangeC <- &remoteread.Filter{
 			StartTimestampMs: r[0].UnixMilli(),
 			EndTimestampMs:   r[1].UnixMilli(),
@@ -105,11 +105,11 @@ func (rrp *remoteReadProcessor) run(ctx context.Context) error {
 	for vmErr := range rrp.dst.Errors() {
 		if vmErr.Err != nil {
 			remoteReadErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, rrp.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
 		}
 	}
 	for err := range errCh {
-		return fmt.Errorf("import process failed: %w", err)
+		return fmt.Errorf("import process failed: %s", err)
 	}
 
 	return nil
@@ -119,7 +119,7 @@ func (rrp *remoteReadProcessor) do(ctx context.Context, filter *remoteread.Filte
 	return rrp.src.Read(ctx, filter, func(series *vm.TimeSeries) error {
 		if err := rrp.dst.Input(series); err != nil {
 			return fmt.Errorf(
-				"failed to read data for time range start: %d, end: %d: %w",
+				"failed to read data for time range start: %d, end: %d, %s",
 				filter.StartTimestampMs, filter.EndTimestampMs, err)
 		}
 		return nil

app/vmctl/remoteread/remoteread.go

@@ -157,7 +157,7 @@ func (c *Client) Read(ctx context.Context, filter *Filter, streamCb StreamCallba
 		if errors.Is(err, context.Canceled) {
 			return fmt.Errorf("fetch request has ben cancelled")
 		}
-		return fmt.Errorf("error while fetching data from remote storage: %w", err)
+		return fmt.Errorf("error while fetching data from remote storage: %s", err)
 	}
 	return nil
 }

app/vmctl/thanos/client.go

@@ -52,7 +52,7 @@ func (f filter) inRange(minV, maxV int64) bool {
 func NewClient(cfg Config) (*Client, error) {
 	minTime, maxTime, err := parseTime(cfg.Filter.TimeMin, cfg.Filter.TimeMax)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse time in filter: %w", err)
+		return nil, fmt.Errorf("failed to parse time in filter: %s", err)
 	}
 	return &Client{
 		snapshotPath: cfg.Snapshot,
@@ -183,14 +183,14 @@ func parseTime(start, end string) (int64, int64, error) {
 	if start != "" {
 		v, err := time.Parse(time.RFC3339, start)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to parse %q: %w", start, err)
+			return 0, 0, fmt.Errorf("failed to parse %q: %s", start, err)
 		}
 		s = v.UnixNano() / int64(time.Millisecond)
 	}
 	if end != "" {
 		v, err := time.Parse(time.RFC3339, end)
 		if err != nil {
-			return 0, 0, fmt.Errorf("failed to parse %q: %w", end, err)
+			return 0, 0, fmt.Errorf("failed to parse %q: %s", end, err)
 		}
 		e = v.UnixNano() / int64(time.Millisecond)
 	}

app/vmctl/thanos_processor.go

@@ -36,7 +36,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
 	// Use the first aggregate type to explore blocks (block list is the same for all types)
 	blocks, err := tp.cl.Explore(tp.aggrTypes[0])
 	if err != nil {
-		return fmt.Errorf("explore failed: %w", err)
+		return fmt.Errorf("explore failed: %s", err)
 	}
 	if len(blocks) < 1 {
 		return fmt.Errorf("found no blocks to import")
@@ -84,7 +84,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
 		log.Println("Processing raw blocks (resolution=0)...")
 		stats, err := tp.processBlocks(rawBlocks, thanos.AggrTypeNone, bar)
 		if err != nil {
-			return fmt.Errorf("migration failed for raw blocks: %w", err)
+			return fmt.Errorf("migration failed for raw blocks: %s", err)
 		}
 		phases = append(phases, phaseStats{
 			name:    "raw",
@@ -108,7 +108,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
 
 		aggrBlocks, err := tp.cl.Explore(aggrType)
 		if err != nil {
-			return fmt.Errorf("explore failed for aggr type %s: %w", aggrType, err)
+			return fmt.Errorf("explore failed for aggr type %s: %s", aggrType, err)
 		}
 
 		var downsampledOnly []thanos.BlockInfo
@@ -128,7 +128,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
 		stats, err := tp.processBlocks(downsampledOnly, aggrType, bar)
 		thanos.CloseBlocks(aggrBlocks)
 		if err != nil {
-			return fmt.Errorf("migration failed for aggr type %s: %w", aggrType, err)
+			return fmt.Errorf("migration failed for aggr type %s: %s", aggrType, err)
 		}
 		phases = append(phases, phaseStats{
 			name:    aggrType.String(),
@@ -153,7 +153,7 @@ func (tp *thanosProcessor) run(ctx context.Context) error {
 	for vmErr := range tp.im.Errors() {
 		if vmErr.Err != nil {
 			thanosErrorsTotal.Inc()
-			return fmt.Errorf("import process failed: %w", wrapErr(vmErr, tp.isVerbose))
+			return fmt.Errorf("import process failed: %s", wrapErr(vmErr, tp.isVerbose))
 		}
 	}
 
@@ -184,7 +184,7 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
 				seriesCount, samplesCount, err := tp.do(bi, aggrType)
 				if err != nil {
 					thanosErrorsTotal.Inc()
-					errCh <- fmt.Errorf("read failed for block %q with aggr %s: %w", bi.Block.Meta().ULID, aggrType, err)
+					errCh <- fmt.Errorf("read failed for block %q with aggr %s: %s", bi.Block.Meta().ULID, aggrType, err)
 					return
 				}
 
@@ -209,12 +209,12 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
 		case thanosErr := <-errCh:
 			close(blockReadersCh)
 			wg.Wait()
-			return processBlocksStats{}, fmt.Errorf("thanos error: %w", thanosErr)
+			return processBlocksStats{}, fmt.Errorf("thanos error: %s", thanosErr)
 		case vmErr := <-tp.im.Errors():
 			close(blockReadersCh)
 			wg.Wait()
 			thanosErrorsTotal.Inc()
-			return processBlocksStats{}, fmt.Errorf("import process failed: %w", wrapErr(vmErr, tp.isVerbose))
+			return processBlocksStats{}, fmt.Errorf("import process failed: %s", wrapErr(vmErr, tp.isVerbose))
 		case blockReadersCh <- bi:
 		}
 	}
@@ -223,7 +223,7 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
 	wg.Wait()
 	close(errCh)
 	for err := range errCh {
-		return processBlocksStats{}, fmt.Errorf("import process failed: %w", err)
+		return processBlocksStats{}, fmt.Errorf("import process failed: %s", err)
 	}
 
 	return processBlocksStats{
@@ -236,7 +236,7 @@ func (tp *thanosProcessor) processBlocks(blocks []thanos.BlockInfo, aggrType tha
 func (tp *thanosProcessor) do(bi thanos.BlockInfo, aggrType thanos.AggrType) (uint64, uint64, error) {
 	ss, err := tp.cl.Read(bi)
 	if err != nil {
-		return 0, 0, fmt.Errorf("failed to read block: %w", err)
+		return 0, 0, fmt.Errorf("failed to read block: %s", err)
 	}
 	defer ss.Close() // Ensure querier is closed even on early returns
 

app/vmctl/vm/vm.go

@@ -163,7 +163,7 @@ func NewImporter(ctx context.Context, cfg Config) (*Importer, error) {
 		importDuration:            metrics.GetOrCreateHistogram(`vmctl_importer_request_duration_seconds`),
 	}
 	if err := im.Ping(); err != nil {
-		return nil, fmt.Errorf("ping to %q failed: %w", addr, err)
+		return nil, fmt.Errorf("ping to %q failed: %s", addr, err)
 	}
 
 	if cfg.BatchSize < 1 {
@@ -289,7 +289,7 @@ func (im *Importer) flush(ctx context.Context, b []*TimeSeries) error {
 	retryableFunc := func() error { return im.Import(b) }
 	attempts, err := im.backoff.Retry(ctx, retryableFunc)
 	if err != nil {
-		return fmt.Errorf("import failed with %d retries: %w", attempts, err)
+		return fmt.Errorf("import failed with %d retries: %s", attempts, err)
 	}
 	im.s.Lock()
 	im.s.retries = attempts
@@ -302,7 +302,7 @@ func (im *Importer) Ping() error {
 	url := fmt.Sprintf("%s/health", im.addr)
 	req, err := http.NewRequest(http.MethodGet, url, nil)
 	if err != nil {
-		return fmt.Errorf("cannot create request to %q: %w", im.addr, err)
+		return fmt.Errorf("cannot create request to %q: %s", im.addr, err)
 	}
 	if im.user != "" {
 		req.SetBasicAuth(im.user, im.password)
@@ -332,7 +332,7 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
 	req, err := http.NewRequest(http.MethodPost, im.importPath, pr)
 	if err != nil {
 		im.importRequestsErrorsTotal.Inc()
-		return fmt.Errorf("cannot create request to %q: %w", im.addr, err)
+		return fmt.Errorf("cannot create request to %q: %s", im.addr, err)
 	}
 	if im.user != "" {
 		req.SetBasicAuth(im.user, im.password)
@@ -352,7 +352,7 @@ func (im *Importer) Import(tsBatch []*TimeSeries) error {
 		zw, err := gzip.NewWriterLevel(w, 1)
 		if err != nil {
 			im.importRequestsErrorsTotal.Inc()
-			return fmt.Errorf("unexpected error when creating gzip writer: %w", err)
+			return fmt.Errorf("unexpected error when creating gzip writer: %s", err)
 		}
 		w = zw
 	}
@@ -411,7 +411,7 @@ var ErrBadRequest = errors.New("bad request")
 func (im *Importer) do(req *http.Request) error {
 	resp, err := im.client.Do(req)
 	if err != nil {
-		return fmt.Errorf("unexpected error when performing request: %w", err)
+		return fmt.Errorf("unexpected error when performing request: %s", err)
 	}
 	defer func() {
 		_ = resp.Body.Close()
@@ -419,7 +419,7 @@ func (im *Importer) do(req *http.Request) error {
 	if resp.StatusCode != http.StatusNoContent {
 		body, err := io.ReadAll(resp.Body)
 		if err != nil {
-			return fmt.Errorf("failed to read response body for status code %d: %w", resp.StatusCode, err)
+			return fmt.Errorf("failed to read response body for status code %d: %s", resp.StatusCode, err)
 		}
 		if resp.StatusCode == http.StatusBadRequest {
 			return fmt.Errorf("%w: unexpected response code %d: %s", ErrBadRequest, resp.StatusCode, string(body))

app/vmctl/vm_native.go

@@ -55,14 +55,14 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
 
 	start, err := vmctlutil.ParseTime(p.filter.TimeStart)
 	if err != nil {
-		return fmt.Errorf("failed to parse %s, provided: %s: %w", vmNativeFilterTimeStart, p.filter.TimeStart, err)
+		return fmt.Errorf("failed to parse %s, provided: %s, error: %w", vmNativeFilterTimeStart, p.filter.TimeStart, err)
 	}
 
 	end := time.Now().In(start.Location())
 	if p.filter.TimeEnd != "" {
 		end, err = vmctlutil.ParseTime(p.filter.TimeEnd)
 		if err != nil {
-			return fmt.Errorf("failed to parse %s, provided: %s: %w", vmNativeFilterTimeEnd, p.filter.TimeEnd, err)
+			return fmt.Errorf("failed to parse %s, provided: %s, error: %w", vmNativeFilterTimeEnd, p.filter.TimeEnd, err)
 		}
 	}
 
@@ -91,7 +91,7 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
 		err := p.runBackfilling(ctx, tenantID, ranges)
 		if err != nil {
 			migrationErrorsTotal.Inc()
-			return fmt.Errorf("migration failed: %w", err)
+			return fmt.Errorf("migration failed: %s", err)
 		}
 
 		if p.interCluster {
@@ -157,7 +157,7 @@ func (p *vmNativeProcessor) runSingle(ctx context.Context, f native.Filter, srcU
 			}
 		default:
 		}
-		return fmt.Errorf("failed to write into %q: %w", p.dst.Addr, err)
+		return fmt.Errorf("failed to write into %q: %s", p.dst.Addr, err)
 	}
 
 	p.s.Lock()
@@ -184,7 +184,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 
 	importAddr, err := vm.AddExtraLabelsToImportPath(importAddr, p.dst.ExtraLabels)
 	if err != nil {
-		return fmt.Errorf("failed to add labels to import path: %w", err)
+		return fmt.Errorf("failed to add labels to import path: %s", err)
 	}
 	dstURL := fmt.Sprintf("%s/%s", p.dst.Addr, importAddr)
 
@@ -222,7 +222,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 		format = fmt.Sprintf(nativeWithBackoffTpl, barPrefix)
 		metricsMap, err = p.explore(ctx, p.src, tenantID, ranges)
 		if err != nil {
-			return fmt.Errorf("failed to explore metric names: %w", err)
+			return fmt.Errorf("failed to explore metric names: %s", err)
 		}
 		if len(metricsMap) == 0 {
 			errMsg := "no metrics found"
@@ -295,7 +295,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 			case <-ctx.Done():
 				return fmt.Errorf("context canceled")
 			case infErr := <-errCh:
-				return fmt.Errorf("export/import error: %w", infErr)
+				return fmt.Errorf("export/import error: %s", infErr)
 			case filterCh <- native.Filter{
 				Match:     match,
 				TimeStart: times[0].Format(time.RFC3339),
@@ -313,7 +313,7 @@ func (p *vmNativeProcessor) runBackfilling(ctx context.Context, tenantID string,
 	close(errCh)
 
 	for err := range errCh {
-		return fmt.Errorf("import process failed: %w", err)
+		return fmt.Errorf("import process failed: %s", err)
 	}
 
 	return nil

app/vmctl/vmctlutil/time_test.go

@@ -20,9 +20,6 @@ func TestGetTime_Failure(t *testing.T) {
 
 	// negative time
 	f("-292273086-05-16T16:47:06Z")
-
-	// relative duration that resolves to a timestamp before 1970
-	f("-9223372036.855")
 }
 
 func TestGetTime_Success(t *testing.T) {
@@ -80,6 +77,9 @@ func TestGetTime_Success(t *testing.T) {
 	// float timestamp representation",
 	f("1562529662.324", time.Date(2019, 7, 7, 20, 01, 02, 324e6, time.UTC))
 
+	// negative timestamp
+	f("-9223372036.855", time.Date(1970, 01, 01, 00, 00, 00, 00, time.UTC))
+
 	// big timestamp
 	f("1223372036855", time.Date(2008, 10, 7, 9, 33, 56, 855e6, time.UTC))
 

app/vminsert/common/insert_ctx.go

@@ -184,7 +184,7 @@ func (ctx *InsertCtx) WriteMetadata(mmpbs []prompb.MetricMetadata) error {
 	}
 	ctx.mms = mms
 
-	err := vmstorage.VMInsertAPI.WriteMetadata(mms)
+	err := vmstorage.AddMetadataRows(mms)
 	if err != nil {
 		return &httpserver.ErrorWithStatusCode{
 			Err:        fmt.Errorf("cannot store metrics metadata: %w", err),
@@ -209,7 +209,7 @@ func (ctx *InsertCtx) WritePromMetadata(mmps []prometheus.Metadata) error {
 	}
 	ctx.mms = mms
 
-	err := vmstorage.VMInsertAPI.WriteMetadata(mms)
+	err := vmstorage.AddMetadataRows(mms)
 	if err != nil {
 		return &httpserver.ErrorWithStatusCode{
 			Err:        fmt.Errorf("cannot store prometheus metrics metadata: %w", err),
@@ -278,7 +278,7 @@ func (ctx *InsertCtx) FlushBufs() error {
 	// since the number of concurrent FlushBufs() calls should be already limited via writeconcurrencylimiter
 	// used at every stream.Parse() call under lib/protoparser/*
 
-	err := vmstorage.VMInsertAPI.WriteRows(ctx.mrs)
+	err := vmstorage.AddRows(ctx.mrs)
 	ctx.Reset(0)
 	if err == nil {
 		return nil

app/vminsert/common/streamaggr.go

@@ -283,7 +283,7 @@ func pushAggregateSeries(tss []prompb.TimeSeries) {
 	}
 	// There is no need in limiting the number of concurrent calls to vmstorage.AddRows() here,
 	// since the number of concurrent pushAggregateSeries() calls should be already limited by lib/streamaggr.
-	if err := vmstorage.VMInsertAPI.WriteRows(ctx.mrs); err != nil {
+	if err := vmstorage.AddRows(ctx.mrs); err != nil {
 		logger.Errorf("cannot flush aggregate series: %s", err)
 	}
 }

app/vminsert/main.go

@@ -89,7 +89,6 @@ var staticServer = http.FileServer(http.FS(staticFiles))
 func Init() {
 	relabel.Init()
 	common.InitStreamAggr()
-	opentelemetry.Init()
 	protoparserutil.StartUnmarshalWorkers()
 	if len(*graphiteListenAddr) > 0 {
 		graphiteServer = graphiteserver.MustStart(*graphiteListenAddr, *graphiteUseProxyProtocol, graphite.InsertHandler)

app/vminsert/opentelemetry/request_handler.go

@@ -20,11 +20,6 @@ var (
 	metadataInserted = metrics.NewCounter(`vm_metadata_rows_inserted_total{type="opentelemetry"}`)
 )
 
-// Init must be called after flag.Parse and before using the opentelemetry package.
-func Init() {
-	stream.InitDecodeOptions()
-}
-
 // InsertHandler processes opentelemetry metrics.
 func InsertHandler(req *http.Request) error {
 	extraLabels, err := protoparserutil.GetExtraLabels(req)

app/vmselect/graphite/metrics_api.go

@@ -1,6 +1,7 @@
 package graphite
 
 import (
+	"flag"
 	"fmt"
 	"math"
 	"net/http"
@@ -20,6 +21,8 @@ import (
 	"github.com/VictoriaMetrics/metricsql"
 )
 
+var maxTagValueSuffixes = flag.Int("search.maxTagValueSuffixesPerSearch", 100e3, "The maximum number of tag value suffixes returned from /metrics/find")
+
 // MetricsFindHandler implements /metrics/find handler.
 //
 // See https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find
@@ -219,11 +222,10 @@ func MetricsIndexHandler(startTime time.Time, w http.ResponseWriter, r *http.Req
 
 // metricsFind searches for label values that match the given qHead and qTail.
 func metricsFind(tr storage.TimeRange, label, qHead, qTail string, delimiter byte, isExpand bool, deadline searchutil.Deadline) ([]string, error) {
-	maxSuffixes := 0 // let vmstorage use its maxTagValueSuffixesPerSearch limit
 	n := strings.IndexAny(qTail, "*{[")
 	if n < 0 {
 		query := qHead + qTail
-		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, maxSuffixes, deadline)
+		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, *maxTagValueSuffixes, deadline)
 		if err != nil {
 			return nil, err
 		}
@@ -243,7 +245,7 @@ func metricsFind(tr storage.TimeRange, label, qHead, qTail string, delimiter byt
 	}
 	if n == len(qTail)-1 && strings.HasSuffix(qTail, "*") {
 		query := qHead + qTail[:len(qTail)-1]
-		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, maxSuffixes, deadline)
+		suffixes, err := netstorage.TagValueSuffixes(nil, tr, label, query, delimiter, *maxTagValueSuffixes, deadline)
 		if err != nil {
 			return nil, err
 		}

app/vmselect/graphite/tags_api.go

@@ -138,9 +138,7 @@ func registerMetrics(startTime time.Time, w http.ResponseWriter, r *http.Request
 		mr.MetricNameRaw = storage.MarshalMetricNameRaw(mr.MetricNameRaw[:0], labels)
 		mr.Timestamp = ct
 	}
-	if err := vmstorage.VMSelectAPI.RegisterMetricNames(nil, mrs, 0); err != nil {
-		return err
-	}
+	vmstorage.RegisterMetricNames(nil, mrs)
 
 	// Return response
 	contentType := "text/plain; charset=utf-8"

app/vmselect/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/stats"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
@@ -35,6 +36,12 @@ var (
 	deleteAuthKey                = flagutil.NewPassword("deleteAuthKey", "authKey for metrics' deletion via /api/v1/admin/tsdb/delete_series and /tags/delSeries. It could be passed via authKey query arg. It overrides -httpAuth.*")
 	metricNamesStatsResetAuthKey = flagutil.NewPassword("metricNamesStatsResetAuthKey", "authKey for resetting metric names usage cache via /api/v1/admin/status/metric_names_stats/reset. It overrides -httpAuth.*. "+
 		"See https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#track-ingested-metrics-usage")
+
+	maxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
+		"It shouldn't be high, since a single request can saturate all the CPU cores, while many concurrently executed requests may require high amounts of memory. "+
+		"See also -search.maxQueueDuration and -search.maxMemoryPerQuery")
+	maxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests "+
+		"limit is reached; see also -search.maxQueryDuration")
 	resetCacheAuthKey    = flagutil.NewPassword("search.resetCacheAuthKey", "Optional authKey for resetting rollup cache via /internal/resetRollupResultCache call. It could be passed via authKey query arg. It overrides -httpAuth.*")
 	logSlowQueryDuration = flag.Duration("search.logSlowQueryDuration", 5*time.Second, "Log queries with execution time exceeding this value. Zero disables slow query logging. "+
 		"See also -search.logQueryMemoryUsage")
@@ -43,21 +50,25 @@ var (
 
 var slowQueries = metrics.NewCounter(`vm_slow_queries_total`)
 
+func getDefaultMaxConcurrentRequests() int {
+	// A single request can saturate all the CPU cores, so there is no sense
+	// in allowing higher number of concurrent requests - they will just contend
+	// for unavailable CPU time.
+	n := min(cgroup.AvailableCPUs()*2, 16)
+	return n
+}
+
 // Init initializes vmselect
-func Init(vmselectMaxConcurrentRequests int, vmselectMaxQueueDuration time.Duration) {
-	tmpDirPath := vmstorage.DataPath() + "/tmp"
+func Init() {
+	tmpDirPath := *vmstorage.DataPath + "/tmp"
 	fs.MustRemoveDirContents(tmpDirPath)
 	netstorage.InitTmpBlocksDir(tmpDirPath)
-	promql.InitRollupResultCache(vmstorage.DataPath() + "/cache/rollupResult")
-
-	maxConcurrentRequests = vmselectMaxConcurrentRequests
-	maxQueueDuration = vmselectMaxQueueDuration
-	concurrencyLimitCh = make(chan struct{}, maxConcurrentRequests)
+	promql.InitRollupResultCache(*vmstorage.DataPath + "/cache/rollupResult")
+	prometheus.InitMaxUniqueTimeseries(*maxConcurrentRequests)
 
+	concurrencyLimitCh = make(chan struct{}, *maxConcurrentRequests)
 	initVMUIConfig()
 	initVMAlertProxy()
-
-	flagutil.RegisterSecretFlag("vmalert.proxyURL")
 }
 
 // Stop stops vmselect
@@ -65,11 +76,7 @@ func Stop() {
 	promql.StopRollupResultCache()
 }
 
-var (
-	maxConcurrentRequests int
-	maxQueueDuration      time.Duration
-	concurrencyLimitCh    chan struct{}
-)
+var concurrencyLimitCh chan struct{}
 
 var (
 	concurrencyLimitReached = metrics.NewCounter(`vm_concurrent_select_limit_reached_total`)
@@ -81,6 +88,9 @@ var (
 	_ = metrics.NewGauge(`vm_concurrent_select_current`, func() float64 {
 		return float64(len(concurrencyLimitCh))
 	})
+	_ = metrics.NewGauge(`vm_search_max_unique_timeseries`, func() float64 {
+		return float64(prometheus.GetMaxUniqueTimeSeries())
+	})
 )
 
 //go:embed vmui
@@ -119,12 +129,12 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	default:
 		// Sleep for a while until giving up. This should resolve short bursts in requests.
 		concurrencyLimitReached.Inc()
-		d := min(searchutil.GetMaxQueryDuration(r), maxQueueDuration)
+		d := min(searchutil.GetMaxQueryDuration(r), *maxQueueDuration)
 		t := timerpool.Get(d)
 		select {
 		case concurrencyLimitCh <- struct{}{}:
 			timerpool.Put(t)
-			qt.Printf("wait in queue because -search.maxConcurrentRequests=%d concurrent requests are executed", maxConcurrentRequests)
+			qt.Printf("wait in queue because -search.maxConcurrentRequests=%d concurrent requests are executed", *maxConcurrentRequests)
 			defer func() { <-concurrencyLimitCh }()
 		case <-r.Context().Done():
 			timerpool.Put(t)
@@ -140,7 +150,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 				Err: fmt.Errorf("couldn't start executing the request in %.3f seconds, since -search.maxConcurrentRequests=%d concurrent requests "+
 					"are executed. Possible solutions: to reduce query load; to add more compute resources to the server; "+
 					"to increase -search.maxQueueDuration=%s; to increase -search.maxQueryDuration; to increase -search.maxConcurrentRequests",
-					d.Seconds(), maxConcurrentRequests, maxQueueDuration),
+					d.Seconds(), *maxConcurrentRequests, maxQueueDuration),
 				StatusCode: http.StatusTooManyRequests,
 			}
 			w.Header().Add("Retry-After", "10")
@@ -252,7 +262,6 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	case "/api/v1/export":
 		exportRequests.Inc()
-		httpserver.EnableCORS(w, r)
 		if err := prometheus.ExportHandler(startTime, w, r); err != nil {
 			exportErrors.Inc()
 			httpserver.Errorf(w, r, "%s", err)
@@ -261,7 +270,6 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	case "/api/v1/export/csv":
 		exportCSVRequests.Inc()
-		httpserver.EnableCORS(w, r)
 		if err := prometheus.ExportCSVHandler(startTime, w, r); err != nil {
 			exportCSVErrors.Inc()
 			httpserver.Errorf(w, r, "%s", err)
@@ -270,7 +278,6 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 		return true
 	case "/api/v1/export/native":
 		exportNativeRequests.Inc()
-		httpserver.EnableCORS(w, r)
 		if err := prometheus.ExportNativeHandler(startTime, w, r); err != nil {
 			exportNativeErrors.Inc()
 			httpserver.Errorf(w, r, "%s", err)

app/vmselect/netstorage/netstorage.go

@@ -27,6 +27,10 @@ import (
 )
 
 var (
+	maxTagKeysPerSearch = flag.Int("search.maxTagKeys", 100e3, "The maximum number of tag keys returned from /api/v1/labels . "+
+		"See also -search.maxLabelsAPISeries and -search.maxLabelsAPIDuration")
+	maxTagValuesPerSearch = flag.Int("search.maxTagValues", 100e3, "The maximum number of tag values returned from /api/v1/label/<label_name>/values . "+
+		"See also -search.maxLabelsAPISeries and -search.maxLabelsAPIDuration")
 	maxSamplesPerSeries = flag.Int("search.maxSamplesPerSeries", 30e6, "The maximum number of raw samples a single query can scan per each time series. This option allows limiting memory usage")
 	maxSamplesPerQuery  = flag.Int("search.maxSamplesPerQuery", 1e9, "The maximum number of raw samples a single query can process across all time series. "+
 		"This protects from heavy queries, which select unexpectedly high number of raw samples. See also -search.maxSamplesPerSeries")
@@ -76,7 +80,7 @@ func (rss *Results) Cancel() {
 }
 
 func (rss *Results) mustClose() {
-	vmstorage.PutSearch(rss.sr)
+	putStorageSearch(rss.sr)
 	rss.sr = nil
 	putTmpBlocksFile(rss.tbf)
 	rss.tbf = nil
@@ -754,7 +758,12 @@ var sbhPool sync.Pool
 func DeleteSeries(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline searchutil.Deadline) (int, error) {
 	qt = qt.NewChild("delete series: %s", sq)
 	defer qt.Done()
-	return vmstorage.VMSelectAPI.DeleteSeries(qt, sq, deadline.Deadline())
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return 0, err
+	}
+	return vmstorage.DeleteSeries(qt, tfss, sq.MaxMetrics)
 }
 
 // LabelNames returns label names matching the given sq until the given deadline.
@@ -764,7 +773,15 @@ func LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames i
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	labels, err := vmstorage.VMSelectAPI.LabelNames(qt, sq, maxLabelNames, deadline.Deadline())
+	if maxLabelNames > *maxTagKeysPerSearch || maxLabelNames <= 0 {
+		maxLabelNames = *maxTagKeysPerSearch
+	}
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+	labels, err := vmstorage.SearchLabelNames(qt, tfss, tr, maxLabelNames, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during labels search on time range: %w", err)
 	}
@@ -824,7 +841,15 @@ func LabelValues(qt *querytracer.Tracer, labelName string, sq *storage.SearchQue
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	labelValues, err := vmstorage.VMSelectAPI.LabelValues(qt, sq, labelName, maxLabelValues, deadline.Deadline())
+	if maxLabelValues > *maxTagValuesPerSearch || maxLabelValues <= 0 {
+		maxLabelValues = *maxTagValuesPerSearch
+	}
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+	labelValues, err := vmstorage.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during label values search on time range for labelName=%q: %w", labelName, err)
 	}
@@ -839,10 +864,7 @@ func GetMetricsMetadata(qt *querytracer.Tracer, limit int, metricName string) ([
 	qt = qt.NewChild("get metrics metadata: limit=%d, metric_name=%q", limit, metricName)
 	defer qt.Done()
 
-	metadata, err := vmstorage.VMSelectAPI.GetMetadataRecords(qt, nil, limit, metricName, 0)
-	if err != nil {
-		return nil, err
-	}
+	metadata := vmstorage.Storage.GetMetadataRows(qt, limit, metricName)
 
 	sort.Slice(metadata, func(i, j int) bool {
 		return string(metadata[i].MetricFamilyName) < string(metadata[j].MetricFamilyName)
@@ -890,11 +912,16 @@ func TagValueSuffixes(qt *querytracer.Tracer, tr storage.TimeRange, tagKey, tagV
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	suffixes, err := vmstorage.VMSelectAPI.TagValueSuffixes(qt, 0, 0, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline.Deadline())
+	suffixes, err := vmstorage.SearchTagValueSuffixes(qt, tr, tagKey, tagValuePrefix, delimiter, maxSuffixes, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during search for suffixes for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s: %w",
 			tagKey, tagValuePrefix, delimiter, tr.String(), err)
 	}
+	if len(suffixes) >= maxSuffixes {
+		return nil, fmt.Errorf("more than -search.maxTagValueSuffixesPerSearch=%d tag value suffixes found for tagKey=%q, tagValuePrefix=%q, delimiter=%c on time range %s; "+
+			"either narrow down the query or increase -search.maxTagValueSuffixesPerSearch command-line flag value",
+			maxSuffixes, tagKey, tagValuePrefix, delimiter, tr.String())
+	}
 	return suffixes, nil
 }
 
@@ -907,7 +934,13 @@ func TSDBStatus(qt *querytracer.Tracer, sq *storage.SearchQuery, focusLabel stri
 	if deadline.Exceeded() {
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	status, err := vmstorage.VMSelectAPI.TSDBStatus(qt, sq, focusLabel, topN, deadline.Deadline())
+	tr := sq.GetTimeRange()
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+	date := uint64(tr.MinTimestamp) / (3600 * 24 * 1000)
+	status, err := vmstorage.GetTSDBStatus(qt, tfss, date, focusLabel, topN, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during tsdb status request: %w", err)
 	}
@@ -921,13 +954,28 @@ func SeriesCount(qt *querytracer.Tracer, deadline searchutil.Deadline) (uint64,
 	if deadline.Exceeded() {
 		return 0, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
-	n, err := vmstorage.VMSelectAPI.SeriesCount(qt, 0, 0, deadline.Deadline())
+	n, err := vmstorage.GetSeriesCount(deadline.Deadline())
 	if err != nil {
 		return 0, fmt.Errorf("error during series count request: %w", err)
 	}
 	return n, nil
 }
 
+func getStorageSearch() *storage.Search {
+	v := ssPool.Get()
+	if v == nil {
+		return &storage.Search{}
+	}
+	return v.(*storage.Search)
+}
+
+func putStorageSearch(sr *storage.Search) {
+	sr.MustClose()
+	ssPool.Put(sr)
+}
+
+var ssPool sync.Pool
+
 // ExportBlocks searches for time series matching sq and calls f for each found block.
 //
 // f is called in parallel from multiple goroutines.
@@ -941,13 +989,21 @@ func ExportBlocks(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline sear
 	if deadline.Exceeded() {
 		return fmt.Errorf("timeout exceeded before starting data export: %s", deadline.String())
 	}
-
 	tr := sq.GetTimeRange()
-	sr, _, err := vmstorage.GetSearch(qt, sq, deadline.Deadline())
+	if err := vmstorage.CheckTimeRange(tr); err != nil {
+		return err
+	}
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
 	if err != nil {
 		return err
 	}
-	defer vmstorage.PutSearch(sr)
+
+	vmstorage.WG.Add(1)
+	defer vmstorage.WG.Done()
+
+	sr := getStorageSearch()
+	defer putStorageSearch(sr)
+	sr.Init(qt, vmstorage.Storage, tfss, tr, sq.MaxMetrics, deadline.Deadline())
 
 	// Start workers that call f in parallel on available CPU cores.
 	workCh := make(chan *exportWork, gomaxprocs*8)
@@ -1040,7 +1096,17 @@ func SearchMetricNames(qt *querytracer.Tracer, sq *storage.SearchQuery, deadline
 		return nil, fmt.Errorf("timeout exceeded before starting to search metric names: %s", deadline.String())
 	}
 
-	metricNames, err := vmstorage.VMSelectAPI.SearchMetricNames(qt, sq, deadline.Deadline())
+	// Setup search.
+	tr := sq.GetTimeRange()
+	if err := vmstorage.CheckTimeRange(tr); err != nil {
+		return nil, err
+	}
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
+	if err != nil {
+		return nil, err
+	}
+
+	metricNames, err := vmstorage.SearchMetricNames(qt, tfss, tr, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("cannot find metric names: %w", err)
 	}
@@ -1059,11 +1125,21 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		return nil, fmt.Errorf("timeout exceeded before starting the query processing: %s", deadline.String())
 	}
 
-	sr, maxSeriesCount, err := vmstorage.GetSearch(qt, sq, deadline.Deadline())
+	// Setup search.
+	tr := sq.GetTimeRange()
+	if err := vmstorage.CheckTimeRange(tr); err != nil {
+		return nil, err
+	}
+	tfss, err := setupTfss(qt, tr, sq.TagFilterss, sq.MaxMetrics, deadline)
 	if err != nil {
 		return nil, err
 	}
 
+	vmstorage.WG.Add(1)
+	defer vmstorage.WG.Done()
+
+	sr := getStorageSearch()
+	maxSeriesCount := sr.Init(qt, vmstorage.Storage, tfss, tr, sq.MaxMetrics, deadline.Deadline())
 	type blockRefs struct {
 		brs []blockRef
 	}
@@ -1101,7 +1177,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		blocksRead++
 		if deadline.Exceeded() {
 			putTmpBlocksFile(tbf)
-			vmstorage.PutSearch(sr)
+			putStorageSearch(sr)
 			return nil, fmt.Errorf("timeout exceeded while fetching data block #%d from storage: %s", blocksRead, deadline.String())
 		}
 		br := sr.MetricBlockRef.BlockRef
@@ -1113,7 +1189,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		samples += br.RowsCount()
 		if *maxSamplesPerQuery > 0 && samples > *maxSamplesPerQuery {
 			putTmpBlocksFile(tbf)
-			vmstorage.PutSearch(sr)
+			putStorageSearch(sr)
 			return nil, fmt.Errorf("cannot select more than -search.maxSamplesPerQuery=%d samples; possible solutions: increase the -search.maxSamplesPerQuery; "+
 				"reduce time range for the query; use more specific label filters in order to select fewer series", *maxSamplesPerQuery)
 		}
@@ -1122,7 +1198,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 		addr, err := tbf.WriteBlockRefData(buf)
 		if err != nil {
 			putTmpBlocksFile(tbf)
-			vmstorage.PutSearch(sr)
+			putStorageSearch(sr)
 			return nil, fmt.Errorf("cannot write %d bytes to temporary file: %w", len(buf), err)
 		}
 
@@ -1180,7 +1256,7 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 
 	if err := sr.Error(); err != nil {
 		putTmpBlocksFile(tbf)
-		vmstorage.PutSearch(sr)
+		putStorageSearch(sr)
 		if errors.Is(err, storage.ErrDeadlineExceeded) {
 			return nil, fmt.Errorf("timeout exceeded during the query: %s", deadline.String())
 		}
@@ -1188,13 +1264,13 @@ func ProcessSearchQuery(qt *querytracer.Tracer, sq *storage.SearchQuery, deadlin
 	}
 	if err := tbf.Finalize(); err != nil {
 		putTmpBlocksFile(tbf)
-		vmstorage.PutSearch(sr)
+		putStorageSearch(sr)
 		return nil, fmt.Errorf("cannot finalize temporary file: %w", err)
 	}
 	qt.Printf("fetch unique series=%d, blocks=%d, samples=%d, bytes=%d", len(m), blocksRead, samples, tbf.Len())
 
 	var rss Results
-	rss.tr = sq.GetTimeRange()
+	rss.tr = tr
 	rss.deadline = deadline
 	pts := make([]packedTimeseries, len(orderedMetricNames))
 	for i, metricName := range orderedMetricNames {
@@ -1235,6 +1311,35 @@ func getBlockRefsEnd(a []blockRef) uintptr {
 	return uintptr(unsafe.Pointer(unsafe.SliceData(a))) + uintptr(len(a))*unsafe.Sizeof(blockRef{})
 }
 
+func setupTfss(qt *querytracer.Tracer, tr storage.TimeRange, tagFilterss [][]storage.TagFilter, maxMetrics int, deadline searchutil.Deadline) ([]*storage.TagFilters, error) {
+	tfss := make([]*storage.TagFilters, 0, len(tagFilterss))
+	for _, tagFilters := range tagFilterss {
+		tfs := storage.NewTagFilters()
+		for i := range tagFilters {
+			tf := &tagFilters[i]
+			if string(tf.Key) == "__graphite__" {
+				query := tf.Value
+				paths, err := vmstorage.SearchGraphitePaths(qt, tr, query, maxMetrics, deadline.Deadline())
+				if err != nil {
+					return nil, fmt.Errorf("error when searching for Graphite paths for query %q: %w", query, err)
+				}
+				if len(paths) >= maxMetrics {
+					return nil, fmt.Errorf("more than %d time series match Graphite query %q; "+
+						"either narrow down the query or increase the corresponding -search.max* command-line flag value; "+
+						"see https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/#resource-usage-limits", maxMetrics, query)
+				}
+				tfs.AddGraphiteQuery(query, paths, tf.IsNegative)
+				continue
+			}
+			if err := tfs.Add(tf.Key, tf.Value, tf.IsNegative, tf.IsRegexp); err != nil {
+				return nil, fmt.Errorf("cannot parse tag filter %s: %w", tf, err)
+			}
+		}
+		tfss = append(tfss, tfs)
+	}
+	return tfss, nil
+}
+
 func applyGraphiteRegexpFilter(filter string, ss []string) ([]string, error) {
 	// Anchor filter regexp to the beginning of the string as Graphite does.
 	// See https://github.com/graphite-project/graphite-web/blob/3ad279df5cb90b211953e39161df416e54a84948/webapp/graphite/tags/localdatabase.py#L157
@@ -1261,12 +1366,13 @@ const maxFastAllocBlockSize = 32 * 1024
 func GetMetricNamesStats(qt *querytracer.Tracer, limit, le int, matchPattern string) (metricnamestats.StatsResult, error) {
 	qt = qt.NewChild("get metric names usage statistics with limit: %d, less or equal to: %d, match pattern=%q", limit, le, matchPattern)
 	defer qt.Done()
-	return vmstorage.VMSelectAPI.GetMetricNamesUsageStats(qt, nil, limit, le, matchPattern, 0)
+	return vmstorage.GetMetricNamesStats(qt, limit, le, matchPattern)
 }
 
 // ResetMetricNamesStats resets state of metric names usage
 func ResetMetricNamesStats(qt *querytracer.Tracer) error {
 	qt = qt.NewChild("reset metric names usage stats")
 	defer qt.Done()
-	return vmstorage.VMSelectAPI.ResetMetricNamesUsageStats(qt, 0)
+	vmstorage.ResetMetricNamesStats(qt)
+	return nil
 }

app/vmselect/prometheus/federate.qtpl

@@ -2,16 +2,13 @@
 	"math"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-    "github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
-    "github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 ) %}
 
 {% stripspace %}
 
 // Federate writes rs in /federate format.
 // See https://prometheus.io/docs/prometheus/latest/federation/
-{% func Federate(rs *netstorage.Result, escapeScheme string) %}
+{% func Federate(rs *netstorage.Result) %}
 	{% code
 		values := rs.Values
 		timestamps := rs.Timestamps
@@ -27,54 +24,10 @@
 			See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3185
 		{% endcomment %}
 		{% return %}
-    {% endif %}
-
-    {% switch escapeScheme %}
-       {% case federateEscapeSchemeUTF8 %}
-          {%= prometheusFederateMetricNameUTF8(&rs.MetricName) %}{% space %}
-
-       {% case federateEscapeSchemeUnderscore %}
-         {%= prometheusFederateMetricNameEscapeUnderscore(&rs.MetricName) %}{% space %}
-
-       {% case "" %}
-         {%= prometheusMetricName(&rs.MetricName) %}{% space %}
-    {% endswitch %}
-
+	{% endif %}
+	{%= prometheusMetricName(&rs.MetricName) %}{% space %}
 	{%f= lastValue %}{% space %}
 	{%dl= timestamps[len(timestamps)-1] %}{% newline %}
 {% endfunc %}
 
-{% func prometheusFederateMetricNameEscapeUnderscore(mn *storage.MetricName) %}
-	{%s= promrelabel.SanitizeMetricName(bytesutil.ToUnsafeString(mn.MetricGroup)) %}
-	{% if len(mn.Tags) > 0 %}
-	{
-		{% code tags := mn.Tags %}
-		{%s= promrelabel.SanitizeLabelName(bytesutil.ToUnsafeString(tags[0].Key)) %}={%= escapePrometheusLabel(tags[0].Value) %}
-		{% code tags = tags[1:] %}
-		{% for i := range tags %}
-			{% code tag := &tags[i] %}
-			,{%s= promrelabel.SanitizeLabelName(bytesutil.ToUnsafeString(tag.Key)) %}={%= escapePrometheusLabel(tag.Value) %}
-		{% endfor %}
-	}
-	{% endif %}
-{% endfunc %}
-
-{% func prometheusFederateMetricNameUTF8(mn *storage.MetricName) %}
-	{
-        {%= escapePrometheusLabel(mn.MetricGroup) %}
-        {% if len(mn.Tags) > 0 %}
-        ,
-		{% code tags := mn.Tags %}
-		{%= escapePrometheusLabel(tags[0].Key) %}={%= escapePrometheusLabel(tags[0].Value) %}
-		{% code tags = tags[1:] %}
-		{% for i := range tags %}
-			{% code tag := &tags[i] %}
-			,{%= escapePrometheusLabel(tag.Key) %}={%= escapePrometheusLabel(tag.Value) %}
-		{% endfor %}
-        {% endif %}
-	}
-{% endfunc %}
-
-
-
 {% endstripspace %}

app/vmselect/prometheus/federate.qtpl.go

@@ -9,241 +9,82 @@ import (
 	"math"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
 )
 
 // Federate writes rs in /federate format.// See https://prometheus.io/docs/prometheus/latest/federation/
 
-//line app/vmselect/prometheus/federate.qtpl:14
+//line app/vmselect/prometheus/federate.qtpl:11
 import (
 	qtio422016 "io"
 
 	qt422016 "github.com/valyala/quicktemplate"
 )
 
-//line app/vmselect/prometheus/federate.qtpl:14
+//line app/vmselect/prometheus/federate.qtpl:11
 var (
 	_ = qtio422016.Copy
 	_ = qt422016.AcquireByteBuffer
 )
 
-//line app/vmselect/prometheus/federate.qtpl:14
-func StreamFederate(qw422016 *qt422016.Writer, rs *netstorage.Result, escapeScheme string) {
-//line app/vmselect/prometheus/federate.qtpl:16
+//line app/vmselect/prometheus/federate.qtpl:11
+func StreamFederate(qw422016 *qt422016.Writer, rs *netstorage.Result) {
+//line app/vmselect/prometheus/federate.qtpl:13
 	values := rs.Values
 	timestamps := rs.Timestamps
 
-//line app/vmselect/prometheus/federate.qtpl:19
+//line app/vmselect/prometheus/federate.qtpl:16
 	if len(timestamps) == 0 || len(values) == 0 {
-//line app/vmselect/prometheus/federate.qtpl:19
+//line app/vmselect/prometheus/federate.qtpl:16
 		return
-//line app/vmselect/prometheus/federate.qtpl:19
+//line app/vmselect/prometheus/federate.qtpl:16
 	}
-//line app/vmselect/prometheus/federate.qtpl:21
+//line app/vmselect/prometheus/federate.qtpl:18
 	lastValue := values[len(values)-1]
 
-//line app/vmselect/prometheus/federate.qtpl:23
+//line app/vmselect/prometheus/federate.qtpl:20
 	if math.IsNaN(lastValue) {
-//line app/vmselect/prometheus/federate.qtpl:29
+//line app/vmselect/prometheus/federate.qtpl:26
 		return
-//line app/vmselect/prometheus/federate.qtpl:30
+//line app/vmselect/prometheus/federate.qtpl:27
 	}
-//line app/vmselect/prometheus/federate.qtpl:32
-	switch escapeScheme {
-//line app/vmselect/prometheus/federate.qtpl:33
-	case federateEscapeSchemeUTF8:
-//line app/vmselect/prometheus/federate.qtpl:34
-		streamprometheusFederateMetricNameUTF8(qw422016, &rs.MetricName)
-//line app/vmselect/prometheus/federate.qtpl:34
-		qw422016.N().S(` `)
-//line app/vmselect/prometheus/federate.qtpl:36
-	case federateEscapeSchemeUnderscore:
-//line app/vmselect/prometheus/federate.qtpl:37
-		streamprometheusFederateMetricNameEscapeUnderscore(qw422016, &rs.MetricName)
-//line app/vmselect/prometheus/federate.qtpl:37
-		qw422016.N().S(` `)
-//line app/vmselect/prometheus/federate.qtpl:39
-	case "":
-//line app/vmselect/prometheus/federate.qtpl:40
-		streamprometheusMetricName(qw422016, &rs.MetricName)
-//line app/vmselect/prometheus/federate.qtpl:40
-		qw422016.N().S(` `)
-//line app/vmselect/prometheus/federate.qtpl:41
-	}
-//line app/vmselect/prometheus/federate.qtpl:43
-	qw422016.N().F(lastValue)
-//line app/vmselect/prometheus/federate.qtpl:43
+//line app/vmselect/prometheus/federate.qtpl:28
+	streamprometheusMetricName(qw422016, &rs.MetricName)
+//line app/vmselect/prometheus/federate.qtpl:28
 	qw422016.N().S(` `)
-//line app/vmselect/prometheus/federate.qtpl:44
+//line app/vmselect/prometheus/federate.qtpl:29
+	qw422016.N().F(lastValue)
+//line app/vmselect/prometheus/federate.qtpl:29
+	qw422016.N().S(` `)
+//line app/vmselect/prometheus/federate.qtpl:30
 	qw422016.N().DL(timestamps[len(timestamps)-1])
-//line app/vmselect/prometheus/federate.qtpl:44
+//line app/vmselect/prometheus/federate.qtpl:30
 	qw422016.N().S(`
 `)
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
 }
 
-//line app/vmselect/prometheus/federate.qtpl:45
-func WriteFederate(qq422016 qtio422016.Writer, rs *netstorage.Result, escapeScheme string) {
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
+func WriteFederate(qq422016 qtio422016.Writer, rs *netstorage.Result) {
+//line app/vmselect/prometheus/federate.qtpl:31
 	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vmselect/prometheus/federate.qtpl:45
-	StreamFederate(qw422016, rs, escapeScheme)
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
+	StreamFederate(qw422016, rs)
+//line app/vmselect/prometheus/federate.qtpl:31
 	qt422016.ReleaseWriter(qw422016)
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
 }
 
-//line app/vmselect/prometheus/federate.qtpl:45
-func Federate(rs *netstorage.Result, escapeScheme string) string {
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
+func Federate(rs *netstorage.Result) string {
+//line app/vmselect/prometheus/federate.qtpl:31
 	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vmselect/prometheus/federate.qtpl:45
-	WriteFederate(qb422016, rs, escapeScheme)
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
+	WriteFederate(qb422016, rs)
+//line app/vmselect/prometheus/federate.qtpl:31
 	qs422016 := string(qb422016.B)
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
 	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vmselect/prometheus/federate.qtpl:45
+//line app/vmselect/prometheus/federate.qtpl:31
 	return qs422016
-//line app/vmselect/prometheus/federate.qtpl:45
-}
-
-//line app/vmselect/prometheus/federate.qtpl:47
-func streamprometheusFederateMetricNameEscapeUnderscore(qw422016 *qt422016.Writer, mn *storage.MetricName) {
-//line app/vmselect/prometheus/federate.qtpl:48
-	qw422016.N().S(promrelabel.SanitizeMetricName(bytesutil.ToUnsafeString(mn.MetricGroup)))
-//line app/vmselect/prometheus/federate.qtpl:49
-	if len(mn.Tags) > 0 {
-//line app/vmselect/prometheus/federate.qtpl:49
-		qw422016.N().S(`{`)
-//line app/vmselect/prometheus/federate.qtpl:51
-		tags := mn.Tags
-
-//line app/vmselect/prometheus/federate.qtpl:52
-		qw422016.N().S(promrelabel.SanitizeLabelName(bytesutil.ToUnsafeString(tags[0].Key)))
-//line app/vmselect/prometheus/federate.qtpl:52
-		qw422016.N().S(`=`)
-//line app/vmselect/prometheus/federate.qtpl:52
-		streamescapePrometheusLabel(qw422016, tags[0].Value)
-//line app/vmselect/prometheus/federate.qtpl:53
-		tags = tags[1:]
-
-//line app/vmselect/prometheus/federate.qtpl:54
-		for i := range tags {
-//line app/vmselect/prometheus/federate.qtpl:55
-			tag := &tags[i]
-
-//line app/vmselect/prometheus/federate.qtpl:55
-			qw422016.N().S(`,`)
-//line app/vmselect/prometheus/federate.qtpl:56
-			qw422016.N().S(promrelabel.SanitizeLabelName(bytesutil.ToUnsafeString(tag.Key)))
-//line app/vmselect/prometheus/federate.qtpl:56
-			qw422016.N().S(`=`)
-//line app/vmselect/prometheus/federate.qtpl:56
-			streamescapePrometheusLabel(qw422016, tag.Value)
-//line app/vmselect/prometheus/federate.qtpl:57
-		}
-//line app/vmselect/prometheus/federate.qtpl:57
-		qw422016.N().S(`}`)
-//line app/vmselect/prometheus/federate.qtpl:59
-	}
-//line app/vmselect/prometheus/federate.qtpl:60
-}
-
-//line app/vmselect/prometheus/federate.qtpl:60
-func writeprometheusFederateMetricNameEscapeUnderscore(qq422016 qtio422016.Writer, mn *storage.MetricName) {
-//line app/vmselect/prometheus/federate.qtpl:60
-	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vmselect/prometheus/federate.qtpl:60
-	streamprometheusFederateMetricNameEscapeUnderscore(qw422016, mn)
-//line app/vmselect/prometheus/federate.qtpl:60
-	qt422016.ReleaseWriter(qw422016)
-//line app/vmselect/prometheus/federate.qtpl:60
-}
-
-//line app/vmselect/prometheus/federate.qtpl:60
-func prometheusFederateMetricNameEscapeUnderscore(mn *storage.MetricName) string {
-//line app/vmselect/prometheus/federate.qtpl:60
-	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vmselect/prometheus/federate.qtpl:60
-	writeprometheusFederateMetricNameEscapeUnderscore(qb422016, mn)
-//line app/vmselect/prometheus/federate.qtpl:60
-	qs422016 := string(qb422016.B)
-//line app/vmselect/prometheus/federate.qtpl:60
-	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vmselect/prometheus/federate.qtpl:60
-	return qs422016
-//line app/vmselect/prometheus/federate.qtpl:60
-}
-
-//line app/vmselect/prometheus/federate.qtpl:62
-func streamprometheusFederateMetricNameUTF8(qw422016 *qt422016.Writer, mn *storage.MetricName) {
-//line app/vmselect/prometheus/federate.qtpl:62
-	qw422016.N().S(`{`)
-//line app/vmselect/prometheus/federate.qtpl:64
-	streamescapePrometheusLabel(qw422016, mn.MetricGroup)
-//line app/vmselect/prometheus/federate.qtpl:65
-	if len(mn.Tags) > 0 {
-//line app/vmselect/prometheus/federate.qtpl:65
-		qw422016.N().S(`,`)
-//line app/vmselect/prometheus/federate.qtpl:67
-		tags := mn.Tags
-
-//line app/vmselect/prometheus/federate.qtpl:68
-		streamescapePrometheusLabel(qw422016, tags[0].Key)
-//line app/vmselect/prometheus/federate.qtpl:68
-		qw422016.N().S(`=`)
-//line app/vmselect/prometheus/federate.qtpl:68
-		streamescapePrometheusLabel(qw422016, tags[0].Value)
-//line app/vmselect/prometheus/federate.qtpl:69
-		tags = tags[1:]
-
-//line app/vmselect/prometheus/federate.qtpl:70
-		for i := range tags {
-//line app/vmselect/prometheus/federate.qtpl:71
-			tag := &tags[i]
-
-//line app/vmselect/prometheus/federate.qtpl:71
-			qw422016.N().S(`,`)
-//line app/vmselect/prometheus/federate.qtpl:72
-			streamescapePrometheusLabel(qw422016, tag.Key)
-//line app/vmselect/prometheus/federate.qtpl:72
-			qw422016.N().S(`=`)
-//line app/vmselect/prometheus/federate.qtpl:72
-			streamescapePrometheusLabel(qw422016, tag.Value)
-//line app/vmselect/prometheus/federate.qtpl:73
-		}
-//line app/vmselect/prometheus/federate.qtpl:74
-	}
-//line app/vmselect/prometheus/federate.qtpl:74
-	qw422016.N().S(`}`)
-//line app/vmselect/prometheus/federate.qtpl:76
-}
-
-//line app/vmselect/prometheus/federate.qtpl:76
-func writeprometheusFederateMetricNameUTF8(qq422016 qtio422016.Writer, mn *storage.MetricName) {
-//line app/vmselect/prometheus/federate.qtpl:76
-	qw422016 := qt422016.AcquireWriter(qq422016)
-//line app/vmselect/prometheus/federate.qtpl:76
-	streamprometheusFederateMetricNameUTF8(qw422016, mn)
-//line app/vmselect/prometheus/federate.qtpl:76
-	qt422016.ReleaseWriter(qw422016)
-//line app/vmselect/prometheus/federate.qtpl:76
-}
-
-//line app/vmselect/prometheus/federate.qtpl:76
-func prometheusFederateMetricNameUTF8(mn *storage.MetricName) string {
-//line app/vmselect/prometheus/federate.qtpl:76
-	qb422016 := qt422016.AcquireByteBuffer()
-//line app/vmselect/prometheus/federate.qtpl:76
-	writeprometheusFederateMetricNameUTF8(qb422016, mn)
-//line app/vmselect/prometheus/federate.qtpl:76
-	qs422016 := string(qb422016.B)
-//line app/vmselect/prometheus/federate.qtpl:76
-	qt422016.ReleaseByteBuffer(qb422016)
-//line app/vmselect/prometheus/federate.qtpl:76
-	return qs422016
-//line app/vmselect/prometheus/federate.qtpl:76
+//line app/vmselect/prometheus/federate.qtpl:31
 }

app/vmselect/prometheus/federate_test.go

@@ -8,15 +8,15 @@ import (
 )
 
 func TestFederate(t *testing.T) {
-	f := func(rs *netstorage.Result, escapeScheme string, expectedResult string) {
+	f := func(rs *netstorage.Result, expectedResult string) {
 		t.Helper()
-		result := Federate(rs, escapeScheme)
+		result := Federate(rs)
 		if result != expectedResult {
 			t.Fatalf("unexpected result; got\n%s\nwant\n%s", result, expectedResult)
 		}
 	}
 
-	f(&netstorage.Result{}, ``, ``)
+	f(&netstorage.Result{}, ``)
 
 	f(&netstorage.Result{
 		MetricName: storage.MetricName{
@@ -39,60 +39,5 @@ func TestFederate(t *testing.T) {
 		},
 		Values:     []float64{1.23},
 		Timestamps: []int64{123},
-	}, ``, `foo{a="b",qqq="\\",abc="a<b\"\\c"} 1.23 123`+"\n")
-
-	f(&netstorage.Result{
-		MetricName: storage.MetricName{
-			MetricGroup: []byte("foo.bar"),
-			Tags: []storage.Tag{
-				{
-					Key:   []byte("some.!other"),
-					Value: []byte("value.unchanged!."),
-				},
-				{
-					Key:   []byte("qqq"),
-					Value: []byte("\\"),
-				},
-				{
-					Key:   []byte("!key"),
-					Value: []byte("value"),
-				},
-				{
-					Key: []byte("abc"),
-					// Verify that < isn't encoded. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5431
-					Value: []byte("a<b\"\\c"),
-				},
-			},
-		},
-		Values:     []float64{1.23},
-		Timestamps: []int64{123},
-	}, federateEscapeSchemeUnderscore, `foo_bar{some__other="value.unchanged!.",qqq="\\",_key="value",abc="a<b\"\\c"} 1.23 123`+"\n")
-
-	f(&netstorage.Result{
-		MetricName: storage.MetricName{
-			MetricGroup: []byte("foo.bar"),
-			Tags: []storage.Tag{
-				{
-					Key:   []byte("some.!other"),
-					Value: []byte("value.unchanged!."),
-				},
-				{
-					Key:   []byte("qqq"),
-					Value: []byte("\\"),
-				},
-				{
-					Key:   []byte("!key"),
-					Value: []byte("value"),
-				},
-				{
-					Key: []byte(`ab"c`),
-					// Verify that < isn't encoded. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5431
-					Value: []byte("a<b\"\\c"),
-				},
-			},
-		},
-		Values:     []float64{1.23},
-		Timestamps: []int64{123},
-	}, federateEscapeSchemeUTF8, `{"foo.bar","some.!other"="value.unchanged!.","qqq"="\\","!key"="value","ab\"c"="a<b\"\\c"} 1.23 123`+"\n")
-
+	}, `foo{a="b",qqq="\\",abc="a<b\"\\c"} 1.23 123`+"\n")
 }

app/vmselect/prometheus/federate_timing_test.go

@@ -9,17 +9,16 @@ import (
 )
 
 func BenchmarkFederate(b *testing.B) {
-
 	rs := &netstorage.Result{
 		MetricName: storage.MetricName{
-			MetricGroup: []byte("foo_bar_?_._bazaaaa_total"),
+			MetricGroup: []byte("foo_bar_bazaaaa_total"),
 			Tags: []storage.Tag{
 				{
-					Key:   []byte("instance:job"),
+					Key:   []byte("instance"),
 					Value: []byte("foobarbaz:2344"),
 				},
 				{
-					Key:   []byte("job.name"),
+					Key:   []byte("job"),
 					Value: []byte("aaabbbccc"),
 				},
 			},
@@ -28,22 +27,12 @@ func BenchmarkFederate(b *testing.B) {
 		Timestamps: []int64{1234567890},
 	}
 
-	f := func(name, escapeScheme string) {
-		b.Helper()
-
-		b.Run(name, func(b *testing.B) {
-			b.ReportAllocs()
-			b.RunParallel(func(pb *testing.PB) {
-				var bb bytes.Buffer
-				for pb.Next() {
-					bb.Reset()
-					WriteFederate(&bb, rs, escapeScheme)
-				}
-			})
-		})
-	}
-
-	f("without escape", "")
-	f("allow-utf-8", federateEscapeSchemeUTF8)
-	f("legacy-underscore", federateEscapeSchemeUnderscore)
+	b.ReportAllocs()
+	b.RunParallel(func(pb *testing.PB) {
+		var bb bytes.Buffer
+		for pb.Next() {
+			bb.Reset()
+			WriteFederate(&bb, rs)
+		}
+	})
 }

app/vmselect/prometheus/prometheus.go

@@ -28,6 +28,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httputil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/querytracer"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/storage"
@@ -48,6 +50,9 @@ var (
 		"If set to true, the query model becomes closer to InfluxDB data model. If set to true, then -search.maxLookback and -search.maxStalenessInterval are ignored")
 	maxStepForPointsAdjustment = flag.Duration("search.maxStepForPointsAdjustment", time.Minute, "The maximum step when /api/v1/query_range handler adjusts "+
 		"points with timestamps closer than -search.latencyOffset to the current time. The adjustment is needed because such points may contain incomplete data")
+
+	maxUniqueTimeseries = flag.Int("search.maxUniqueTimeseries", 0, "The maximum number of unique time series, which can be selected during /api/v1/query and /api/v1/query_range queries. This option allows limiting memory usage. "+
+		"When set to zero, the limit is automatically calculated based on -search.maxConcurrentRequests (inversely proportional) and memory available to the process (proportional).")
 	maxFederateSeries       = flag.Int("search.maxFederateSeries", 1e6, "The maximum number of time series, which can be returned from /federate. This option allows limiting memory usage")
 	maxExportSeries         = flag.Int("search.maxExportSeries", 10e6, "The maximum number of time series, which can be returned from /api/v1/export* APIs. This option allows limiting memory usage")
 	maxTSDBStatusSeries     = flag.Int("search.maxTSDBStatusSeries", 10e6, "The maximum number of time series, which can be processed during the call to /api/v1/status/tsdb. This option allows limiting memory usage")
@@ -103,11 +108,6 @@ func PrettifyQuery(w http.ResponseWriter, r *http.Request) {
 	_ = bw.Flush()
 }
 
-const (
-	federateEscapeSchemeUnderscore = "underscore"
-	federateEscapeSchemeUTF8       = "utf-8"
-)
-
 // FederateHandler implements /federate . See https://prometheus.io/docs/prometheus/latest/federation/
 func FederateHandler(startTime time.Time, w http.ResponseWriter, r *http.Request) error {
 	defer federateDuration.UpdateDuration(startTime)
@@ -132,21 +132,6 @@ func FederateHandler(startTime time.Time, w http.ResponseWriter, r *http.Request
 		return fmt.Errorf("cannot fetch data for %q: %w", sq, err)
 	}
 
-	// add best-effort format negotiation
-	// modern version of Prometheus always set allow-utf-8 in order to properly parse utf-8 names and labels
-	// prometheus below v3 uses underscore escaping by default and it's the most common standard
-	var escapeScheme string
-	accept := r.Header.Get("Accept")
-	if len(accept) > 0 && strings.Contains(accept, "allow-utf-8") {
-		escapeScheme = federateEscapeSchemeUTF8
-	}
-	// try fallback to legacy underscore escaping if needed for Prometheus only,
-	// it's not widely used after Prometheus v3.0 release
-	// most of the Prometheus scrapers already use allow-utf-8 header
-	isPrometheus := strings.HasPrefix(r.UserAgent(), "Prometheus")
-	if len(escapeScheme) == 0 && isPrometheus {
-		escapeScheme = federateEscapeSchemeUnderscore
-	}
 	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
 	bw := bufferedwriter.Get(w)
 	defer bufferedwriter.Put(bw)
@@ -156,7 +141,7 @@ func FederateHandler(startTime time.Time, w http.ResponseWriter, r *http.Request
 			return err
 		}
 		bb := sw.getBuffer(workerID)
-		WriteFederate(bb, rs, escapeScheme)
+		WriteFederate(bb, rs)
 		return sw.maybeFlushBuffer(bb)
 	})
 	if err == nil {
@@ -868,7 +853,7 @@ func QueryHandler(qt *querytracer.Tracer, startTime time.Time, w http.ResponseWr
 		End:                 start,
 		Step:                step,
 		MaxPointsPerSeries:  *maxPointsPerTimeseries,
-		MaxSeries:           0, // let vmstorage use maxUniqueTimeseries by default
+		MaxSeries:           GetMaxUniqueTimeSeries(),
 		QuotedRemoteAddr:    httpserver.GetQuotedRemoteAddr(r),
 		Deadline:            deadline,
 		MayCache:            mayCache,
@@ -979,7 +964,7 @@ func queryRangeHandler(qt *querytracer.Tracer, startTime time.Time, w http.Respo
 		End:                 end,
 		Step:                step,
 		MaxPointsPerSeries:  *maxPointsPerTimeseries,
-		MaxSeries:           0, // let vmstorage use maxUniqueTimeseries by default
+		MaxSeries:           GetMaxUniqueTimeSeries(),
 		QuotedRemoteAddr:    httpserver.GetQuotedRemoteAddr(r),
 		Deadline:            deadline,
 		MayCache:            mayCache,
@@ -1315,6 +1300,43 @@ func (sw *scalableWriter) flush() error {
 	return sw.bw.Flush()
 }
 
+var (
+	maxUniqueTimeseriesValueOnce sync.Once
+	maxUniqueTimeseriesValue     int
+)
+
+// InitMaxUniqueTimeseries init the max metrics limit calculated by available resources.
+// The calculation is split into calculateMaxUniqueTimeSeriesForResource for unit testing.
+func InitMaxUniqueTimeseries(maxConcurrentRequests int) {
+	maxUniqueTimeseriesValueOnce.Do(func() {
+		maxUniqueTimeseriesValue = *maxUniqueTimeseries
+		if maxUniqueTimeseriesValue <= 0 {
+			maxUniqueTimeseriesValue = calculateMaxUniqueTimeSeriesForResource(maxConcurrentRequests, memory.Remaining())
+		}
+	})
+}
+
+// calculateMaxUniqueTimeSeriesForResource calculate the max metrics limit calculated by available resources.
+func calculateMaxUniqueTimeSeriesForResource(maxConcurrentRequests, remainingMemory int) int {
+	if maxConcurrentRequests <= 0 {
+		// This line should NOT be reached unless the user has set an incorrect `search.maxConcurrentRequests`.
+		// In such cases, fallback to unlimited.
+		logger.Warnf("limiting -search.maxUniqueTimeseries to %v because -search.maxConcurrentRequests=%d.", 2e9, maxConcurrentRequests)
+		return 2e9
+	}
+
+	// Calculate the max metrics limit for a single request in the worst-case concurrent scenario.
+	// The approximate size of 1 unique series that could occupy in the vmstorage is 200 bytes.
+	mts := remainingMemory / 200 / maxConcurrentRequests
+	logger.Infof("limiting -search.maxUniqueTimeseries to %d according to -search.maxConcurrentRequests=%d and remaining memory=%d bytes. To increase the limit, reduce -search.maxConcurrentRequests or increase memory available to the process.", mts, maxConcurrentRequests, remainingMemory)
+	return mts
+}
+
+// GetMaxUniqueTimeSeries returns the max metrics limit calculated by available resources.
+func GetMaxUniqueTimeSeries() int {
+	return maxUniqueTimeseriesValue
+}
+
 // 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 {

app/vmselect/prometheus/prometheus_test.go

@@ -4,6 +4,7 @@ import (
 	"math"
 	"net/http"
 	"reflect"
+	"runtime"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/netstorage"
@@ -229,3 +230,29 @@ func TestGetLatencyOffsetMillisecondsFailure(t *testing.T) {
 	}
 	f("http://localhost?latency_offset=foobar")
 }
+
+func TestCalculateMaxMetricsLimitByResource(t *testing.T) {
+	f := func(maxConcurrentRequest, remainingMemory, expect int) {
+		t.Helper()
+		maxMetricsLimit := calculateMaxUniqueTimeSeriesForResource(maxConcurrentRequest, remainingMemory)
+		if maxMetricsLimit != expect {
+			t.Fatalf("unexpected max metrics limit: got %d, want %d", maxMetricsLimit, expect)
+		}
+	}
+
+	// Skip when GOARCH=386
+	if runtime.GOARCH != "386" {
+		// 8 CPU & 32 GiB
+		f(16, int(math.Round(32*1024*1024*1024*0.4)), 4294967)
+		// 4 CPU & 32 GiB
+		f(8, int(math.Round(32*1024*1024*1024*0.4)), 8589934)
+	}
+
+	// 2 CPU & 4 GiB
+	f(4, int(math.Round(4*1024*1024*1024*0.4)), 2147483)
+
+	// other edge cases
+	f(0, int(math.Round(4*1024*1024*1024*0.4)), 2e9)
+	f(4, 0, 0)
+
+}

app/vmselect/promql/rollup.go

@@ -2439,15 +2439,8 @@ func rollupIntegrate(rfa *rollupFuncArg) float64 {
 		prevTimestamp = timestamp
 		prevValue = v
 	}
-	// Only extrapolate the last value through to currTimestamp when the time
-	// series has any sample after the lookbehind window. When realNextValue is
-	// NaN the series has effectively ended at prevTimestamp, so accruing area
-	// past it would overcount the integral.
-	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9474
-	if !math.IsNaN(rfa.realNextValue) {
-		dt := float64(rfa.currTimestamp-prevTimestamp) / 1e3
-		sum += prevValue * dt
-	}
+	dt := float64(rfa.currTimestamp-prevTimestamp) / 1e3
+	sum += prevValue * dt
 	return sum
 }
 

app/vmselect/promql/rollup_test.go

@@ -1385,65 +1385,10 @@ func TestRollupFuncsNoWindow(t *testing.T) {
 		if samplesScanned != 24 {
 			t.Fatalf("expecting 24 samplesScanned from rollupConfig.Do; got %d", samplesScanned)
 		}
-		// At tEnd=160 the series has no samples past the window (last sample is at
-		// ts=130), so integrate() must not extrapolate prevValue through tEnd.
-		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9474
-		valuesExpected := []float64{nan, 2.148, 1.593, 1.156, 0.34}
+		valuesExpected := []float64{nan, 2.148, 1.593, 1.156, 1.36}
 		timestampsExpected := []int64{0, 40, 80, 120, 160}
 		testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected)
 	})
-	t.Run("integrate_past_series_end", func(t *testing.T) {
-		// Constant series of value 1.0 from t=0..3600s (1h) at 60s step.
-		// Query integrate(metric[1h]) across t=0..10800s with 600s step.
-		// For t=0..3600s the window overlap with the data is [0,t], so the integral grows from 0 to 3600 (seconds).
-		// After the series ends, integrate must NOT keep accruing 3600 — it
-		// should taper to 0 once the lookbehind window is entirely past the
-		// last sample.
-		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/9474
-		var testValues []int64
-		var testTimestamps []float64
-		for t := int64(0); t <= 3600_000; t += 60_000 {
-			testValues = append(testValues, t)
-			testTimestamps = append(testTimestamps, 1.0)
-		}
-		rc := rollupConfig{
-			Func:               rollupIntegrate,
-			Start:              0,
-			End:                10800_000,
-			Step:               600_000,
-			Window:             3600_000,
-			MaxPointsPerSeries: 1e4,
-		}
-		rc.Timestamps = rc.getTimestamps()
-		values, _ := rc.Do(nil, testTimestamps, testValues)
-		for i, ti := range rc.Timestamps {
-			v := values[i]
-
-			// For t<=3600s: window overlap is [0,ti], integral equals ti in seconds.
-			if ti <= 3600_000 {
-				expV := float64(ti / 1e3)
-				if v != expV {
-					t.Fatalf("unexpected integrate result at t=%ds, want=%.3f got=%.3f", ti/1e3, expV, v)
-				}
-				continue
-			}
-			// For 3600s<t<7200s: data is partially outside the window, so the
-			// integral shrinks linearly from 3600 to 0 as t approaches 7200s.
-			if ti > 3600_000 && ti < 7200_000 {
-				expV := float64((7200_000 - ti) / 1e3)
-				if v != expV {
-					t.Fatalf("unexpected integrate result at t=%ds, want=%.3f got=%.3f", ti/1e3, expV, v)
-				}
-				continue
-			}
-			if ti >= 7200_000 {
-				// Window entirely past data end: must be NaN.
-				if !math.IsNaN(v) {
-					t.Fatalf("unexpected integrate result at t=%ds, want=NaN got=%.3f", ti/1e3, v)
-				}
-			}
-		}
-	})
 	t.Run("distinct_over_time_1", func(t *testing.T) {
 		rc := rollupConfig{
 			Func:               rollupDistinct,

app/vmselect/searchutil/searchutil.go

@@ -132,20 +132,9 @@ func (d *Deadline) String() string {
 //
 //	{env="prod",team="devops",t1="v1",t2="v2"}
 //	{env=~"dev|staging",team!="devops",t1="v1",t2="v2"}
-//
-//	Query args from URL path have precedence over post form args.
 func GetExtraTagFilters(r *http.Request) ([][]storage.TagFilter, error) {
 	var tagFilters []storage.TagFilter
-	urlQueryValues := r.URL.Query()
-	getRequestParam := func(key string) []string {
-		// query request param must always take precedence over form values
-		// in order to simplify security enforcement policy for extra_label and extra_filters
-		if uv, ok := urlQueryValues[key]; ok {
-			return uv
-		}
-		return r.Form[key]
-	}
-	for _, match := range getRequestParam("extra_label") {
+	for _, match := range r.Form["extra_label"] {
 		tmp := strings.SplitN(match, "=", 2)
 		if len(tmp) != 2 {
 			return nil, fmt.Errorf("`extra_label` query arg must have the format `name=value`; got %q", match)
@@ -159,8 +148,8 @@ func GetExtraTagFilters(r *http.Request) ([][]storage.TagFilter, error) {
 			Value: []byte(tmp[1]),
 		})
 	}
-	extraFilters := append([]string{}, getRequestParam("extra_filters")...)
-	extraFilters = append(extraFilters, getRequestParam("extra_filters[]")...)
+	extraFilters := append([]string{}, r.Form["extra_filters"]...)
+	extraFilters = append(extraFilters, r.Form["extra_filters[]"]...)
 	if len(extraFilters) == 0 {
 		if len(tagFilters) == 0 {
 			return nil, nil

app/vmselect/searchutil/searchutil_test.go

@@ -20,7 +20,6 @@ func TestGetExtraTagFilters(t *testing.T) {
 		}
 		return &http.Request{
 			Form: q,
-			URL:  &url.URL{RawQuery: q.Encode()},
 		}
 	}
 	f := func(t *testing.T, r *http.Request, want []string, wantErr bool) {
@@ -80,24 +79,6 @@ func TestGetExtraTagFilters(t *testing.T) {
 		nil,
 		false,
 	)
-
-	formValues, err := url.ParseQuery(`extra_label=env=prod&extra_label=job=vmsingle&extra_label=tenant=prod&extra_filters[]={foo="bar"}&extra_filters[]={tenant="prod"}`)
-	if err != nil {
-		t.Fatalf("BUG: cannot parse query: %s", err)
-	}
-	urlValues, err := url.ParseQuery(`extra_label=job=vmagent&extra_label=env=dev&extra_filters[]={tenant="dev"}`)
-	if err != nil {
-		t.Fatalf("BUG: cannot parse query: %s", err)
-	}
-	httpReqWithBothFormAndURLParams := &http.Request{
-		Form: formValues,
-		URL: &url.URL{
-			RawQuery: urlValues.Encode(),
-		},
-	}
-	f(t, httpReqWithBothFormAndURLParams,
-		[]string{`{tenant="dev",job="vmagent",env="dev"}`},
-		false)
 }
 
 func TestParseMetricSelectorSuccess(t *testing.T) {